def dry_run_query()

in opmon/dryrun.py [0:0]

• 34 lines of code
• 15 McCabe index (conditional complexity)

def dry_run_query(sql: Union[str, List[str]]) -> None:
    """Dry run the provided SQL query."""
    if isinstance(sql, list):
        for query in sql:
            dry_run_query(sql)
        return
    try:
        r = requests.post(
            DRY_RUN_URL,
            headers={"Content-Type": "application/json"},
            data=json.dumps({"dataset": "mozanalysis", "query": sql}).encode("utf8"),
        )
        response = r.json()
    except Exception as e:
        # This may be a JSONDecode exception or something else.
        # If we got a HTTP exception, that's probably the most interesting thing to raise.
        try:
            r.raise_for_status()
        except requests.exceptions.RequestException as request_exception:
            e = request_exception
        raise DryRunFailedError(e, sql)

    if response["valid"]:
        logger.info("Dry run OK")
        return

    if "errors" in response and len(response["errors"]) == 1:
        error = response["errors"][0]
    else:
        error = None

    if (
        error
        and error.get("code", None) in [400, 403]
        and "does not have bigquery.tables.create permission for dataset"
        in error.get("message", "")
    ):
        # We want the dryrun service to only have read permissions, so
        # we expect CREATE VIEW and CREATE TABLE to throw specific
        # exceptions.
        logger.info("Dry run OK")
        return

    raise DryRunFailedError((error and error.get("message", None)) or response["errors"], sql=sql)