def create_job()

in functions/orchestration-helpers/scheduling/main.py [0:0]

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

def create_job(job_name, crond_expression, time_zone, workflow_parameters):
    """
    creates a scheduler job , using given parameters.

    Args:
        job_name: name for the scheduler job, should be the same as cloud workflows name
        crond_expression:  crond linux expression used to trigger the cloud scheduler rule
        time_zone: timezone associated with scheduler execution.
        workflow_parameters: parameters sent to the cloud workflows invocation

    """
    parent= scheduler_client.common_location_path(WORKFLOW_SCHEDULING_PROJECT_ID,WORKFLOW_SCHEDULING_PROJECT_REGION)
    job={
        "name":"projects/"+ WORKFLOW_SCHEDULING_PROJECT_ID+ "/locations/"+WORKFLOW_SCHEDULING_PROJECT_REGION+"/jobs/" + job_name,
        "description":"workflows scheduler job create",
        "http_target": {
            "http_method": "POST",
            "uri": f"https://{WORKFLOW_SCHEDULING_PROJECT_REGION}-{WORKFLOW_SCHEDULING_PROJECT_ID}.cloudfunctions.net/{PIPELINE_EXECUTION_FUNCTION_NAME}" ,
            "headers": {"Content-Type": "application/json"},
            "oidc_token": {"service_account_email": WORKFLOW_SCHEDULING_PROJECT_NUMBER + "-compute@developer.gserviceaccount.com"},
            "body": json.dumps(workflow_parameters).encode("utf-8"),
        },
        "schedule":crond_expression,
        "time_zone":time_zone,
    }
    scheduler_client.create_job(parent=parent,job=job)
    print("JOB CREATED...........")