pudl.dbt_schema#

Define dbt schema types and merging logic.

We generate dbt schema.yml files by translating our metadata into schema.yml format, then applying human-sourced patches to the auto-generated schemas.

Attributes#

`_DESCRIPTION_WRAP_WIDTH`
`_NormalizedDescription`
`_NormalizedDataTests`

Classes#

`_LiteralStr`	Marker subclass telling the dumper to use YAML literal block style (`\|`).
`DbtColumn`	Define yaml structure of a dbt column.
`DbtTable`	Define yaml structure of a dbt table.
`DbtSource`	Define basic dbt yml structure to add a pudl table as a dbt source.
`DbtSchema`	Define basic structure of a dbt models yaml file.

Functions#

`_normalize_whitespace`(→ str)	Collapse all whitespace (including blank lines) to single spaces.
`_normalize_descriptions`(→ Any)	Recursively collapse whitespace in every `description` field.
`_normalize_description_field`(→ str \| None)
`_normalize_data_tests_field`(→ list \| None)
`_wrap_description`(→ str \| _LiteralStr)	Re-wrap a description string to short lines, for readability on disk.
`_wrap_descriptions`(→ Any)	Recursively re-wrap every `description` field in a dumped schema dict.
`_prettier_yaml_dumps`(→ str)	Dump YAML to string that Prettier likes.
`_foreign_key_data_tests`(→ list[dict] \| None)	Build `foreign_key` data test entries for a resource's outgoing FKs.
`merge_schema`(→ DbtSchema)	Merge two DbtSchemas by applying human-schema as a patch on top of machine-schema.
`merge_by_name`(→ list)	Perform a generic merge of two lists of dbt elements, matching by name.
`merge_sources_by_name`(→ list[DbtSource])	Match machine/human sources by name, then merge them.
`merge_source`(→ DbtSource)	Merge two DbtSources by applying human-source as a patch on top of machine-source.
`merge_tables_by_name`(→ list[DbtTable])	Match machine/human tables by name, then merge them.
`merge_table`(→ DbtTable)	Merge two DbtTables by applying human-table as a patch on top of machine-table.
`merge_columns_by_name`(→ list[DbtColumn])	Match machine/human columns by name, then merge them.
`merge_column`(→ DbtColumn)	Merge two DbtColumns by applying human-column as a patch on top of machine-column.

Module Contents#

pudl.dbt_schema._DESCRIPTION_WRAP_WIDTH = 88[source]#

pudl.dbt_schema._normalize_whitespace(text: str) → str[source]#: Collapse all whitespace (including blank lines) to single spaces.

pudl.dbt_schema._normalize_descriptions(obj: Any) → Any[source]#

Recursively collapse whitespace in every description field.

Normalizing whitespace at parse time reduces spurious diffs and round trip errors, treating all whitespace as semantically identical.

The description fields show up in two places: the typed description field on DbtColumn/DbtTable/DbtSource, and nested inside arbitrary data_tests entries which are untyped list content that pydantic doesn’t otherwise inspect. This function is used as a validator for both.

pudl.dbt_schema._normalize_description_field(value: str | None) → str | None[source]#

pudl.dbt_schema._normalize_data_tests_field(value: list | None) → list | None[source]#

pudl.dbt_schema._NormalizedDescription[source]#

pudl.dbt_schema._NormalizedDataTests[source]#

class pudl.dbt_schema._LiteralStr[source]#

Bases: str

Marker subclass telling the dumper to use YAML literal block style (|).

Only used for description fields that we’ve re-wrapped ourselves, so they read as human-friendly paragraphs on disk instead of one giant line. Everything else (regexes, SQL snippets, argument lists) is left completely alone, since forcing a global line width in the dumper risks reflowing content where whitespace is significant.

pudl.dbt_schema._wrap_description(text: str) → str | _LiteralStr[source]#

Re-wrap a description string to short lines, for readability on disk.

A description may already contain embedded newlines (e.g. from a blank line in a folded YAML block scalar, or from a previous pass of this same function). We preserve that line structure and only wrap the text within each line, so we don’t invent new paragraph breaks the human didn’t write. We use block style whenever the result spans multiple lines – including when wrapping made no change to an already-wrapped multi-line value – since a bare multi-line str would otherwise fall back to an ugly quoted flow scalar. Single-line results are left as a plain string so short descriptions keep their current compact description: ... formatting.

pudl.dbt_schema._wrap_descriptions(obj: Any) → Any[source]#: Recursively re-wrap every description field in a dumped schema dict.

pudl.dbt_schema._prettier_yaml_dumps(yaml_contents: dict[str, Any]) → str[source]#: Dump YAML to string that Prettier likes.

pudl.dbt_schema._foreign_key_data_tests(resource: pudl.metadata.classes.Resource) → list[dict] | None[source]#

Build foreign_key data test entries for a resource’s outgoing FKs.

One entry per foreign key relationship declared on the resource, in declaration order, so regenerating a table’s schema.yml produces a stable diff.

class pudl.dbt_schema.DbtColumn(/, **data: Any)[source]#

Bases: pydantic.BaseModel

Define yaml structure of a dbt column.

model_config[source]#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str[source]#

description: _NormalizedDescription = None[source]#

data_tests: _NormalizedDataTests = None[source]#

meta: dict | None = None[source]#

tags: list[str] | None = None[source]#

class pudl.dbt_schema.DbtTable(/, **data: Any)[source]#

Bases: pydantic.BaseModel

Define yaml structure of a dbt table.

model_config[source]#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str[source]#

description: _NormalizedDescription = None[source]#

data_tests: _NormalizedDataTests = None[source]#

columns: list[DbtColumn] | None = None[source]#

meta: dict | None = None[source]#

tags: list[str] | None = None[source]#

config: dict | None = None[source]#

classmethod from_table_name(table_name: str) → DbtTable[source]#: Construct configuration defining table from PUDL metadata.

class pudl.dbt_schema.DbtSource(/, **data: Any)[source]#

Bases: pydantic.BaseModel

Define basic dbt yml structure to add a pudl table as a dbt source.

model_config[source]#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str = 'pudl'[source]#

tables: list[DbtTable] | None = None[source]#

description: _NormalizedDescription = None[source]#

meta: dict | None = None[source]#

class pudl.dbt_schema.DbtSchema(/, **data: Any)[source]#

Bases: pydantic.BaseModel

Define basic structure of a dbt models yaml file.

model_config[source]#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

version: int = 2[source]#

sources: list[DbtSource] | None = None[source]#

models: list[DbtTable] | None = None[source]#

classmethod from_table_name(table_name: str) → DbtSchema[source]#: Construct configuration defining table from PUDL metadata.

classmethod from_yaml(schema_path: pathlib.Path) → DbtSchema[source]#: Load a DbtSchema object from a YAML file.

to_yaml(schema_path: pathlib.Path)[source]#: Write DbtSchema object to YAML file.

validate_humanity()[source]#

Make sure the human schema matches expectations.

We expect that all human overrides on source tables are data tests or column-level data tests. We allow the ‘name’ field so we can match human tables/columns with machine ones.

We do not have any expectations about model definitions since those are human-only.

pudl.dbt_schema.merge_schema(machine_schema: DbtSchema, human_schema: DbtSchema) → DbtSchema[source]#

Merge two DbtSchemas by applying human-schema as a patch on top of machine-schema.

Empty merged sources will be stored in the DbtSchema model as None to avoid serializing them.

pudl.dbt_schema.merge_by_name(machine_elements: list, human_elements: list, merger: collections.abc.Callable, element_factory: collections.abc.Callable) → list[source]#

Perform a generic merge of two lists of dbt elements, matching by name.

Parameters:

machine_elements – can be empty list.
human_elements – can be empty list.
merger – callable that takes two elements of the same dbt type (source, table, column) and returns a new element that is the merged version.
element_factory – callable that takes the element name and returns an empty instance - used if e.g. the human element doesn’t exist.

pudl.dbt_schema.merge_sources_by_name(machine_sources: list[DbtSource], human_sources: list[DbtSource]) → list[DbtSource][source]#: Match machine/human sources by name, then merge them.

pudl.dbt_schema.merge_source(machine_source: DbtSource, human_source: DbtSource) → DbtSource[source]#

Merge two DbtSources by applying human-source as a patch on top of machine-source.

Returns a deep copy of the machine source to avoid aliasing, updating with tables as the merge of the tables of the machine and human sources.

pudl.dbt_schema.merge_tables_by_name(machine_tables: list[DbtTable], human_tables: list[DbtTable]) → list[DbtTable][source]#: Match machine/human tables by name, then merge them.

pudl.dbt_schema.merge_table(machine_table: DbtTable, human_table: DbtTable) → DbtTable[source]#

Merge two DbtTables by applying human-table as a patch on top of machine-table.

Returns a deep copy of the machine table to avoid aliasing, updating with columns and table-level data tests as the merge of the respective machine and human data.

pudl.dbt_schema.merge_columns_by_name(machine_columns: list[DbtColumn], human_columns: list[DbtColumn]) → list[DbtColumn][source]#: Match machine/human columns by name, then merge them.

pudl.dbt_schema.merge_column(machine_column: DbtColumn, human_column: DbtColumn) → DbtColumn[source]#

Merge two DbtColumns by applying human-column as a patch on top of machine-column.

Returns a deep copy of the machine column to avoid aliasing, updating with data tests as the merge of the data tests of the machine and human columns.

Does not update any other attributes (descriptions, etc.).