def generate_queries()

in sql_generators/experiment_monitoring/__init__.py [0:0]

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

def generate_queries(project, path, write_dir):
    """Generate experiment monitoring views."""
    with open(Path(path) / "templating.yaml", "r") as f:
        template_config = yaml.safe_load(f) or {}

    for query, args in template_config["queries"].items():
        template_query_dir = FILE_PATH / "templates" / query
        env = Environment(
            loader=FileSystemLoader(template_query_dir),
            keep_trailing_newline=True,
        )
        sql_templates = list(template_query_dir.glob("*.sql"))
        sql_template_file = sql_templates[0].name
        sql_template = env.get_template(sql_template_file)
        metadata_template = env.get_template("metadata.yaml")

        args["destination_table"] = query
        args["search_metrics"] = template_config["search_metrics"]

        if args["per_app"]:
            # generate a separate query for each application dataset
            for dataset in template_config["applications"]:
                if "skip_applications" in args:
                    if any(fnmatchcase(dataset, skip_app) for skip_app in args["skip_applications"]):
                        continue

                args["dataset"] = dataset

                write_sql(
                    write_dir / project,
                    f"{project}.{dataset}_derived.{query}",
                    sql_template_file,
                    reformat(sql_template.render(**args)),
                )

                write_path = Path(write_dir) / project / (dataset + "_derived") / query
                (write_path / "metadata.yaml").write_text(
                    metadata_template.render(**args)
                )
        else:
            # generate a single query that UNIONs application datasets
            # these queries are written to `telemetry`
            args["applications"] = template_config["applications"]

            if "skip_applications" in args:
                args["applications"] = [
                    app
                    for app in args["applications"]
                    if not any(fnmatchcase(app, skip_app) for skip_app in args["skip_applications"])
                ]

            destination_dataset = args.get("destination_dataset", "telemetry_derived")

            write_sql(
                write_dir / project,
                f"{project}.{destination_dataset}.{query}",
                sql_template_file,
                reformat(sql_template.render(**args)),
            )

            write_path = Path(write_dir) / project / destination_dataset / query
            (write_path / "metadata.yaml").write_text(metadata_template.render(**args))