Robert Esteves
05/30/2024, 2:50 PMfrom helper_functions import calc_date
from prefect.deployments import Deployment
from prefect.server.schemas.schedules import CronSchedule
from datetime import date, timedelta
_yesterday_str = calc_date.fn()
deploy_flow = Deployment.build_from_flow(
flow=my_test_deployment,
name="Test",
storage=,
path="",
entrypoint=f"xxx/yyy",
schedule=CronSchedule(cron="0 10 * * *", timezone="America/New_York"),
work_queue_name="Hawk1",
tags=["ETL", 'JSON'],
description="This Prefect flow performs ingests, loads, and Transforms data for ...",
parameters=dict(calendar_date=f'{_yesterday_str}'
)
)
if __name__ == "__main__":
deploy_flow.apply()
Nate
05/30/2024, 8:52 PM@flow
def test_flow(..., lookback_days: int = 1):
_yesterday_str = datetime.now() - timedelta(days=lookback_days)
...
using pydantic to avoid this logic being in your flow
from datetime import datetime, timedelta
from prefect import flow
from pydantic import BaseModel, computed_field # comes with prefect
class DateInfo(BaseModel):
lookback_days: int = 1
@computed_field
@property
def yesterday(self) -> datetime:
return datetime.now() - timedelta(days=self.lookback_days)
@flow
def test_flow(date_info: DateInfo = DateInfo()):
assert date_info.yesterday == "the actual date string you want"
Robert Esteves
06/21/2024, 3:57 PMNate
06/21/2024, 3:58 PM