def compile_workflow()

in functions/data-processing-engines/dataform-tag-executor/main.py [0:0]

• 22 lines of code
• 1 McCabe index (conditional complexity)

def compile_workflow(gcp_project: str, repo_name: str, repo_uri: str, branch: str, query_variables: dict):
    """Compiles a Dataform workflow using a specified Git branch.

    Returns:
        str: The name of the created compilation result.
    """

    github_token = access_secret_version(gcp_project, repo_name + "_secret")

    dataform_json_content = get_dataform_json_from_github(
        df_client
            .get_repository(name=repo_uri)
            .git_remote_settings.url
            .replace(".git", ""),
        github_token)

    compilation_result = dataform_v1beta1.CompilationResult(
        git_commitish=branch,
    )

    merge_compilation_config(compilation_result.code_compilation_config, query_variables, dataform_json_content)

    print("compilation_result.code_compilation_config.vars::::::   " + str(
        compilation_result.code_compilation_config.vars))

    request = dataform_v1beta1.CreateCompilationResultRequest(
        parent=repo_uri,
        compilation_result=compilation_result
    )

    response = df_client.create_compilation_result(request=request)
    name = response.name
    logging.info(f'compiled workflow {name}')
    return name