Functory
functory.
All Articles
7 min read
Functory

Compute throughput from Linear project boards in Python and expose as a Zapier API for sprint retrospectives

This article shows a practical, deployable pattern: a single-file Python function that computes pull request throughput and basic review metrics from Linear project board exports (or the Linear GraphQL API) and is packaged as an API you can call from Zapier, Make, or n8n. It is aimed at analytics leads and consultants who need reliable per-sprint metrics (merged PR count, average time-to-merge, reviewers per PR) for sprint retrospectives, without hand-editing CSVs or running ad-hoc notebooks.

We cover exact inputs and outputs, a tested pandas-based implementation you can run locally or publish on Functory as a no-ops cloud function, and a short Zapier integration workflow so reports can be scheduled and emailed automatically.

What this function expects and produces

Inputs (two supported modes):

  • CSV mode: a UTF-8 CSV file exported from Linear or a CI system. Required columns: pr_id (string), title (string), author (string), created_at (ISO8601 datetime), merged_at (ISO8601 datetime or empty), reviewers (semicolon-separated emails), linear_cycle (sprint name), state (open/merged/closed).
  • API mode: a Linear GraphQL token (string) and a project id; the function will query issues/PR-like references and normalize the same columns as above. (Example GraphQL query is provided in the notes.)

Transformations performed:

  • Normalize timestamps to UTC and compute duration metrics: time_to_merge = merged_at - created_at (hours), review_delay = first_reviewed_at - created_at if available.
  • Assign each PR to a sprint (linear_cycle) based on a provided sprint mapping or the linear_cycle column.
  • Aggregate per-sprint by developer and overall: merged_count, median_time_to_merge (hours), mean_reviewers_per_pr.
  • Optionally produce a CSV summary and JSON-ready dictionary of metrics for API consumers.

Outputs:

  • summary.json: mapping of sprint -> {merged_count, median_time_to_merge_hours, avg_reviewers}
  • summary.csv: flattened table with columns sprint, developer, merged_count, median_time_to_merge_hours, avg_reviewers
  • Return value (when used as a Functory function): path to summary.csv so the platform exposes it as a downloadable result, and print() logs for debugging.

Real-world scenario

Analytics lead: Dana at a 12-person product company runs 2-week sprints. For the March 4–17 sprint she needs:

  • Total merged PRs by developer for the sprint (to discuss load and bottlenecks).
  • Median time-to-merge to catch slow review cycles (flag PRs with >48h).
  • Average number of reviewers per PR to evaluate code review coverage.

Concrete input (CSV snippet):

pr_id,title,author,created_at,merged_at,reviewers,linear_cycle,state
PR-531,Fix login crash,alice,2025-03-05T09:12:34Z,2025-03-06T14:02:10Z,bob;carol,Sprint-2025-03-04-03,merged
PR-540,Add metrics endpoint,bob,2025-03-07T11:00:00Z,,dave,Sprint-2025-03-04-03,open

Expected aggregated output (JSON excerpt):

{
  "Sprint-2025-03-04-03": {
    "merged_count": 42,
    "median_time_to_merge_hours": 18.2,
    "avg_reviewers_per_pr": 1.7
  }
}

Example dataset and the problem it solves

Example dataset: 1,200 rows exported from Linear covering 3 months (six 2-week sprints). Columns match the CSV schema above. Problem: the analytics lead currently copies rows into Google Sheets, filters by sprint, and writes manual pivot tables. That takes ~3 hours per retrospective and is error-prone.

This function automates the ETL step: ingest CSV → normalize datetimes and reviewers → compute per-sprint throughput and review metrics → output CSV/JSON that drives the retrospective slide deck and Slack summary.

Step-by-step workflow (how a developer uses it)

  1. Export a CSV from Linear or allow the function to fetch data via the Linear API token.
  2. Call the Python function locally or upload it to Functory with Python 3.11.11 and pinned deps.
  3. Provide inputs: path/to/linear_export.csv, sprint_start, sprint_end (or project id + token).
  4. Function returns path to summary.csv and prints debug logs; download summary.csv.
  5. Plug summary.csv into the Zapier/Make workflow to email a report, create a retrospective Trello card, or post a Slack summary automatically.

Processing algorithm (high-level)

  1. Load CSV or call Linear GraphQL; select PR-like records and ensure columns: pr_id, author, created_at, merged_at, reviewers, linear_cycle.
  2. Parse datetimes to UTC; compute time_to_merge (hours) = (merged_at - created_at).total_seconds()/3600; compute reviewers_count = len(reviewers.split(';')).
  3. Filter to target sprint(s) and only merged PRs for throughput; groupby sprint and author to compute merged_count and median time_to_merge; also compute avg_reviewers per sprint.
  4. Write summary.csv and return JSON summary for the API consumer.

Minimal working Python example

This single-file example reads a CSV, computes throughput, writes a CSV, and returns the path. It is intentionally small and ready to be wrapped into Functory's main(...) entrypoint.

from typing import Optional
import pandas as pd
from pathlib import Path
from datetime import datetime


def compute_throughput(csv_path: str, sprint_name: Optional[str] = None) -> str:
    df = pd.read_csv(csv_path, parse_dates=["created_at", "merged_at"], keep_default_na=False)

    # normalize reviewers and compute counts
    df["reviewers_count"] = df["reviewers"].replace("", "").apply(lambda s: len([r for r in s.split(";") if r]))

    # compute time to merge in hours; ignore rows without merged_at
    df["time_to_merge_hours"] = (df["merged_at"] - df["created_at"]).dt.total_seconds() / 3600

    # filter merged PRs and optional sprint
    merged = df[df["state"].str.lower() == "merged"].copy()
    if sprint_name:
        merged = merged[merged["linear_cycle"] == sprint_name]

    # aggregate per sprint and overall
    agg = merged.groupby("linear_cycle").agg(
        merged_count=("pr_id", "nunique"),
        median_time_to_merge_hours=("time_to_merge_hours", lambda x: float(x.median() if not x.isna().all() else 0)),
        avg_reviewers_per_pr=("reviewers_count", "mean"),
    ).reset_index()

    out_path = Path(csv_path).with_name("throughput_summary.csv")
    agg.to_csv(out_path, index=False)
    print(f"Wrote throughput summary to {out_path}")
    return str(out_path)


# quick run example
if __name__ == "__main__":
    print(compute_throughput("data/linear_export.csv", sprint_name="Sprint-2025-03-04-03"))

How Functory Makes It Easy

On Functory you would wrap the core logic above in a single Python entrypoint named main(...). Functory expects that main(...) accepts simple types (str, int, float, FilePath). For this example main(csv_path: str, sprint_name: Optional[str] = None) -> str returns a path-like string (the platform will expose the produced throughput_summary.csv as a downloadable result).

Concrete developer steps on Functory:

  • Choose an exact Python patch version such as 3.11.11 for reproducibility.
  • Create requirements.txt with pinned dependencies, e.g. pandas==2.1.0, python-dateutil==2.8.2, requests==2.31.0 (one per line).
  • Commit a single file that defines main(csv_path: str, sprint_name: Optional[str] = None) and implements the logic above; no CLI wrapper is needed—Functory calls main directly.
  • The UI/API automatically maps parameters to fields; uploaded CSVs are passed as FilePath and the returned path becomes a downloadable file in the UI or the API response.

Benefits on Functory: zero server ops, autoscaling for concurrent report requests, built-in print() logs for debugging, option to run on CPU/GPU tiers when heavier processing is added, and pay-per-use billing handled by the platform. You can chain this function in a Functory workflow: pre-processing function → throughput function → reporting function that sends Slack/Email, enabling a fully automated retrospective pipeline.

Alternatives and why this function-based approach is better

Common alternatives: hand-edited spreadsheets (Google Sheets pivot tables), ad-hoc Jupyter notebooks, or Linear's built-in limited reports. Spreadsheets are error-prone and not scriptable; notebooks are flexible but not easily shareable as an API for Zapier/Make; Linear reports can be too coarse (no reviewer counts or median time-to-merge). A single-file function wrapped as an API provides reproducible, testable logic, easy integration with automation platforms (Zapier, Make, n8n), and versioned code you can pin and publish.

Business impact

Automating this ETL reduces manual retrospective preparation time from ~3 hours to under 10 minutes per sprint for a typical analytics lead, a ~85% reduction in manual processing. For a consultancy billing at $120/hour, that saves ~$330 per sprint in billable time or frees analyst capacity for higher-value work.

Industry note: According to a 2024 DevOps Pulse report by GitLab, teams that automated reporting and dashboards reduced context-switching time by 42% (source: GitLab DevOps Pulse 2024).

When to use this

  • You need repeatable per-sprint metrics for retrospectives and want an API triggerable from Zapier/Make/n8n.
  • You have structured exports from Linear (CSV) or an API token and want consistent, auditable aggregation logic.
  • You want to avoid manual spreadsheets and expose results to automation workflows or non-technical stakeholders.

Practical comparison

Notebooks are great for exploration but poor as production endpoints; spreadsheets are simple but break for team-scale automation. Publishing the logic as a single-function API gives the reproducibility of code with the accessibility of an integration trigger. It is also easier to unit test and to pin dependency versions to avoid 'it worked yesterday' issues.

Conclusion

Automating throughput computation from Linear project boards converts messy exports into repeatable, auditable sprint metrics you can call from Zapier, Make, or n8n. The pattern is lightweight—single-file Python plus pinned deps—and yields measurable time savings for analytics leads. Next steps: implement unit tests for the aggregation logic, add a small webhook-triggered Functory function that runs after each sprint close, and publish the function to your team or the Functory marketplace so other teams can benefit.

Thanks for reading.