deployments.py

def deploy_factory(name:str, flow, param: dict = dict(), cron: str = '') -> Deployment:

    kwargs = {
        'flow': flow,
        'name': name,
        'infrastructure': CloudRunJob.load(f"aether-flows-cloud-run-job-{ENV}"),
        'work_queue_name': f"{ENV}_aether",
        'version': VERSION,
        'output': f'{name}.yaml',
        'skip_upload': True,
    }

    if param:
        kwargs['parameters'] = param,
    if cron:
        kwargs['schedule'] = CronSchedule(cron=cron)

    return Deployment.build_from_flow(**kwargs)


aether_space_metrics_1h_parquet_rsync = deploy_factory(
    name=f'aether_space_metrics_1h_parquet_rsync_{ENV}',
    flow=space_metrics_parquet_rsync,
    param={'date': 'current', 'agg_window': '1h'},
    cron=sched('0 23 * * *')
)
aether_space_metrics_1h_parquet_rsync.apply(work_queue_concurrency=20)