Skip to content

DBT: Parish

File location: s3://trase-storage/ecuador/spatial/BOUNDARIES/out/parish.csv

DBT model name: parish

Explore on Metabase: Full table; summary statistics

Explore dependencies/lineage: link

Relies on script: trase/tools/aws/metadata.py

Description

This model was auto-generated based off .yml 'lineage' files in S3. The DBT model just raises an error; the actual script that created the data lives elsewhere. The script is located at trase/tools/aws/metadata.py [permalink]. It was last run by Harry Biddle.

Details

Column Type Description
geocode VARCHAR
name VARCHAR
DPA_VALOR VARCHAR
DPA_ANIO VARCHAR
canton_geocode VARCHAR
canton_name VARCHAR
DPA_PROVIN VARCHAR
DPA_DESPRO VARCHAR
parish_trase_id VARCHAR
canton VARCHAR

No data tests defined 🧐

Models

Sources

"""
This module creates and manages metadata for objects uploaded to AWS S3.

The aim of this metadata is to record some basic provenance for our S3 objects:

 - The script that produced the object (file path relative to Git and commit hash)
 - The person that ran the script

The metadata is written to a YAML file on disk (next to the data file) so that it can
be inspected before upload. It is *not* uploaded as a separate S3 object. Instead, the
same information is attached to the data object itself as S3 user-defined metadata (see
`s3_metadata_from_dict`). This can be viewed with, for example:

    aws s3api head-object --bucket trase-storage --key path/to/object.csv

or via the Trase CLI / boto3 `head_object` call (see the "Viewing S3 metadata" section
of the preprocessing documentation).

Metadata generation is always best-effort: if it fails for any reason (for example, the
script is not in a Git repository) the upload of the data object still proceeds.
"""

import dataclasses
import inspect
import json
import os
import posixpath
import subprocess
import sys
from getpass import getuser
from logging import getLogger
from tempfile import gettempdir
from typing import Optional

import yaml
from termcolor import colored

from trase.config import aws_session, settings

# include (True) or exclude (False) some objects from pdoc
__pdoc__ = {"DataclassEncoder": False, "NotAGitRepository": False}


LOGGER = getLogger(__name__)

# Keys used when storing the metadata as S3 user-defined metadata on the object itself.
TRASE_SCRIPT_PATH = "trase-script-path"
TRASE_COMMIT_HASH = "trase-commit-hash"
TRASE_GITHUB_USERNAME = "trase-github-username"
TRASE_GITHUB_REPOSITORY = "trase-github-repository"
TRASE_USER = "trase-user"


@dataclasses.dataclass
class Script:
    """The location of a GitHub script"""

    # Path of the file within the repository
    path: Optional[str]

    # Commit hash in the repository, if available
    commit_hash: Optional[str]

    # here we record in which GitHub repository the script is located. For now let's
    # just assume all scripts are located github.com/sei-international/TRASE. If we ever
    # use this python module for scripts that are in another repository we'll have to
    # come up with a clever solution
    github_username: Optional[str] = "sei-international"
    github_repository: Optional[str] = "TRASE"

    # currently the only supported location of a script is in a GitHub repository,
    # but we have a field called "type" just in case we want to add other sources in the
    # future
    type: Optional[str] = "github_script"


@dataclasses.dataclass
class Metadata:
    script: Script
    user: str


def generate_metadata(script_path: str) -> Metadata:
    try:
        script = get_metadata_from_git(script_path)
    except (NotAGitRepository, ValueError):
        # This is the fallback for when the script is not run from a Git repository
        # (for example, when the codebase is installed as a Python module). In that
        # case we record the path we were given but cannot determine a commit hash.
        script = Script(script_path, None)

    return Metadata(script=script, user=getuser())


class NotAGitRepository(Exception):
    pass


def _execute(command, script_path, **kwargs):
    # run a command inside the git repository of the file (we do not assume that
    # the file is in the TRASE repository)
    directory = os.path.dirname(script_path) or "." if script_path is not None else "."
    stdout = subprocess.check_output(command, cwd=directory, **kwargs)
    return stdout.decode().strip()


def get_metadata_from_git(script_path) -> Script:
    """
    Fetch git commit hash information: requires user to have "git" installed and also
    assumes that `file` is in a git repository
    """
    commit_hash = get_commit_hash(script_path)
    script_path = get_script_path_relative_to_root(script_path)

    return Script(script_path, commit_hash)


def get_commit_hash(script_path):
    try:
        commit_hash = _execute(
            ["git", "rev-parse", "HEAD"],
            script_path=script_path,
            stderr=subprocess.STDOUT,
        )
    except subprocess.CalledProcessError as e:
        if "fatal: not a git repository" in e.output.decode():
            LOGGER.warning(f"The script {script_path} is not in a Git repository")
            raise NotAGitRepository
        else:
            # some other error - maybe git isn't installed?
            LOGGER.exception("Could not find git")
            raise ValueError("Could not find git")
    except:
        LOGGER.exception("Could not find git")
        raise ValueError("Could not find git")

    return commit_hash


def get_script_path_relative_to_root(script_path):
    # ensure that the path is relative to the root of the git repository
    root = _execute(["git", "rev-parse", "--show-toplevel"], script_path)
    return os.path.relpath(script_path, root)


class DataclassEncoder(json.JSONEncoder):
    def default(self, o):
        if dataclasses.is_dataclass(o):
            return dataclasses.asdict(o)
        return super().default(o)


def dataclass_to_dict(object: dataclasses.dataclass) -> dict:
    data = json.dumps(object, cls=DataclassEncoder)
    return json.loads(data)


def write_metadata(metadata: Metadata, path):
    metadata_dictionary = dataclass_to_dict(metadata)
    with open(path, "w") as file:
        yaml.dump(metadata_dictionary, file)


def s3_metadata_from_dict(metadata: dict) -> dict:
    """Flatten a metadata dictionary (as produced by `dataclass_to_dict` or read back
    from the YAML file) into the string key/value pairs that can be attached to an S3
    object as user-defined metadata.

    Keys whose value is missing are omitted. All values are coerced to strings, as
    required by S3.
    """
    script = metadata.get("script") or {}
    values = {
        TRASE_SCRIPT_PATH: script.get("path"),
        TRASE_COMMIT_HASH: script.get("commit_hash"),
        TRASE_GITHUB_USERNAME: script.get("github_username"),
        TRASE_GITHUB_REPOSITORY: script.get("github_repository"),
        TRASE_USER: metadata.get("user"),
    }
    return {key: str(value) for key, value in values.items() if value is not None}


def s3_metadata_from_file(metadata_path: str) -> dict:
    """Read the metadata YAML file from disk and return the S3 user-defined metadata
    dictionary (see `s3_metadata_from_dict`). Returns an empty dictionary if the file
    does not exist."""
    if not os.path.isfile(metadata_path):
        LOGGER.warning(f"Metadata file {metadata_path} does not exist")
        return {}
    with open(metadata_path) as file:
        return s3_metadata_from_dict(yaml.safe_load(file) or {})


def generate_metadata_path(path):
    return os.path.splitext(path)[0] + ".yml"


def upload_file_with_metadata(
    client,
    bucket: str,
    key: str,
    path: str,
    metadata_path: str = None,
    printer=LOGGER.debug,
):
    """Uploads a file to S3, attaching the contents of its metadata file (if present) as
    S3 user-defined metadata on the object.

    The metadata is read from the YAML file on disk; it is not uploaded as a separate S3
    object. If the metadata file cannot be read for any reason the object is still
    uploaded, just without the extra metadata.

    Args:
        bucket: the s3 bucket that the file should be uploaded to
        key: the s3 key that the file should be uploaded to
        path: the location of the file on disk
        metadata_path: the location of the metadata file on disk. Defaults to the data
            file with a ".yml" extension.
    """
    metadata_path = metadata_path or generate_metadata_path(path)

    try:
        s3_metadata = s3_metadata_from_file(metadata_path)
    except Exception:
        LOGGER.warning(f"Could not read metadata from {metadata_path}", exc_info=True)
        s3_metadata = {}

    extra_args = {"Metadata": s3_metadata} if s3_metadata else {}
    client.upload_file(path, Bucket=bucket, Key=key, ExtraArgs=extra_args)
    printer(f"Uploaded {path} to s3://{bucket}/{key}")


def write_file_for_upload(
    key: str,
    do_write,
    script_path: str = None,
    path: str = None,
    metadata_path: str = None,
    bucket: str = settings.bucket,
    do_upload: bool = None,
    skip_metadata: bool = False,
):
    if do_upload is None:
        do_upload = "--upload" in sys.argv

    path = path or os.path.join(gettempdir(), posixpath.basename(key))
    script_path = script_path or inspect.stack()[2].filename
    metadata_path = metadata_path or generate_metadata_path(path)

    do_write(path)
    print(f"Written {path}")

    # Generate the metadata YAML file on disk. This is best-effort: if anything goes
    # wrong we warn but continue, so that metadata problems never prevent an upload.
    if not skip_metadata:
        try:
            metadata_object = generate_metadata(script_path)
            write_metadata(metadata_object, metadata_path)
            print(f"Written {metadata_path}")
        except Exception:
            LOGGER.warning("Failed to generate metadata", exc_info=True)

    if do_upload:
        upload_file_with_metadata(
            client=aws_session().client("s3"),
            bucket=bucket,
            key=key,
            path=path,
            metadata_path=None if skip_metadata else metadata_path,
            printer=print,
        )
    else:
        # print instructions to the user on how they can do the upload themselves

        # we need to single-quote file paths: not only in case there are spaces in the
        # filenames, but in particular so that users on Windows systems who are using
        # Bash shells do not get backslashes interpreted as escape characters
        def single_quote(text):
            return f"'{text}'"

        # build the command string. if bucket/metadata are default then omit them from
        # the command, just so it's not so long and scary.
        command = ["trase", "s3", "upload"]
        if bucket != settings.bucket:
            command.extend(["--bucket", bucket])
        if metadata_path != generate_metadata_path(path):
            command.extend(["--metadata", single_quote(metadata_path)])
        command.extend([single_quote(path), single_quote(key)])
        c = " ".join(command)

        # print message to the user
        print(
            f"\nThe file has been {colored('written but not uploaded', attrs=['bold'])}"
            f". Once you have inspected the file you can either:\n\n"
            f"  1) Re-run the script with --upload, or\n\n"
            f"  2) Use the Trase CLI:\n\n       {c}\n"
        )


def write_csv_for_upload(
    df: "pd.DataFrame",
    key: str,
    script_path: str = None,
    path: str = None,
    metadata_path: str = None,
    bucket: str = settings.bucket,
    do_upload: bool = None,
    skip_metadata: bool = False,
    **pd_to_csv_kwargs,
):
    """Writes a CSV file and either indicates to the user how they should upload it to,
    S3 or actually performs the upload itself.

    It is a very intentional design that, by default, the upload to S3 does not occur.
    We want the user to _explicitly choose_ to do the upload to minimise the chance that
    they accidentally overwrite data on S3. By putting a "human in the loop" we also
    give the user a chance to inspect the data before upload.

    Args:
        df: the Pandas dataframe that should be serialized
        key: the s3 key that the dataframe should be uploaded to
        script_path (optional): the path of the script that generated the file. Often
            this can be taken from `__file__`. If it is not supplied then we will
            inspect the call stack and take the filename of the file which called this
            function, which may or may not be what was intended.
        path (optional): the (temporary) file location where the Pandas dataframe will
            be written to before uploading. The reason it is written to disk is to give
            the user the chance to inspect its contents before uploading. If not
            provided then a path will be generated in the operating system's temporary
            directory.
        metadata_path (optional): the (temporary) file location where the metadata will
            be written to before uploading. If not provided then it will be written next
            to the data file with a ".yml" extension.
        bucket (optional): the s3 bucket to upload the file and metadata to.
            Defaults to bucket defined in `trase.config.settings`.
        do_upload (optional): whether to actually do the upload or just print to the
            user that they should do it themselves. If not provided then we will look
            for the presence of "--upload" in `sys.argv` to determine its value.
        skip_metadata (optional): if true, no metadata file is generated and no S3
            metadata is attached to the object.
        pd_to_csv_kwargs: keyword arguments that will be passed to pd.to_csv
    """

    def do_write(path):
        df.to_csv(
            path, **{"sep": ";", "encoding": "utf8", "index": False, **pd_to_csv_kwargs}
        )

    write_file_for_upload(
        key=key,
        do_write=do_write,
        script_path=script_path,
        path=path,
        metadata_path=metadata_path,
        bucket=bucket,
        do_upload=do_upload,
        skip_metadata=skip_metadata,
    )


def write_json_for_upload(
    data,
    key: str,
    script_path: str = None,
    path: str = None,
    metadata_path: str = None,
    bucket: str = settings.bucket,
    do_upload: bool = None,
    skip_metadata: bool = False,
    **json_dump_kwargs,
):
    def do_write(path):
        with open(path, "w") as file:
            json.dump(data, file, **json_dump_kwargs)

    write_file_for_upload(
        key=key,
        do_write=do_write,
        script_path=script_path,
        path=path,
        metadata_path=metadata_path,
        bucket=bucket,
        do_upload=do_upload,
        skip_metadata=skip_metadata,
    )


def write_parquet_for_upload(
    df: "pd.DataFrame",
    key: str,
    is_polars=False,
    script_path: str = None,
    path: str = None,
    metadata_path: str = None,
    bucket: str = settings.bucket,
    do_upload: bool = None,
    skip_metadata: bool = False,
    **to_parquet_kwargs,
):
    def do_write(path):
        if is_polars:
            df.write_parquet(path, **to_parquet_kwargs)
        else:
            df.to_parquet(path, **to_parquet_kwargs)

    write_file_for_upload(
        key=key,
        do_write=do_write,
        script_path=script_path,
        path=path,
        metadata_path=metadata_path,
        bucket=bucket,
        do_upload=do_upload,
        skip_metadata=skip_metadata,
    )


def write_geopandas_for_upload(
    gdf: "gpd.GeoDataFrame",
    key: str,
    script_path: str = None,
    path: str = None,
    metadata_path: str = None,
    bucket: str = settings.bucket,
    do_upload: bool = None,
    skip_metadata: bool = False,
    driver: str = "geojson",  # "parquet" | "geojson" | "gpkg"
    **to_file_kwargs,
):
    """
    Writes a GeoDataFrame to disk and optionally uploads to S3 with metadata.

    Supported drivers:
      - "parquet"  -> GeoParquet
      - "geojson"  -> GeoJSON
      - "gpkg"     -> GeoPackage
    """

    import geopandas as gpd
    import os

    def do_write(path):
        if driver == "parquet":
            gdf.to_parquet(path, **to_file_kwargs)
        elif driver == "geojson":
            gdf.to_file(path, driver="GeoJSON", **to_file_kwargs)
        elif driver == "gpkg":
            gdf.to_file(path, driver="GPKG", **to_file_kwargs)
        else:
            raise ValueError(f"Unsupported driver: {driver}")

    # default extension based on driver
    if path is None:
        ext = {
            "parquet": ".parquet",
            "geojson": ".geojson",
            "gpkg": ".gpkg",
        }.get(driver, ".dat")

        path = os.path.join(gettempdir(), posixpath.basename(key) + ext)

    write_file_for_upload(
        key=key,
        do_write=do_write,
        script_path=script_path,
        path=path,
        metadata_path=metadata_path,
        bucket=bucket,
        do_upload=do_upload,
        skip_metadata=skip_metadata,
    )

import pandas as pd


def model(dbt, cursor):
    dbt.source("trase-storage-raw", "boundaries_originals_nxparroquias")

    raise NotImplementedError()
    return pd.DataFrame({"hello": ["world"]})