Dagster graph without Dagstd

job_without_dagstd.py

import zipfile

from datetime import datetime

from dagster import op, job


@op
def get_todays_date() -> str:
    return datetime.today().strftime()


@op
def five() -> int:
    return 5


@op
def get_download_file_url(date: str) -> str:
    return f'https://example.com/{date}.csv'


@op
def get_nth_file_name(n: int) -> str:
    return f'file_{n:02}.txt'


@op
def extract_file_from_zip(context, zip_path: str, file_name: str) -> str:
    with zipfile.ZipFile(zip_path) as zip_file:
        with(f'/tmp/{file_name}', 'wb') as f:
            f.write(zip_file.read(file_name))
        context.log.info(f'Extracted {file_name} from {zip_path}')
        return f'/tmp/{file_name}'


@job
def process_data():
    date = get_todays_date()
    url = get_download_file_url(date)
    zip_path = download_large_file(url)

    file_name = get_nth_file_name(five())
    file_path = extract_file_from_zip(zip_path, file_name)