def dry_run_query()

in jetstream/dryrun.py [0:0]

• 27 lines of code
• 13 McCabe index (conditional complexity)

def dry_run_query(sql: str) -> None:
    """Dry run the provided SQL query."""
    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:
        # 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) from e

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

    error = response["errors"][0] if "errors" in response and len(response["errors"]) == 1 else 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)