Module `dynamicio.cli`

Implements the dynamicio Command Line Interface (CLI).

Expand source code

"""Implements the dynamicio Command Line Interface (CLI)."""
import argparse
import glob
import os
import pprint
from typing import Mapping, MutableMapping, Optional, Sequence

import pandas as pd  # type: ignore
import yaml

from dynamicio.errors import InvalidDatasetTypeError


def parse_args(args: Optional[Sequence] = None) -> argparse.Namespace:
    """Arguments parser for dynamicio cli.py.

    Args:
        args: List of args to be parsed. Defaults to None, in which case
            sys.argv[1:] is used.

    Returns:
        An instance of ArgumentParser populated with the provided args.
    """
    parser = argparse.ArgumentParser(prog="dynamicio", description="Generate dataset schemas")
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument(
        "-b",
        "--batch",
        action="store_true",
        help="flag, used to generate multiple schemas provided a datasets directory.",
    )
    group.add_argument(
        "-s",
        "--single",
        action="store_true",
        help="flag, used to generate a schema provided a single dataset.",
    )
    parser.add_argument("-p", "--path", required=True, help="the path to the dataset/datasets-directory.", type=str)
    parser.add_argument("-o", "--output", required=True, help="the path to the schemas output directory.", type=str)
    return parser.parse_args(args)


def generate_schema_for(dataset: str) -> Mapping:
    """Generate a schema for a dataset.

    Args:
        dataset: The path to the dataset for which we want to generate a schema

    Returns:
        A dictionary containing the schema for the dataset, or None if the dataset is not valid.

    Raises:
        InvalidDatasetTypeError: If the dataset type is not supported by dynamicio.
    """
    dataset_name, file_type = os.path.splitext(os.path.basename(dataset))

    if file_type == ".parquet":
        df = pd.read_parquet(dataset)
    elif file_type == ".csv":
        df = pd.read_csv(dataset)
    elif file_type == ".json":
        df = pd.read_json(dataset)
    elif file_type == ".h5":
        df = pd.read_hdf(dataset)
    else:
        raise InvalidDatasetTypeError(dataset)

    print(f"Generating schema for: {dataset}")
    json_schema: MutableMapping = {"name": dataset_name, "columns": {}}
    for column, d_type in zip(list(df.columns), list(df.dtypes)):
        json_schema["columns"][column] = {"type": "", "validations": {}, "metrics": []}
        json_schema["columns"][column]["type"] = d_type.name

    return json_schema


def main(args: argparse.Namespace):
    """Main function for dynamicio cli.py.

    Args:
        args: Parsed args.
    """
    if args.batch:
        dataset_files = glob.glob(os.path.join(args.path, "*.*"))
        for dataset in dataset_files:
            try:
                json_schema = generate_schema_for(dataset)
            except InvalidDatasetTypeError as exception:
                print(f"Skipping {exception.message}! You may want to remove this file from the datasets directory")
            else:
                with open(os.path.join(args.output, f"{json_schema['name']}.yaml"), "w") as file:  # pylint: disable=unspecified-encoding]
                    file.write("---\n")
                    yaml.safe_dump(json_schema, file)

    if args.single:
        json_schema = generate_schema_for(str(args.path))
        with open(os.path.join(args.output, f"{json_schema['name']}.yaml"), "w") as file:  # pylint: disable=unspecified-encoding]
            file.write("---\n")
            yaml.safe_dump(json_schema, file)
        pprint.pprint(json_schema)


def run():
    """Entry point for the dynamicio cli.py."""
    args = parse_args()
    main(args)

Functions

def generate_schema_for(dataset: str) ‑> Mapping

Generate a schema for a dataset.

Args

dataset: The path to the dataset for which we want to generate a schema

Returns

A dictionary containing the schema for the dataset, or None if the dataset is not valid.

Raises

InvalidDatasetTypeError: If the dataset type is not supported by dynamicio.

Expand source code

def generate_schema_for(dataset: str) -> Mapping:
    """Generate a schema for a dataset.

    Args:
        dataset: The path to the dataset for which we want to generate a schema

    Returns:
        A dictionary containing the schema for the dataset, or None if the dataset is not valid.

    Raises:
        InvalidDatasetTypeError: If the dataset type is not supported by dynamicio.
    """
    dataset_name, file_type = os.path.splitext(os.path.basename(dataset))

    if file_type == ".parquet":
        df = pd.read_parquet(dataset)
    elif file_type == ".csv":
        df = pd.read_csv(dataset)
    elif file_type == ".json":
        df = pd.read_json(dataset)
    elif file_type == ".h5":
        df = pd.read_hdf(dataset)
    else:
        raise InvalidDatasetTypeError(dataset)

    print(f"Generating schema for: {dataset}")
    json_schema: MutableMapping = {"name": dataset_name, "columns": {}}
    for column, d_type in zip(list(df.columns), list(df.dtypes)):
        json_schema["columns"][column] = {"type": "", "validations": {}, "metrics": []}
        json_schema["columns"][column]["type"] = d_type.name

    return json_schema

def main(args: argparse.Namespace)

Main function for dynamicio cli.py.

Args

args: Parsed args.

Expand source code

def main(args: argparse.Namespace):
    """Main function for dynamicio cli.py.

    Args:
        args: Parsed args.
    """
    if args.batch:
        dataset_files = glob.glob(os.path.join(args.path, "*.*"))
        for dataset in dataset_files:
            try:
                json_schema = generate_schema_for(dataset)
            except InvalidDatasetTypeError as exception:
                print(f"Skipping {exception.message}! You may want to remove this file from the datasets directory")
            else:
                with open(os.path.join(args.output, f"{json_schema['name']}.yaml"), "w") as file:  # pylint: disable=unspecified-encoding]
                    file.write("---\n")
                    yaml.safe_dump(json_schema, file)

    if args.single:
        json_schema = generate_schema_for(str(args.path))
        with open(os.path.join(args.output, f"{json_schema['name']}.yaml"), "w") as file:  # pylint: disable=unspecified-encoding]
            file.write("---\n")
            yaml.safe_dump(json_schema, file)
        pprint.pprint(json_schema)

def parse_args(args: Optional[Sequence] = None) ‑> argparse.Namespace

Arguments parser for dynamicio cli.py.

Args

args: List of args to be parsed. Defaults to None, in which case sys.argv[1:] is used.

Returns

An instance of ArgumentParser populated with the provided args.

Expand source code

def parse_args(args: Optional[Sequence] = None) -> argparse.Namespace:
    """Arguments parser for dynamicio cli.py.

    Args:
        args: List of args to be parsed. Defaults to None, in which case
            sys.argv[1:] is used.

    Returns:
        An instance of ArgumentParser populated with the provided args.
    """
    parser = argparse.ArgumentParser(prog="dynamicio", description="Generate dataset schemas")
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument(
        "-b",
        "--batch",
        action="store_true",
        help="flag, used to generate multiple schemas provided a datasets directory.",
    )
    group.add_argument(
        "-s",
        "--single",
        action="store_true",
        help="flag, used to generate a schema provided a single dataset.",
    )
    parser.add_argument("-p", "--path", required=True, help="the path to the dataset/datasets-directory.", type=str)
    parser.add_argument("-o", "--output", required=True, help="the path to the schemas output directory.", type=str)
    return parser.parse_args(args)

def run()

Entry point for the dynamicio cli.py.

Expand source code

def run():
    """Entry point for the dynamicio cli.py."""
    args = parse_args()
    main(args)