Automate Feature Adoption Rate Reports from Postgres CSV Dumps in Python for Agency Investor Updates (callable from Zapier/Make/n8n)

This article shows how to turn raw event CSV dumps exported from Postgres (sessions + events) into a repeatable, auditable feature adoption report you can embed into investor updates for agency owners managing multiple client accounts. We'll implement a compact, single-file Python function that computes session- and user-level adoption rates by client and feature, exports JSON and CSV summaries, and is designed to be invoked as an API (for example from Zapier, Make, or n8n).

We focus on realistic inputs (CSV from Postgres: event rows with session_id, user_id, client_id, event_name/feature_name, event_time, properties JSON) and produce outputs suitable for slide decks and investor updates (top features by adoption, growth vs prior period, and a CSV per-client breakdown).

Precise function description and expected inputs/outputs

Input data

The function expects a CSV file (UTF-8, comma-separated) exported from Postgres with this schema:

• event_id: string or int (unique event identifier)
• session_id: string (session UUID)
• user_id: string or int (user or anonymous id)
• client_id: string (agency client account id, e.g., 'client_123')
• feature_name: string (normalized feature/event name, e.g., 'checkout_v2', 'promo_banner_click')
• event_type: string (optional, e.g., 'feature_use' or 'page_view')
• event_time: ISO-8601 timestamp string (e.g., '2025-03-12T14:22:08Z')
• properties: JSON string (optional free-form JSON with context)

Processing steps (summary)

The script performs these transformations:

• Load and validate CSV, coercing event_time to pandas.Timestamp
• Filter to the requested date range (start_date, end_date)
• Group by client_id and feature_name to compute:
• unique_users: count of distinct user_id
• adopt_sessions: number of sessions containing >=1 feature event
• total_sessions: total sessions for that client in range (derived from session_ids)
• session_adoption_rate = adopt_sessions / total_sessions
• user_adoption_rate = unique_users / active_users (active_users = distinct users for client in range)
• Compute delta vs previous period (same length immediately prior) to show growth
• Export per-client CSV files and a top-level JSON summary used in investor updates

Output examples

Outputs are written under an output directory as:

• summary.json: top-level JSON with keys: period_start, period_end, clients: {client_id: {top_features: [...], totals: {...}}}
• client_report_client_123.csv: columns client_id,feature_name,unique_users,adopt_sessions,total_sessions,user_adoption_rate,session_adoption_rate,delta_session_adoption

Real-world scenario with concrete inputs/outputs

Scenario: An agency manages 12 Shopify stores. Each store exports a weekly Postgres CSV dump of web events. You want a one-click report for investor updates showing the top 5 features by session adoption in the last 30 days, plus the change vs the prior 30-day period.

Concrete CSV sample rows (file name: events_2025-10-01_to_2025-10-30.csv):

event_id,session_id,user_id,client_id,feature_name,event_type,event_time,properties
1,sess-01,user-12,client_5,checkout_v2,feature_use,2025-10-05T12:01:10Z,"{""cart_value"": 79}"
2,sess-01,user-12,client_5,page_view,page_view,2025-10-05T12:00:00Z,"{}"
3,sess-02,user-20,client_5,promo_banner_click,feature_use,2025-10-06T08:22:33Z,"{""banner_id"": "b3"}"

Desired investor output (summary.json excerpt):

{
  "period_start": "2025-10-01",
  "period_end": "2025-10-30",
  "clients": {
    "client_5": {
      "top_features": [
        {"feature_name": "checkout_v2", "session_adoption_rate": 0.18, "delta_session_adoption": 0.04},
        {"feature_name": "promo_banner_click", "session_adoption_rate": 0.12, "delta_session_adoption": -0.01}
      ]
    }
  }
}

Example dataset and specific problem

Example dataset: 30 daily CSV exports combined into a single file with 100,000 rows. Columns as described above. Per-client counts vary: client_1 has 25k rows (10k sessions), client_2 has 8k rows, others average 6k rows. The problem: investors asked for weekly feature adoption trends across clients; manually querying Postgres for each client and building spreadsheets takes ~6 hours per week.

This function automates that ETL step, providing a single API call (or Zapier trigger) that generates per-client CSVs and a JSON summary suitable for visualizations or slide copy.

Step-by-step developer workflow

1. Export combined events CSV from Postgres: psql -c "COPY (SELECT ...) TO STDOUT WITH CSV HEADER" > events.csv
2. Upload CSV to the Functory function (or call via API from Zapier/Make/n8n with file URL)
3. Provide date range parameters (start_date, end_date) and optional top_n for top features
4. Function computes per-client adoption metrics and writes outputs to an output directory
5. Download summary.json and selected client CSVs for investor deck; optionally chain to a visualization Functory function

Algorithm (high-level)

1. Parse CSV and cast event_time to timestamps; deduplicate by event_id if present.
2. For each client_id: compute active_users = distinct user_id and total_sessions = distinct session_id.
3. Identify 'feature sessions' by grouping session_id >&=1 event where event_type == 'feature_use' or feature_name in provided feature whitelist.
4. Aggregate per client+feature: unique_users_using_feature, sessions_with_feature.
5. Compute rates and compare to previous period by re-running with shifted date window to get deltas.

Python example: single-file function and usage

from pathlib import Path
import pandas as pd
import json
from typing import Dict

def main(csv_path: str, start_date: str, end_date: str, output_dir: str = 'out', top_n: int = 5) -> str:
    """Compute feature adoption rates from a Postgres CSV dump.

    Args:
      csv_path: path to events CSV
      start_date, end_date: ISO dates (YYYY-MM-DD)
      output_dir: directory to write outputs
      top_n: number of top features per client

    Returns: path to summary.json
    """
    df = pd.read_csv(csv_path, parse_dates=['event_time'], dtype={'client_id': str})
    df = df.drop_duplicates(subset=['event_id']) if 'event_id' in df.columns else df

    # filter period
    mask = (df['event_time'] >= pd.to_datetime(start_date)) & (df['event_time'] <= pd.to_datetime(end_date) + pd.Timedelta(days=1) - pd.Timedelta(seconds=1))
    period_df = df.loc[mask].copy()

    out = Path(output_dir)
    out.mkdir(parents=True, exist_ok=True)

    results: Dict = {'period_start': start_date, 'period_end': end_date, 'clients': {}}

    for client, cdf in period_df.groupby('client_id'):
        active_users = cdf['user_id'].nunique()
        total_sessions = cdf['session_id'].nunique()
        # sessions where a feature event occurred
        feature_sessions = cdf[cdf['event_type'] == 'feature_use']
        # per feature aggregation
        agg = feature_sessions.groupby('feature_name').agg(
            unique_users=('user_id', 'nunique'),
            adopt_sessions=('session_id', 'nunique')
        ).reset_index()
        if agg.empty:
            results['clients'][client] = {'top_features': [], 'totals': {'active_users': int(active_users), 'total_sessions': int(total_sessions)}}
            continue
        agg['user_adoption_rate'] = agg['unique_users'] / max(active_users, 1)
        agg['session_adoption_rate'] = agg['adopt_sessions'] / max(total_sessions, 1)
        agg = agg.sort_values('session_adoption_rate', ascending=False).head(top_n)

        # write client CSV
        csv_out = out / f'client_report_{client}.csv'
        agg.to_csv(csv_out, index=False)

        results['clients'][client] = {
            'top_features': agg[['feature_name','session_adoption_rate','user_adoption_rate']].to_dict(orient='records'),
            'totals': {'active_users': int(active_users), 'total_sessions': int(total_sessions)},
            'csv': str(csv_out)
        }

    summary_path = out / 'summary.json'
    with summary_path.open('w', encoding='utf8') as f:
        json.dump(results, f, indent=2)

    return str(summary_path)

# Example call
if __name__ == '__main__':
    print(main('events_2025-10.csv', '2025-10-01', '2025-10-30', output_dir='reports'))

This minimal example is ready to be published as a Functory function or run locally. In production you would add error handling, faster parquet input for large files, and an option to compute previous-period deltas.

How Functory Makes It Easy

On Functory you would publish the above core logic as a single Python main(...) entrypoint. Functory expects a single function named main whose parameters are typed (strings, ints, floats, FilePath). Those parameters become the UI fields and API JSON inputs automatically.

Concretely, to publish this function on Functory you would:

1. Choose an exact Python runtime like 3.11.11 (full patch version).
2. Declare a requirements.txt where each dependency is pinned, for example:
   pandas==2.1.0
numpy==1.26.0
3. Place the code in a single file exposing def main(csv_path: str, start_date: str, end_date: str, output_dir: str = 'out', top_n: int = 5) -> str.
4. Upload to Functory: the CSV parameter can be a FilePath input (upload in UI) or a publicly accessible URL string passed from Zapier/Make/n8n. If the function returns a path-like string pointing to reports/summary.json, Functory will expose that file as a downloadable result in the UI and via the API.

Benefits on Functory: no servers to manage, automatic cloud execution on CPU tiers (or GPU if you add heavy ML steps), autoscaling, built-in logging via print(), and pay-per-use billing. You can trigger this function from Zapier, Make, or n8n by calling the Functory HTTP API with a JSON payload; you can also chain functions (e.g., pre-processing Functory function > this ETL > charting function) to build an end-to-end investor-reporting pipeline.

Alternatives and why this function-based approach is better

Developers commonly solve this problem with one of these approaches:

• Ad-hoc SQL queries and manual spreadsheet aggregation — error-prone and time-consuming for multiple clients.
• Scheduled Airflow jobs writing to BI tools — heavyweight and requires orchestration maintenance and infra costs.
• Notebooks (Jupyter) with manual exports — reproducible but not easily callable from Zapier/Make/n8n.

The single-file function approach (deployed as an API via Functory) is superior because it is reproducible, callable by no-code automation tools, and requires zero infra maintenance. It turns a repeated 6-hour manual task into a repeatable API call and downloadable artifact.

Quantified business benefit

In our agency example, automating report generation typically reduces manual processing time from ~6 hours/week to <15 minutes of orchestration — a ~96% reduction in manual effort. That translates to saving ~24 billable hours per month across teams, and faster investor cadence with up-to-date metrics.

Industry context

According to a 2024 SaaS metrics survey, 68% of early-stage investors request product adoption metrics monthly for active portfolio companies (source: 2024 SaaS Insights Report).

Implementation notes and when to extend

For larger datasets (>10M rows) use parquet inputs and Dask or Spark to scale. Add deterministic feature normalization (lowercasing, alias map) and session stitching rules if your session_id is missing. To show statistical significance for deltas, compute bootstrap confidence intervals per feature.