Skip to content

generate_report

Main library function — the entry point for generating any report.

generate_report(*, data_source: T, spec: ReportSpec, destination: str | Path, input_adapter: InputAdapter[T] | None = None, renderer_registry: dict[str, Renderer] | None = None) -> Path

Source code in src/pyreps/service.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def generate_report[T](
    *,
    data_source: T,
    spec: ReportSpec,
    destination: str | Path,
    input_adapter: InputAdapter[T] | None = None,
    renderer_registry: dict[str, Renderer] | None = None,
) -> Path:
    adapter = input_adapter or _resolve_adapter(data_source)
    logger.debug("adapter resolved: %s", type(adapter).__name__)

    records = adapter.adapt(data_source)
    # Rastreia erros de streaming vindos do adaptador (ex: queda de conexão SQL)
    tracked_records = track_stream(records, "adapter", InputAdapterError)

    mapped_rows = map_records(tracked_records, spec)
    # Rastreia erros de mapeamento e coerção durante o streaming
    tracked_rows = track_stream(mapped_rows, "mapping", MappingError)

    registry = renderer_registry or default_renderer_registry()
    renderer = registry.get(spec.output_format)
    if renderer is None:
        raise ReportError(f"no renderer registered for format '{spec.output_format}'")

    output_path = Path(destination)
    with _report_transaction(output_path, spec.output_format):
        return renderer.render(tracked_rows, spec, output_path)

Example

from pyreps import ColumnSpec, ReportSpec, generate_report

spec = ReportSpec(
    output_format="xlsx",
    columns=[
        ColumnSpec(label="ID", source="id", type="int", required=True),
        ColumnSpec(label="Name", source="name"),
    ],
)

path = generate_report(
    data_source=[{"id": 1, "name": "Ana"}],
    spec=spec,
    destination="report.xlsx",
)