in opmon/cli.py [0:0]
def _before_execute_callback(sql_output_dir: Optional[str], query, job_config, annotations={}):
"""Maybe write SQL query to disk.
If `annotations` contain all of `slug`, `submission_date`, and
`type`, write `query` to given `sql_output_dir`.
"""
if not sql_output_dir:
return
# Some annotations are required to write output.
if "slug" not in annotations:
return
if "submission_date" not in annotations:
return
if "type" not in annotations:
return
# We could avoid some IO by remembering what we've already created, but for
# now this will do.
Path(sql_output_dir).mkdir(parents=True, exist_ok=True)
# The submission date is actually a datetime.
submission_date = annotations["submission_date"].strftime("%Y-%m-%d")
part = ""
if "part" in annotations:
part = f"-{annotations['part']}"
fname = f"{annotations['slug']}-{annotations['type']}-{submission_date}{part}.sql"
(Path(sql_output_dir) / fname).write_text(query)