Configure the Dagster Project

src/dagster_teradata_azure/definitions.py

import os

from dagster import DagsterError, Definitions, job, op
from dagster_azure.adls2 import ADLS2Resource, ADLS2SASToken
from dagster_teradata import TeradataResource

azure_resource = ADLS2Resource(
    storage_account="",
    credential=ADLS2SASToken(token=""),
)

td_resource = TeradataResource(
    host=os.getenv("TERADATA_HOST"),
    user=os.getenv("TERADATA_USER"),
    password=os.getenv("TERADATA_PASSWORD"),
    database=os.getenv("TERADATA_DATABASE"),
)

@op(required_resource_keys={"teradata"})
def drop_existing_table(context) -> str:
    try:
        context.resources.teradata.drop_table("people")
        context.log.info("Table 'people' dropped successfully")
        return "Tables Dropped"
    except Exception as error:
        context.log.error(f"Failed to drop table: {error}")
        raise

@op(required_resource_keys={"teradata", "azure"})
def ingest_azure_to_teradata(context, status: str) -> str:
    try:
        if status != "Tables Dropped":
            raise DagsterError("Table was not dropped")

        azure_blob_location = (
            "/az/akiaxox5jikeotfww4ul.blob.core.windows.net/"
            "td-usgs/CSVDATA/09380000/2018/06/"
        )

        context.log.info(
            f"Using Azure Blob Storage location: {azure_blob_location}"
        )

        context.resources.teradata.azure_blob_to_teradata(
            context.resources.azure,
            azure_blob_location,
            "people",
            public_bucket=True,
        )

        context.log.info(
            "Data ingested successfully from Azure Blob Storage to Teradata"
        )
        return "Data Ingested"
    except Exception as error:
        context.log.error(f"Failed to ingest data: {error}")
        raise

@job
def example_job():
    ingest_azure_to_teradata(drop_existing_table())

defs = Definitions(
    jobs=[example_job],
    resources={
        "teradata": td_resource,
        "azure": azure_resource,
    },
)