Blueprint — Reports & Exports

Status: partial (DISA-side /api/reports, /api/export/db/excel live) · Slug: reports · Kind: Cross-cutting · Module #17 FOUNDATIONAL. Every other module produces outputs through this service. Designed and migrated alongside Comms in the second wave.

1. Module summary

Reports & Exports is the unified report generation service of Aayojana — one registry of named reports owned by their respective modules, one rendering engine that emits PDF (WeasyPrint), Excel (openpyxl), CSV, and government-portal-ready JSON from a single source-of-truth definition. It owns the scheduled-report runner that fires monthly trustee packs and annual 80G mailings via Comms, the on-demand report-runner for ad-hoc queries, and the large-export job system that handles multi-GB DB exports asynchronously. Multi-tenant scoping is enforced at the report-query layer (no report can read across tenants), and every run is recorded in report_runs for audit. The existing DISA-side routers (reports.py, export.py) become the first registered report definitions and migrate behind the new registry without breaking external callers.

2. Data model

All tables live in aayojana.reports.models. Tenant-scoped via TenantMixin. Report definitions are seeded from disk (Python files registering ReportDefinition specs) and synced to the DB on app startup — DB rows are the runtime catalog, disk modules are the source of truth.

# src/aayojana/reports/models.py
from datetime import datetime
from sqlalchemy import (
    JSON, Boolean, DateTime, ForeignKey, Integer, String, Text,
    UniqueConstraint, Index, func,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship

from aayojana.models.base import AuditMixin, Base, TenantMixin, uuid4_str


# ============ DEFINITIONS (the catalog) ============

class ReportDefinition(Base, AuditMixin):
    """A named, parameterised report. NOT tenant-scoped — definitions are
    platform-wide; every tenant gets the same catalog. Tenant scoping is
    applied at run() time via parameters.

    Definitions are seeded from `aayojana.reports.definitions.*` Python modules
    that register a ReportSpec. The DB row is the runtime catalog used by UI;
    the function reference is resolved by `function_ref` (dotted path)."""
    __tablename__ = "report_definitions"
    __table_args__ = (
        UniqueConstraint("name", name="uq_report_definition_name"),
    )

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=uuid4_str)
    name: Mapped[str] = mapped_column(String(120), nullable=False)
    title: Mapped[str] = mapped_column(String(200), nullable=False)
    description: Mapped[str | None] = mapped_column(Text, nullable=True)
    owner_module: Mapped[str] = mapped_column(String(32), nullable=False)
    category: Mapped[str] = mapped_column(String(40), nullable=False)
    # 'financial' | 'operational' | 'statutory' | 'audit' | 'communications' | 'people'

    function_ref: Mapped[str] = mapped_column(String(160), nullable=False)
    sql_template: Mapped[str | None] = mapped_column(Text, nullable=True)

    parameter_schema: Mapped[dict] = mapped_column(JSON, nullable=False, default=dict)
    output_formats: Mapped[list] = mapped_column(JSON, nullable=False)
    default_format: Mapped[str] = mapped_column(String(8), nullable=False, default="pdf")
    template_path: Mapped[str | None] = mapped_column(String(200), nullable=True)

    default_audience_segment: Mapped[str | None] = mapped_column(String(120), nullable=True)
    requires_role: Mapped[str | None] = mapped_column(String(40), nullable=True)
    legal_weight: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
    enabled: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)


# ============ RUNS (history of executions) ============

class ReportRun(Base, TenantMixin, AuditMixin):
    """One row per report execution. Output blob stored externally (GCS) when
    > 1 MB; embedded in `output_inline` when small."""
    __tablename__ = "report_runs"
    __table_args__ = (
        Index("ix_report_runs_tenant_definition", "tenant_id", "definition_id"),
        Index("ix_report_runs_status", "status"),
        Index("ix_report_runs_produced", "produced_at"),
    )

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=uuid4_str)
    definition_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("report_definitions.id"), nullable=False
    )
    parameters: Mapped[dict] = mapped_column(JSON, nullable=False, default=dict)
    output_format: Mapped[str] = mapped_column(String(8), nullable=False)

    status: Mapped[str] = mapped_column(String(20), nullable=False, default="queued")
    # 'queued' | 'running' | 'completed' | 'failed' | 'expired'
    started_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
    produced_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
    duration_ms: Mapped[int | None] = mapped_column(Integer, nullable=True)

    output_url: Mapped[str | None] = mapped_column(String(500), nullable=True)
    output_inline: Mapped[bytes | None] = mapped_column(nullable=True)
    output_bytes: Mapped[int | None] = mapped_column(Integer, nullable=True)
    output_sha256: Mapped[str | None] = mapped_column(String(64), nullable=True)

    requested_by_user_id: Mapped[int | None] = mapped_column(
        Integer, ForeignKey("users.id"), nullable=True
    )
    scheduled_report_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("scheduled_reports.id"), nullable=True
    )
    error: Mapped[str | None] = mapped_column(Text, nullable=True)
    row_count: Mapped[int | None] = mapped_column(Integer, nullable=True)


# ============ SCHEDULES (recurring runs) ============

class ScheduledReport(Base, TenantMixin, AuditMixin):
    """A cron-driven recurring report. On fire, creates a ReportRun and (if
    `recipients_segment` is set) hands the output URL to Comms for delivery."""
    __tablename__ = "scheduled_reports"
    __table_args__ = (
        Index("ix_scheduled_next_fire", "next_fire_at"),
    )

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=uuid4_str)
    name: Mapped[str] = mapped_column(String(160), nullable=False)
    definition_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("report_definitions.id"), nullable=False
    )
    parameters: Mapped[dict] = mapped_column(JSON, nullable=False, default=dict)
    output_format: Mapped[str] = mapped_column(String(8), nullable=False, default="pdf")
    cron_expr: Mapped[str] = mapped_column(String(60), nullable=False)
    timezone: Mapped[str] = mapped_column(String(40), nullable=False, default="Asia/Kolkata")

    recipients_segment: Mapped[str | None] = mapped_column(String(120), nullable=True)
    delivery_template: Mapped[str | None] = mapped_column(String(120), nullable=True)

    enabled: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
    last_fired_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
    next_fire_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
    failures_in_a_row: Mapped[int] = mapped_column(Integer, default=0, nullable=False)


# ============ EXPORT JOBS (large async exports) ============

class ExportJob(Base, TenantMixin, AuditMixin):
    """For multi-GB exports — full DB dump, full transactions table, audit log
    archive."""
    __tablename__ = "export_jobs"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=uuid4_str)
    name: Mapped[str] = mapped_column(String(160), nullable=False)
    kind: Mapped[str] = mapped_column(String(40), nullable=False)
    # 'db_full_excel' | 'transactions_csv' | 'members_csv' | 'audit_archive' |
    # 'fcra_disclosure_pack' | 'annual_handover_zip'
    parameters: Mapped[dict] = mapped_column(JSON, nullable=False, default=dict)

    status: Mapped[str] = mapped_column(String(20), nullable=False, default="queued")
    started_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
    completed_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
    output_url: Mapped[str | None] = mapped_column(String(500), nullable=True)
    output_bytes: Mapped[int | None] = mapped_column(Integer, nullable=True)
    output_expires_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)

    requested_by_user_id: Mapped[int | None] = mapped_column(
        Integer, ForeignKey("users.id"), nullable=True
    )
    error: Mapped[str | None] = mapped_column(Text, nullable=True)

3. Reuse map

4. API surface

5. Service layer — the public API

# src/aayojana/reports/service.py
from typing import Literal
from datetime import datetime
from sqlalchemy.ext.asyncio import AsyncSession

OutputFormat = Literal["pdf", "xlsx", "csv", "json"]


async def run_report(
    db: AsyncSession, *,
    tenant_id: str, name: str,
    parameters: dict | None = None,
    output_format: OutputFormat | None = None,
    user_id: int | None = None,
) -> "ReportRun":
    """Resolves definition by name → validates parameters against schema →
    invokes function_ref → renders to requested format → uploads to GCS or
    inlines → returns ReportRun. Records audit event if definition.legal_weight."""


async def schedule_report(
    db: AsyncSession, *,
    tenant_id: str, definition_name: str,
    cron_expr: str,
    parameters: dict | None = None,
    output_format: OutputFormat = "pdf",
    recipients_segment: str | None = None,
    delivery_template: str | None = None,
    name: str,
) -> "ScheduledReport":
    """Creates a ScheduledReport. Computes next_fire_at. The reports worker
    picks it up at fire time, runs the report, and (if recipients_segment is
    set) calls comms.send_to_segment with the run's output URL embedded."""


async def render_pdf(
    template_path: str, context: dict, *, options: dict | None = None,
) -> bytes:
    """WeasyPrint-based HTML→PDF renderer. Used internally by run_report,
    exposed for Comms postal channel and Newsletter print issues."""


async def render_excel(
    rows: list[dict] | list[list], columns: list[str], *,
    sheet_name: str = "Sheet1", multi_sheet: dict[str, tuple] | None = None,
) -> bytes: ...


async def render_csv(rows: list[dict], columns: list[str]) -> bytes: ...


async def render_json_govt(
    spec_name: str, data: dict, *, validate: bool = True,
) -> bytes:
    """Government-portal-ready JSON. spec_name selects the schema (e.g.
    'fcra_fc4', 'gstr1', 'tds_24q'). Validates against bundled JSON schemas."""


async def register_report(
    *,
    name: str, title: str, owner_module: str, category: str,
    function_ref: str,
    parameter_schema: dict,
    output_formats: list[OutputFormat],
    default_format: OutputFormat = "pdf",
    template_path: str | None = None,
    default_audience_segment: str | None = None,
    requires_role: str | None = None,
    legal_weight: bool = False,
) -> None:
    """Idempotent. Called by each module's `register_definitions()` at app
    startup. Upserts ReportDefinition row by name."""


async def enqueue_export(
    db: AsyncSession, *,
    tenant_id: str, kind: str,
    parameters: dict | None = None, user_id: int | None = None,
) -> "ExportJob": ...

Catalog of pre-registered reports

6. UI / Templates

src/aayojana/templates/reports/. PDF templates are HTML+CSS for WeasyPrint (page-break-aware, print-stylesheet driven).

Per-report PDF templates: templates/reports/<owner_module>/<name>.html. Layout fragments shared via Jinja {% include %}. Excel rendering via openpyxl with header styling, freeze panes, autofilter, multi-sheet support.

7. Migration plan

Run AFTER comms_jobs_log (0030).

8. Cross-module dependencies

Reports is consumed by: every other module. Each registers definitions:

# Vitta
def register_definitions():
    reports.register_report(
        name="vitta_trial_balance", title="Trial Balance",
        owner_module="vitta", category="financial",
        function_ref="aayojana.vitta.reports.trial_balance:run",
        parameter_schema={
            "type": "object", "required": ["period_end"],
            "properties": {
                "period_end": {"type":"string","format":"date"},
                "fund": {"type":"string","enum":["all","general","annadana","building","fcra","endowment"]},
                "branch_id": {"type":"string"},
            },
        },
        output_formats=["pdf","xlsx"],
        template_path="reports/vitta/trial_balance.html",
        requires_role="module-admin:vitta",
    )

Reports invokes Comms for scheduled-report delivery:

run = await run_report(db, tenant_id=ten, name=sched.definition.name,
                       parameters=sched.parameters, output_format=sched.output_format)
if sched.recipients_segment and sched.delivery_template:
    await comms.send_to_segment(
        db, tenant_id=ten, segment_name=sched.recipients_segment,
        template_name=sched.delivery_template, channels=["email"],
        vars={"run": run, "output_url": run.output_url},
    )

Reports invokes Audit for legal-weight reports — record_audit_event with action='export' after each successful run.

9. Implementation phases

Phase A — Manual run + PDF/Excel (2 weeks): 1. Land migration 0031. 2. Implement run_report, render_pdf (WeasyPrint), render_excel, render_csv. 3. Build registry + register_report + startup hook. 4. Migrate DISA reports to the registry; keep old URLs as shims. 5. Build /admin/reports catalog + run-form + run-viewer. 6. Wire WeasyPrint with VijayaDV font for Sanskrit accent rendering.

Phase B — Scheduled + statutory (3 weeks): 7. APScheduler-based reports worker (next_fire_at scan). 8. Comms delivery integration. 9. Statutory generators: FC-4 JSON, GSTR-1 JSON, TDS 24Q JSON, Form 10B PDF. 10. ExportJob worker with GCS upload + signed URLs. 11. Schedule manager UI.

Phase C — Trustee pack + analytics (2 weeks): 12. Trustee Pack composite (Vitta + Compliance + Operational + Audit). 13. Annual Handover ZIP — auditor-ready bundle as ExportJob. 14. Comms deliverability dashboard. 15. Cross-module KPIs dashboard.

10. Open questions

1. PDF rendering — WeasyPrint vs browser-print? Rec: WeasyPrint for archived/legal-weight; browser-print only for ad-hoc admin views.
2. Output blob storage — bucket per tenant, or one bucket prefixed? Rec: aayojana-reports-prod with tenants/{tenant_id}/runs/{run_id}.{ext}, signed URLs 7-day TTL.
3. Output retention — Rec: legal_weight 8 years; ops 90 days; audit archive indefinite.
4. Parameter validation — Rec: Strict for legal_weight, lenient for ops.
5. Cross-tenant reports — Rec: Yes for requires_role='platform-admin'.
6. Async worker — Rec: Same APScheduler as Comms, separate queue tables.
7. Statutory JSON schemas — Rec: Bundle pinned versions; refresh manually.
8. Trustee pack composition — Rec: Server-side PyPDF2 merge; one file out.
9. Excel "DB export" — Rec: Multi-sheet workbook starting Phase A; keep old endpoint as shim for one release.