DBT typer app

dbt.py

import typer
import os
import glob
from cli.apps.libs.dbt_contracts.generate_dbt_model import generate_model_yaml
from typing_extensions import Annotated
from typing import Tuple, List, Text
from pathlib import Path

CONTRACT_DIR = f"yourpath/data_contracts/users"

app = typer.Typer(help="Infinite Lambda CLI DBT module")


def get_files_in_subfolders(directory: Text) -> Tuple[List[Text], int]:
    """
    Retrieve a list of files in subfolders of the specified directory.

    :param directory: The directory path to search for files in subfolders.
    :return: List of file paths and the count of files.
    """

    subfolders = [f.path for f in os.scandir(directory) if f.is_dir()]
    contract_files = []

    for subfolder in subfolders:
        files = glob.glob(os.path.join(subfolder, "*"))
        for file in files:
            contract_files.append(file)

    return contract_files, len(contract_files)


def ask_user_for_contract_to_use(number_of_contract_files: int) -> int:
    """
    Prompt the user to choose a data contract among the available options.

    :param number_of_contract_files: The number of available data contract files.
    :return: The index of the chosen data contract.
    """

    while True:
        try:
            contract_to_use = int(typer.prompt("Please choose a data contract"))
            if 1 <= contract_to_use <= number_of_contract_files:
                return contract_to_use - 1
            else:
                typer.echo("Not a valid option!")
        except ValueError:
            print("Sorry, I didn't understand that. Maybe it's not a number?")
            continue


@app.command("model_yaml_from_data_contract")
def create_dbt_model(
    model_name: Annotated[
        Text,
        typer.Option(
            prompt="Please provide dbt model name",
            help="Name of the model to be generated",
        ),
    ],
    result_path = "/yourpath/wheredbtmodel/willbesaved"
):
    """
    Create a dbt model yaml using the provided model name and a selected data contract.
    """

    latest_files, number_of_files = get_files_in_subfolders(CONTRACT_DIR)
    for t, i in enumerate(latest_files, start=1):
        typer.echo(f"{t} --- {i}")
    contract_to_use = ask_user_for_contract_to_use(number_of_files)
    data_contract_path = latest_files[contract_to_use]

    # Generate model.yaml
    generate_model_yaml(
        data_contract_path=data_contract_path,
        model_name=model_name,
        result_path=result_path,
    )


if __name__ == "__main__":
    app()