contracts.rules_data_shapes

Page actions AI-ready formats and sharing
Open LLM text
Share with AI
Ask Claude Ask ChatGPT Ask Gemini Ask Copilot
module

Typed-SQL column-mapping contract (SQL in, frozen dataclasses out).

Chirp's data layer maps SQL rows onto frozen dataclasses via db.fetch(cls, sql) / db.fetch_one(cls, sql) / db.stream(cls, sql) (chirp.data._mapping). A column SELECTed…

Typed-SQL column-mapping contract (SQL in, frozen dataclasses out).

Chirp's data layer maps SQL rows onto frozen dataclasses via db.fetch(cls, sql) / db.fetch_one(cls, sql) / db.stream(cls, sql) (chirp.data._mapping). A column SELECTed by the SQL that exists on neither the target dataclass nor the declared table schema is drift: the query runs, the column is silently ignored bymap_row(extra columns are dropped), and the bug only surfaces when the row is missing data at runtime.

This rule promotes that drift to anapp.check() ERRORwhen it is statically analyzable. It is deliberately conservative -- it does not become an ORM, a model registry, or a SQL engine. It only fires when all of the following can be resolved from static handler source:

  • a literal first positionalclsargument that names a frozen dataclass reachable in the handler's module globals, and
  • a string-literal SQL with an explicitSELECT col, col, ...list (SELECT *, expressions, joins with ambiguous columns, and dynamic SQL are skipped -- no false positives).

A SELECTed column is flagged only when it is absent from the dataclass fields and (when a declared schema is available) absent from every table column in that schema. That double-guard keeps the check quiet for column aliases the schema cannot see and for db-less / schema-less apps, honoring the data steward's "data stays optional" rule and the contracts steward's "no noisy warnings" rule.

Severity isERRORand overridable via app.override_contract_severity("data", Severity.ERROR/WARNING/INFO).

contracts.rules_data_shapes

Name Type Default Description
type —
qualified_name —
element_type —
description —
source_file —
line_number —
is_autodoc —
autodoc_element —
_autodoc_template —
_autodoc_url_path —
_autodoc_page_type —
title —
doc_content_hash —

Symbols on this page

function _strip_sql_comments

Returnsqlwith SQL comments replaced by a single space, string-aware.

_parse_select_columnsis the single-source-of-truth SELECT parser (also consumed by codegen and the data…

Jump to symbol function _split_top_level

Split a projection on top-level commas (ignoring commas inside parens).

Jump to symbol function _parse_select_columns

Return the output column names of a simpleSELECT a, bquery.

ReturnsNone(skip -- not statically analyzable) when the projection is SELECT…

Jump to symbol function _dataclass_fields

Return the field-name set for a frozen dataclass type, elseNone.

Jump to symbol function _resolve_cls

Resolve aclsargument AST node to a runtime object, if statically known.

Only resolves bareNamereferences against the handler's module globals (e.g. …

Jump to symbol function _is_db_receiver

True when the call receiver looks like a chirp.dataDatabasehandle.

Recognizesdb / database / *_dbnames, attribute access ending in those (…

Jump to symbol function _iter_fetch_calls

Yield(cls_node, sql_literal)for statically analyzable fetch calls.

A call is analyzable when it is <db>.fetch|fetch_one|stream(CLS, "literal sql", ...) where <db>is an idiomatic…

Jump to symbol function _schema_columns

Union of all column names across all tables in the declared schema.

ReturnsNonewhen no schema is available -- the check then relies…

Jump to symbol function _dedent

Normalize indentation soast.parseaccepts a nested handler source.

Jump to symbol function check_data_shapes

Flagdb.fetch(cls, sql)columns that map to no dataclass field.

For each statically-resolvable fetch call on a route handler:

  • resolveclsto a frozen…
Jump to symbol
_strip_sql_comments
function 
def _strip_sql_comments(sql: str) -> str

Returnsqlwith SQL comments replaced by a single space, string-aware.

_parse_select_columnsis the single-source-of-truth SELECT parser (also consumed by codegen and the data contract, and -- viashapes._scope_injectable -- the tenant-scope injectability gate). A comment in the PROJECTION list (SELECT id /* note */, name FROM t or SELECT id -- c\nFROM t) is NOT opaque SQL, but a naive regex parse choked on it and returnedNone-- surfacing a misleading "CTE / UNION / SELECT * / derived-table" rejection for what is really just a comment. We strip comments first so a commented column list still parses.

Comment forms (mirroring the SQL standard): line comments-- ... EOLand block comments/* ... */(non-nesting). Each is replaced by a single space (never deleted) so adjacent tokens cannot merge (a/**/bmust not becomeab). The scan is string-literal aware: a -- / /*inside a'...' or "..." literal (honoring doubled-quote '' / "" escapes) is NOT a comment and is preserved verbatim, so a quoted identifier or string body is never corrupted. A line comment's terminating newline is preserved (kept outside the replacement) so line structure is unchanged.

Parameters

Name Type Default Description
sql str —
_split_top_level
function 
def _split_top_level(projection: str) -> list[str]

Split a projection on top-level commas (ignoring commas inside parens).

Parameters

Name Type Default Description
projection str —
_parse_select_columns
function 
def _parse_select_columns(sql: str) -> tuple[str, ...] | None

Return the output column names of a simpleSELECT a, bquery.

ReturnsNone(skip -- not statically analyzable) when the projection is SELECT *, contains an expression / function call / subquery / DISTINCT, or the SQL has no analyzableSELECT ... FROMshape. Returns the resolved output names (alias-aware) otherwise.

Parameters

Name Type Default Description
sql str —
_dataclass_fields
function 
def _dataclass_fields(cls: Any) -> set[str] | None

Return the field-name set for a frozen dataclass type, elseNone.

Parameters

Name Type Default Description
cls Any —
_resolve_cls
function 
def _resolve_cls(node: ast.expr, handler_globals: dict[str, Any]) -> Any

Resolve aclsargument AST node to a runtime object, if statically known.

Only resolves bareNamereferences against the handler's module globals (e.g.db.fetch(User, ...) where Useris a module-level dataclass). Attribute access, subscripts, calls, and locals are not resolvable -> skip.

Parameters

Name Type Default Description
node ast.expr —
handler_globals dict[str, Any] —
_is_db_receiver
function 
def _is_db_receiver(value: ast.expr) -> bool

True when the call receiver looks like a chirp.dataDatabasehandle.

Recognizesdb / database / *_dbnames, attribute access ending in those (self.db, app.db, g.db), and get_db()calls. Anything else (an arbitrary object that merely exposes afetch/streammethod) is rejected so the rule never ERRORs on an unrelated API.

Parameters

Name Type Default Description
value ast.expr —
_iter_fetch_calls
function 
def _iter_fetch_calls(tree: ast.AST) -> list[tuple[ast.expr, str]]

Yield(cls_node, sql_literal)for statically analyzable fetch calls.

A call is analyzable when it is <db>.fetch|fetch_one|stream(CLS, "literal sql", ...) where <db>is an idiomatic chirp.dataDatabase receiver (see _is_db_receiver()), with a string-literal SQL and at least theclspositional argument. Dynamic SQL (f-strings, concatenation, names) and calls on unrelated receivers are skipped.

Parameters

Name Type Default Description
tree ast.AST —
_schema_columns
function 
def _schema_columns(schema: SchemaSnapshot | None) -> set[str] | None

Union of all column names across all tables in the declared schema.

ReturnsNonewhen no schema is available -- the check then relies on the dataclass fields alone and only flags columns that match no field, which is still a real drift (a SELECTed column the dataclass cannot receive).

Parameters

Name Type Default Description
schema SchemaSnapshot | None —
_dedent
function 
def _dedent(src: str) -> str

Normalize indentation soast.parseaccepts a nested handler source.

Parameters

Name Type Default Description
src str —
check_data_shapes
function 
def check_data_shapes(router: object, schema: SchemaSnapshot | None) -> list[ContractIssue]

Flagdb.fetch(cls, sql)columns that map to no dataclass field.

For each statically-resolvable fetch call on a route handler:

  • resolveclsto a frozen dataclass (skip if not resolvable),
  • parse the SELECT column list (skipSELECT */ expressions / dynamic SQL),
  • a SELECTed column absent from the dataclass fields and (when a schema is available) absent from every declared table column is real drift ->ERROR.

Returns one issue per (route, dataclass, offending column).

Parameters

Name Type Default Description
router object —
schema SchemaSnapshot | None —

View source · /home/runner/work/chirp/chirp/site/../src/chirp/contracts/rules_data_shapes.py:1