Jake Kagan
08/28/2023, 12:26 PMJake Kagan
08/28/2023, 2:17 PMsandy
08/28/2023, 3:09 PMDefinitions
object iterate over the values in the database table?Jake Kagan
08/28/2023, 10:49 PMfrom dagster import op, configured, get_dagster_logger, job, Definitions
## -- NAMER
def name_job(job_name: str):
def decorator(func):
func.__name__ = job_name
return func
return decorator
## -- OP SETUP
@op(config_schema={"passed_arg": str})
def op_names(context):
arg = context.op_config['passed_arg']
get_dagster_logger().info(arg)
def op_configger(arg):
configged_op = configured(op_names, name=f'{arg}_op')(
{"passed_arg": arg}
)
return configged_op
## -- JOB FACTORY
def job_configged(name, job_ops):
@job()
@name_job(name)
def build():
job_ops()
return build
def job_factory(names):
final_jobs = []
for name in names:
job_ops = op_configger(name)
final_jobs.append(job_configged(name, job_ops))
return final_jobs
## -- DEFINITIONS
names = ["job1", "job2", "job3", "job4"]
db_jobs= job_factory(names)
jobs = Definitions(jobs=db_jobs)
Jake Kagan
08/29/2023, 12:38 AM