Module `trase.tools.data_release_package.snapshot.ingesters`

Reads the parts of a data release package and writes them to PostgreSQL.

Every ingester here descends from TableIngester (see trase.database.ingest_from_s3), so they all share its machinery: chunked writes with a progress bar, a common table owner, and a table comment recording who snapshotted what, from where and when.

The data ingesters go one step further and descend from DataframeIngester, which already knows how to read a Parquet file from S3. All they add is where the S3 path comes from – a *.data.txt pointer file committed to Git – and, for the spatial layers, what to do with a geometry column.

Functions

def split_s3_url(url: str) ‑> Tuple[str, str]: Splits 's3://bucket/some/key.parquet' into ('bucket', 'some/key.parquet').

Classes

class DataPointerParquetIngester (destination_table: str, destination_schema: str = 's3', chunk_size: int = 100000, source_s3_key: str = None, source_s3_bucket: str = 'trase-storage', pointer_path: pathlib.Path = None)

Snapshots the Parquet file behind one *.data.txt pointer.

The data itself is too large for Git, so the package commits a pointer to it instead: a text file holding the fully-qualified S3 path. Resolving that pointer is the only thing this adds to DataframeIngester, which does the rest.

Attributes

pointer_path : Path: The *.data.txt file to follow.

Ancestors

trase.database.ingest_from_s3.DataframeIngester
trase.database.ingest_from_s3.TableIngester

Subclasses

SpatialLayersIngester

Instance variables

var pointer_path : pathlib.Path

class JsonIngester (destination_table: str, destination_schema: str = 's3', chunk_size: int = 100000, source_path: pathlib.Path = None)

Snapshots one metadata/*.json file of a data release package.

Unlike the data, metadata is small and lives in Git rather than S3, so it is read straight from the package folder. metadata/context.json holds a single object and the rest hold arrays of them; either way the result is one row per object.

Attributes

source_path : Path: The local JSON file to snapshot.

Ancestors

trase.database.ingest_from_s3.TableIngester

Static methods

def flatten_record(record: dict) ‑> dict: JSON-encodes any nested value, so that every column is a scalar.

Instance variables

var source_path : pathlib.Path
prop source_uri : str: Where the data is read from, for progress messages and the table comment.

Methods

def load_data(self) ‑> polars.dataframe.frame.DataFrame: Read the source data into a Polars DataFrame.

class SpatialLayersIngester (destination_table: str, destination_schema: str = 's3', chunk_size: int = 100000, source_s3_key: str = None, source_s3_bucket: str = 'trase-storage', pointer_path: pathlib.Path = None, pointer_paths: Tuple[pathlib.Path, ...] = (), geometry_column: str = 'geometry')

Snapshots every spatial_layers/*.data.txt pointer into the one table.

The specification lets a data scientist split the entity registry over as many layer files as suits them, but requires that they all share one schema, because consumers read them as a single table. That is what this does: it unions the layers on the way in.

Geometry needs a second step. A GeoParquet geometry column arrives as WKB, and Polars has no geometry type to hold it, so it is written as hex text and converted into a PostGIS geometry column afterwards, within the same transaction.

Attributes

pointer_paths : tuple: The *.data.txt files to follow, one per layer.

Ancestors

DataPointerParquetIngester
trase.database.ingest_from_s3.DataframeIngester
trase.database.ingest_from_s3.TableIngester

Instance variables

var geometry_column : str
var pointer_paths : Tuple[pathlib.Path, ...]
prop source_uri : str: Where the data is read from, for progress messages and the table comment.

Methods

def after_ingest(self, connection, schema: str, table: str) ‑> None: Converts the hex WKB written by modify_data into a PostGIS column.
def describe_source(self) ‑> str: The S3 paths, each with the version of the object that was read.
def has_wkb_geometry(self, data: polars.dataframe.frame.DataFrame) ‑> bool: True if data carries a geometry column that arrived as WKB.
def load_data(self) ‑> polars.dataframe.frame.DataFrame: Read the source data into a Polars DataFrame.
def modify_data(self, data: polars.dataframe.frame.DataFrame) ‑> polars.dataframe.frame.DataFrame: Writes WKB geometry as hex, ready for after_ingest to convert.