DBT: Brazil Cd Concat Beef 2018
File location: s3://trase-storage/brazil/trade/cd/concat/BRAZIL_CD_CONCAT_BEEF_2018.csv
DBT model name: brazil_cd_concat_beef_2018
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 |
|---|---|---|
country_of_destination.label |
VARCHAR |
|
country_of_destination.name |
VARCHAR |
|
country_of_destination.trase_id |
VARCHAR |
|
exporter.cnpj |
VARCHAR |
|
exporter.cnpj.valid |
VARCHAR |
|
exporter.label |
VARCHAR |
|
exporter.municipality.name |
VARCHAR |
|
exporter.municipality.trase_id |
VARCHAR |
|
importer.label |
VARCHAR |
|
fob |
VARCHAR |
|
hs4 |
VARCHAR |
|
hs6 |
VARCHAR |
|
hs8 |
VARCHAR |
|
month |
VARCHAR |
|
source |
VARCHAR |
|
via |
VARCHAR |
|
via.name |
VARCHAR |
|
port_of_export.name |
VARCHAR |
|
vol |
VARCHAR |
|
year |
VARCHAR |
No data tests defined 🧐
Models
"""
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.ref("brazil_beef_datamyne_cd_2018")
dbt.ref("cd_dashboard_brazil_2018")
raise NotImplementedError()
return pd.DataFrame({"hello": ["world"]})
-
Dbt path:
memory.main.brazil_cd_concat_beef_2018 -
Containing yaml link: trase/data_pipeline/models/brazil/trade/cd/concat/_schema.yml
-
Model file: trase/data_pipeline/models/brazil/trade/cd/concat/brazil_cd_concat_beef_2018.py
-
Calls script:
trase/tools/aws/metadata.py -
Tags:
mock_model,brazil,cd,concat,trade