Module trase.tools.data_release_package.snapshot.dbt_sources

Keeps the trase.earth dbt project's sources in step with what has been snapshotted.

The trase.earth pipeline reads a data release package through a dbt source named after the schema it was snapshotted into, and its models build their unions out of the tables that source declares. The declaration therefore has to say exactly what the package holds: a table declared but not snapshotted fails the build, and one snapshotted but not declared is invisible to it.

Not every package holds every table - data/spatial_metrics.data.txt and data/supply_chains.data.txt are both optional (see doc/Data-Release-Package.md) - so the declaration is what tells the pipeline which parts of a package there are to read. Rather than leave that to whoever snapshots a package to remember, the snapshot step writes it, here.

The declarations go in a file of their own, _data_package_sources.yml. dbt reads every *.yml under the models folder, so it makes no difference to the pipeline which file a source is declared in, but it makes a difference to us: _sources.yml next to it is hand-edited, and nothing written by a CLI belongs in a file people maintain by hand.

Functions

def declare_source(schema: str,
tables: Iterable[str],
path: pathlib.Path = PosixPath('/home/runner/work/TRASE/TRASE/trase/database/trase_earth_data_pipeline/models/trase_earth/_data_package_sources.yml')) ‑> bool

Declares schema as a dbt source holding exactly tables.

The entry is added if the file has none for schema, and replaced if it has one that says something else. Every other source in the file, and the comment at the top of it, is left untouched. The file is created if it is not there at all.

Args

schema : str

The schema the package was snapshotted into.

tables : iterable

The tables that were written into it.

path : Path

The sources file to edit.

Returns

bool

True if the file was changed, False if it already said this.

def declared_tables(entry) ‑> List[str]

The table names one source declaration holds.

def empty_document() ‑> ruamel.yaml.comments.CommentedMap

The file as it is written the first time, holding no packages yet.

def read_document(path: pathlib.Path, yaml: ruamel.yaml.main.YAML) ‑> ruamel.yaml.comments.CommentedMap

Reads the sources file, which the first package snapshotted has to create.

def source_entry(schema: str, tables: Iterable[str]) ‑> ruamel.yaml.comments.CommentedMap

The dbt source declaration for one snapshotted package.