Step 2: Implement AWS S3 to Teradata Transfer in Dagster

import os

from dagster import job, op, Definitions, DagsterError
from dagster_aws.s3 import S3Resource
from dagster_teradata import TeradataResource

# Configure S3 resource for public bucket access (no credentials needed)
s3_resource = S3Resource()

# Configure Teradata resource with connection details from environment variables
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 e:
        context.log.error(f"Failed to drop table: {e}")
        raise

@op(required_resource_keys={"teradata", "s3"})
def ingest_s3_to_teradata(context, status: str) -> str:
    try:
        if status == "Tables Dropped":
            # Validate that AWS_S3_LOCATION is set
            s3_location = os.getenv("AWS_S3_LOCATION")
            if not s3_location:
                raise DagsterError(
                    "AWS_S3_LOCATION environment variable is not set. "
                    "Expected format: /s3/your-bucket.s3.amazonaws.com/path/to/file.csv"
                )

            context.log.info(f"Using S3 location: {s3_location}")

            # For public S3 buckets, create an S3Resource without credentials
            s3_resource_from_context = context.resources.s3
            context.resources.teradata.s3_to_teradata(
                s3_resource_from_context, 
                s3_location, 
                "people",
                public_bucket=True  # This example uses a public bucket for simplicity
            )
            context.log.info("Data ingested successfully from S3 to Teradata")
            return "Data Ingested"
        else:
            raise DagsterError("Tables not dropped")
    except Exception as e:
        context.log.error(f"Failed to ingest data: {e}")
        raise

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

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