def _before_execute_callback()

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)