Tom Reilly
07/11/2022, 2:26 PMDEFAULT_OP_PROPS = dict(
required_resource_keys={"dbt"},
ins={"start_after": In(Nothing)},
out=Out(DbtOutput, description="Parsed output from running the dbt command."),
tags={"kind": "dbt"},
config_schema={
"select": str,
"exclude": list
}
)
@op(**DEFAULT_OP_PROPS)
def dbt_build(context: OpExecutionContext):
"""Execute the dbt build command so run and test occur together"""
context.resources.dbt.cli(command="deps")
dbt_result = context.resources.dbt.cli(command="build", select=context.op_config['select'], exclude=context.op_config['exclude'])
for materialization in generate_materializations(dbt_result):
context.log_event(materialization)
return dbt_result
yuhan
07/11/2022, 6:22 PMowen
07/11/2022, 6:28 PMselect
and exclude
parameters (which will apply to every dbt command that is executed with that resource). this will work as long as you don't have jobs that contain ops with different select/exclude requirements. If you do, then your solution of creating your own custom op is the recommend approach