def main()

in probe_scraper/glean_push.py [0:0]
62 lines of code
12 McCabe index (conditional complexity)

def main(request: Request) -> Response:
    """Scrape probes from a single glean commit."""
    output_bucket = os.environ.get("OUTPUT_BUCKET", None)
    if output_bucket is None:
        return Response("Cloud function has no configured output bucket\n", 500)

    args = request.get_json(force=True)
    if not isinstance(args, dict):
        return Response(f"request body must be a JSON object but got: {args}\n", 400)
    try:
        url = args["url"]
        commit = args["commit"]
        branch = args["branch"]
    except KeyError as e:
        return Response(f"request JSON missing key: {e}\n", 400)

    if not isinstance(url, str):
        return Response("Error: url must be a string\n", 400)
    if not isinstance(commit, str):
        return Response("Error: commit must be a string\n", 400)
    if not isinstance(branch, str):
        return Response("Error: branch must be a string\n", 400)

    with tempfile.TemporaryDirectory() as tmpdirname:
        tmp = Path(tmpdirname)
        out_dir = tmp / "output"
        cache_dir = tmp / "cache"
        email_file = tmp / "emails.txt"
        out_dir.mkdir()
        cache_dir.mkdir()
        try:
            updated_paths = runner.main(
                cache_dir=cache_dir,
                out_dir=out_dir,
                firefox_version=None,
                min_firefox_version=None,
                process_moz_central_probes=False,
                process_glean_metrics=True,
                repositories_file=Path("repositories.yaml"),
                dry_run=True,
                glean_repos=None,
                firefox_channel=None,
                output_bucket=output_bucket,
                cache_bucket=None,
                env="prod",
                bugzilla_api_key=None,
                glean_urls=[url],
                glean_commit=commit,
                glean_commit_branch=branch,
                update=True,
                email_file=email_file,
            )
        except ProbeScraperError as e:
            return Response(f"Error: {e.message}\n", e.status_code)
        if updated_paths:
            updates = ", ".join(str(p.relative_to(out_dir)) for p in updated_paths)
            message = f"update published for {updates}\n"
        else:
            message = "update is valid, but not published\n"
        try:
            emails = email_file.read_text()
        except FileNotFoundError:
            pass  # no emails means no warnings or expiring metrics, which is good
        else:
            message += f"additional messages: {emails}\n"
        return Response(message, 200)