from dagster import ScheduleDefinition, job
from util_services.util_ops import preview_dataframe

from ..ops.ops import (fetch_data, generate_example_dataset, process_data,
                       transform_example_dataset)


@job
def data_processing_job():
    """A simple job that fetches and processes data."""
    raw = fetch_data()
    process_data(raw)

data_processing_job_daily_9am = ScheduleDefinition(
    job=data_processing_job,
    cron_schedule="0 9 * * *",
)

@job
def example_job():
    original_df = generate_example_dataset()
    preview_dataframe(original_df)

@job
def example_dataframe_demo_job():
    """Demonstrates generating and transforming a pandas DataFrame with previews."""
    original_df = generate_example_dataset()
    preview_dataframe(original_df)
    transformed_df = transform_example_dataset(original_df)
    preview_dataframe(transformed_df)