Data-plane helpers¶

params ¶

Query-parameter parsing for the SDMX-REST data plane.

Everything here is driven by the request's query string plus a sdmxlib.DataflowBinding — no catalog-specific knowledge. A consumer wires these into its own /data handler.

sql_ident ¶

sql_ident(name: str) -> str

Quote a SQL identifier. Source is producer-controlled binding metadata.

parse_bracketed ¶

parse_bracketed(
    items: Iterable[tuple[str, str]], pattern: Pattern[str]
) -> dict[str, list[str]]

Extract bracketed query params (c[<id>]=… / within[<id>]=…).

Repeated keys (foo[X]=A&foo[X]=B) and +-joined values (foo[X]=A+B) are both accepted and merged into one value list. Non-matching items are ignored.

parse_c_params ¶

parse_c_params(
    items: Iterable[tuple[str, str]],
) -> tuple[
    dict[str, list[str]], dict[str, list[tuple[str, str]]]
]

Parse c[<id>]=<expr> filters per SDMX-REST 2.2.2.

Each <expr> is a +-joined list of terms; each term may carry an optional <op>: prefix where <op> is one of eq/ne/lt/ le/gt/ge/co/nc/sw/ew. Default (no prefix) is eq. The + separator is interpreted as a per-component AND across operator terms — matches the spec example c[TIME_PERIOD]=ge:2020-01+le:2020-12.

Parameters:

Name	Type	Description	Default
`items`	`Iterable[tuple[str, str]]`	Output of `request.query_params.multi_items()`.	required

Returns:

Type	Description
`dict[str, list[str]]`	`(eq_values, non_eq_terms)`. `eq_values[dim_id]` is the list of
`dict[str, list[tuple[str, str]]]`	values matched as equals (multi-value collapses to `IN (…)` via
`tuple[dict[str, list[str]], dict[str, list[tuple[str, str]]]]`	sdmxlib's `ComponentFilter`). `non_eq_terms[dim_id]` is a list of
`tuple[dict[str, list[str]], dict[str, list[tuple[str, str]]]]`	`(op, value)` tuples for the nine remaining operators; the caller
`tuple[dict[str, list[str]], dict[str, list[tuple[str, str]]]]`	renders them as SQL fragments.

like_pattern ¶

like_pattern(op: str, raw: str) -> str

Wrap a user value in LIKE wildcards per operator semantics, escaping LIKE metachars.

build_non_eq_where ¶

build_non_eq_where(
    non_eq_terms: dict[str, list[tuple[str, str]]],
    binding: DataflowBinding,
) -> tuple[list[str], list[object]]

Render {dim_id: [(op, value), …]} as (fragments, params) SQL clauses.

Resolves each component id against the binding's dimension + attribute column maps; unknown ids raise HTTPException(400) so callers see exactly which one was wrong. Each (op, value) becomes one WHERE fragment using the template in NON_EQ_OPERATOR_SQL.

For LIKE-based operators (co/nc/sw/ew) the user-supplied value is escaped (%, _, \\) and wrapped in % wildcards per operator semantics — so co:Census matches …Census…, sw:Census matches Census…, ew:Province matches …Province, nc:foo is the NOT counterpart of co:foo.

validate_dimension_at_observation ¶

validate_dimension_at_observation(
    value: str | None, binding: DataflowBinding
) -> None

Validate the SDMX-REST 2.2.2 dimensionAtObservation param.

Accepts None (param omitted — defaults to AllDimensions), the literal AllDimensions, or any key-dimension ID declared in the binding. Anything else is 400.

parse_sort ¶

parse_sort(
    raw: str, binding: DataflowBinding
) -> list[tuple[str, Literal["ASC", "DESC"]]]

Parse SDMX-REST 2.2.2 sort= into [(physical_column, direction), …].

Syntax (per spec + upstream PaginationHelper.cs):

comp_id[:asc|:desc] — single component
comp_id[:asc|:desc]+comp_id[:asc|:desc]… — multi-component, +-joined
*[:asc|:desc] — every key dimension in DSD order
series_key[:asc|:desc] — every non-TIME key dimension in DSD order, for clients wanting a stable series listing independent of TIME_PERIOD

series_key combined with dimensionAtObservation=AllDimensions is upstream-invalid (AllDimensions already flattens series); that check belongs in the caller, which sees both params.

Each segment names a key dimension or attribute. Unknown ids raise HTTPException(400) so callers see exactly which one was wrong.

reject_unsupported_standard_params ¶

reject_unsupported_standard_params(
    *,
    includeHistory: bool | None,
    updatedAfter: str | None,
    reportingYearStartDay: str | None,
) -> None

Return 501 if the caller asked for a SDMX-REST 2.2.2 param this server won't implement.

Consumers backed by a single static edition have no version history and no fiscal-period semantics, so these three can never be implemented for them.

projection ¶

Column projection for the data plane — SDMX-REST attributes= / measures=.

Projection is applied by trimming the binding's column maps before they reach the serialiser: the underlying query still SELECT *s, and the writer emits only the columns the trimmed binding lists. The arrow/parquet path needs the projection applied to the batches themselves, which project_arrow_batches does.

resolve_projection_set ¶

resolve_projection_set(
    available: dict[str, str],
    raw: str | None,
    *,
    keep_all_keywords: frozenset[str],
    drop_all_keywords: frozenset[str],
    kind: str,
) -> set[str] | None

Parse a SDMX-REST 2.2.2 projection param.

Returns None (no filter — keep every column) when the param is missing or matches a "keep all" keyword, the empty set on none or any drop_all_keywords token, or the set of IDs to retain on a comma-separated list. Raises HTTPException(400) if any listed ID isn't in available.

project_binding ¶

project_binding(
    binding: DataflowBinding,
    *,
    attributes_param: str | None,
    measures_param: str | None,
) -> sl.DataflowBinding

Return a copy of binding with attribute / measure column maps projected.

Parses the SDMX-REST 2.2.2 attributes= / measures= query strings — a comma-separated list of IDs, the keyword all / dsd (keep everything, the default), or none (drop everything). Unknown IDs raise so callers surface them as 400. Dimension columns are never projected (they always appear in the response).

project_arrow_batches ¶

project_arrow_batches(
    batches: Iterable[RecordBatch], binding: DataflowBinding
) -> Iterator[pa.RecordBatch]

Yield each arrow batch projected to the columns the binding declares.

The CSV writer reads from the trimmed binding to pick columns, but a parquet writer fed the raw SELECT * batches would emit everything — so the arrow path has to project explicitly to honour attributes= / measures=.

materialise_dsd_for_flow ¶

materialise_dsd_for_flow(
    registry: EditionRegistry, flow_urn: SdmxUrn[Dataflow]
) -> sl.DataStructure | None

Resolve flow_urn to its bound DSD, materialised.

sdmxlib's SDMX-JSON writers need the DSD object (not just a URN) to pull localised dimension / attribute / code labels.

key_dimension_ids ¶

key_dimension_ids(
    registry: EditionRegistry,
    flow_urn: SdmxUrn[Dataflow],
    binding: DataflowBinding,
) -> list[str]

Resolve the flow's key-dimension ids in DSD order.

Dotted-key parsing is positional, so the order is the contract — segment i of A5.2021.._T means whatever dimension sits at index i here. The DSD is authoritative; the binding's column map is the fallback for flows whose DSD is missing or not materialisable, which is lossy only if the producer wrote that map in a different order.

Memoised per registry and flow URN: the answer is fixed for an edition, and the lookup is on the hot path of every /data and /availability request. A reopened edition is a new registry, so it starts empty.

Deliberately does not materialise a lazy DSD — an already-materialised one is used, anything else takes the binding fallback. Forcing the materialise here would change the dimension order this returns for lazy flows, which would silently re-interpret every dotted key against them.

Parameters:

Name	Type	Description	Default
`registry`	`EditionRegistry`	Registry used to resolve the flow and its DSD.	required
`flow_urn`	`SdmxUrn[Dataflow]`	URN of the dataflow.	required
`binding`	`DataflowBinding`	Supplies the fallback dimension order.	required

Returns:

Type	Description
`list[str]`	The key-dimension ids, in the order dotted keys are parsed against.

label_maps_for_dims ¶

label_maps_for_dims(
    registry: EditionRegistry,
    dim_codelist_urns: dict[str, SdmxUrn[Codelist]],
    binding: DataflowBinding,
    lang: str,
) -> dict[str, dict[str, str]]

Map each key dim bound to a codelist to its code labels.

Takes the dim_id -> codelist URN mapping as an argument so consumers can supply a cached one. Uses LazyCodelist.code_map, which is itself cached per (codelist_urn, lang). Dims without an enumeration, or whose codelist the registry cannot resolve, yield no entry — the CSV emit path passes their ids through unchanged.

Parameters:

Name	Type	Description	Default
`registry`	`EditionRegistry`	Registry used to resolve each codelist URN.	required
`dim_codelist_urns`	`dict[str, SdmxUrn[Codelist]]`	`dim_id -> codelist URN` for the flow's key dims.	required
`binding`	`DataflowBinding`	Supplies the `dim_id -> physical column` mapping.	required
`lang`	`str`	Language tag passed to `code_map`.	required

Returns:

Type	Description
`dict[str, dict[str, str]]`	`{column_name: {code_id: label}}`, keyed on the physical column.

sql ¶

SQL fragment builders driven by the dataflow binding.

These shape the generated query. Each one is binding-driven and degrades to a no-op when the columns it needs are absent, so a catalog that doesn't model them pays nothing.

decimals_format_replace_clause ¶

decimals_format_replace_clause(
    binding: DataflowBinding,
    *,
    targets: Iterable[str],
    decimals_id: str = "DECIMALS",
) -> str

Build REPLACE arguments formatting numeric columns to per-row precision.

Wraps each target in printf('%.*f', DECIMALS, value) so the row's own DECIMALS attribute drives the decimal places.

The problem this solves is generic to SDMX: a single OBS_VALUE column typed DOUBLE has to carry counts, percentages, rates and medians at once, so the wire format needs per-row precision. DECIMALS is the standard attribute that carries it. NULL inputs (missing DECIMALS, suppressed value) propagate to NULL — printf short-circuits.

Each target is looked up in the measure map and the attribute map: a producer may model confidence bounds either way, and the underlying column is numeric regardless.

Parameters:

Name	Type	Description	Default
`binding`	`DataflowBinding`	Supplies both column maps.	required
`targets`	`Iterable[str]`	Component IDs to format, e.g. `("OBS_VALUE",)`.	required
`decimals_id`	`str`	Attribute ID carrying the per-row precision.	`'DECIMALS'`

Returns:

Type	Description
`str`	The `REPLACE` arguments, or an empty string when the binding
`str`	declares no `decimals_id` attribute or none of the targets resolve.
`str`	In either case a plain `SELECT *` is correct.

qualify_is_active ¶

qualify_is_active(cardinality: int, n_cap: int) -> bool

Decide whether firstNObservations / lastNObservations needs a QUALIFY.

The QUALIFY row_number() OVER (PARTITION BY <series> ORDER BY <time>) <= N pattern is expensive: the Window operator blocks partition pruning from being pushed through to the parquet read (measured ~85x slower than the bare scan on a filtered census query). When the time dimension's cardinality is already <= N the cap would keep every row anyway, so the QUALIFY can be skipped entirely for the same result.

A cardinality of 0 means "unknown" and takes the safe path — apply the QUALIFY.

Parameters:

Name	Type	Description	Default
`cardinality`	`int`	Distinct values of the time dimension for this flow. Callers typically supply a cached probe.	required
`n_cap`	`int`	The requested first/last N.	required

Returns:

Type	Description
`bool`	False when the cap would keep every row, so the QUALIFY can be skipped.

paging ¶

Range request / Content-Range response handling for observation slices.

SDMX-REST has no pagination of its own for the data plane; clients that need it (the .Stat Data Explorer in particular) reuse HTTP Range with a row-count unit in place of bytes. This module parses the request side and builds the matching response headers.

parse_items_range ¶

parse_items_range(
    header: str | None,
) -> tuple[int, int, str] | None

Parse a Range: <unit>=<start>-<end> header into (start, end, unit).

The unit is echoed back on the response so Range: values=0-9 yields Content-Range: values 0-9/<total>.

Returns None for an absent header or any other unit (e.g. bytes) — callers fall through to the unranged stream. Validation of the resulting bounds belongs to the caller, which can raise 400 with route context.

build_range_headers ¶

build_range_headers(
    start: int, end: int, total: int, unit: str
) -> dict[str, str]

Build the Content-Range + Accept-Ranges headers for a 206 response.

with_range ¶

with_range(
    base_headers: dict[str, str],
    range_info: tuple[int, int, int, str] | None,
    *,
    sort_skipped: bool = False,
) -> dict[str, str]

Layer Content-Range / Accept-Ranges onto a headers dict when paginating.

sort_skipped=True adds X-Sort-Skipped: true — for callers that enforce a ceiling on sorted pagination and served an unordered first page instead of an expensive global sort.

range_status ¶

range_status(
    range_info: tuple[int, int, int, str] | None,
) -> int

206 Partial Content when the response is a Range slice, 200 OK otherwise.

encoding ¶

Wire-format encoders for observation streams.

Every encoder here takes an iterable of Arrow record batches (or of CSV byte chunks) and yields bytes, so memory stays bounded by batch size. Bulk SDMX downloads are routinely larger than a server process's memory budget.

emit_csv_with_labels ¶

emit_csv_with_labels(
    batches: Iterable[RecordBatch],
    binding: DataflowBinding,
    *,
    structure_id: str,
    label_maps: dict[str, dict[str, str]],
    structure_kind: str = "dataflow",
    action: str = "I",
) -> Iterator[bytes]

SDMX-CSV writer emitting an extra <dim>_label column after each coded dim.

Mirrors the shape sdmxlib's write_sdmx_csv produces (STRUCTURE, STRUCTURE_ID, ACTION, dims, measures, attrs) but interleaves a <dim>_label column after each dim column present in label_maps. Attributes and measures pass through unchanged — labels only make sense for enumerated key dims.

Parameters:

Name	Type	Description	Default
`batches`	`Iterable[RecordBatch]`	The result rows, in the binding's physical column names.	required
`binding`	`DataflowBinding`	Maps each component to its physical column.	required
`structure_id`	`str`	The `STRUCTURE_ID` value.	required
`label_maps`	`dict[str, dict[str, str]]`	`{column_name: {code_id: label}}`, from `label_maps_for_dims`. A dim missing from it gets no label column.	required
`structure_kind`	`str`	The `STRUCTURE` value.	`'dataflow'`
`action`	`str`	The `ACTION` value.	`'I'`

Yields:

Type	Description
`bytes`	One encoded chunk per batch, the first carrying the header line; the
`bytes`	header alone when `batches` is empty.

Raises:

Type	Description
`KeyError`	If a batch is missing a column the binding declares.

prepend_utf8_bom ¶

prepend_utf8_bom(
    csv_chunks: Iterable[bytes],
) -> Iterator[bytes]

Emit a leading UTF-8 BOM then forward every chunk from csv_chunks.

For the SDMX-REST 2.2.2 format=csvfile / csvfilewithlabels download flow: Excel opens a UTF-8 CSV cleanly with the BOM present and mangles diacritics without it. Upstream gates this on file=true.

stream_csv_zip ¶

stream_csv_zip(
    csv_chunks: Iterable[bytes], *, inner_filename: str
) -> Iterator[bytes]

Wrap a CSV byte iterator in a single-entry streaming zip.

Uses zipstream-ng so the central directory is appended at the end of the body — no rewinding, no buffering of the whole archive. Compression level 1 matches the gzip path's CPU budget; the same DEFLATE runs underneath, so the speed/ratio trade-off is identical.

encode_parquet ¶

encode_parquet(
    batches: Iterable[RecordBatch],
) -> Iterator[bytes]

Stream Arrow batches as a parquet file.

The parquet wire format is sequential — magic header, row groups, then a footer ending in the file metadata. Readers consume in order and parse the footer last, so HTTP streaming is fine as long as no Content-Length is advertised. Each yielded chunk is whatever the ParquetWriter flushed since the previous yield; the final yield carries the closing footer.

Memory is bounded by row-group size (one RecordBatch), and TTFB is the first batch's scan + encode.

write_statistics and use_dictionary are off deliberately: SDMX-CSV-shaped data is string-heavy with high repetition, zstd compresses the raw bytes better than dictionary encoding represents them, and the page min/max scan costs CPU that bulk-download readers get nothing for (they don't do row-group predicate pushdown).