scheduled snapshot query creation

snapshot_query_definition.py

def _snapshot_query(
    snapshot_dataset: str,
    source_dataset: str,
    source_table: str,
    retention_days: int | None,
) -> str:
    """Contsruct the query for snapshotting a table.
    Taken from: https://cloud.google.com/bigquery/docs/table-snapshots-scheduled#schedule_the_monthly_query
    See the docstring for _create_scheduled_snapshot for details on parameters.
    """
    
    if retention_days is not None:
        expiration = f"DATE_ADD(current_timestamp(), INTERVAL {retention_days} DAY);"
        clone_opts = (
            r'" OPTIONS(expiration_timestamp = TIMESTAMP \"", expiration, "\")"'
        )
    else:
        expiration = '""'
        clone_opts = '""'
    

    return f"""
-- Declare variables
DECLARE snapshot_name STRING;
DECLARE expiration TIMESTAMP;
DECLARE query STRING;

-- Set variables
SET expiration = {expiration}
SET snapshot_name = CONCAT(
    "{snapshot_dataset}.{source_dataset}-{source_table}",
    FORMAT_DATETIME('%Y%m%d', current_date())
);

-- Construct the query to create the snapshot
SET query = CONCAT(
    "CREATE SNAPSHOT TABLE ", snapshot_name,
    " CLONE {source_dataset}.{source_table}", {clone_opts}, ";"
    ");"
);

-- Run the query
EXECUTE IMMEDIATE query;"""