def handler()

in src/workflow-trigger/main.py [0:0]

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

def handler(event, context):
    """Triggered by a change to a Cloud Storage bucket.
    Args:
        event (dict): Event payload.
            context (google.cloud.functions.Context): Metadata for the event.
    """
    arguments = {"bucket": event['bucket'], "file": event['name']}
    print(
        f"Received file: {arguments['file']}, from bucket: {arguments['bucket']}"
    )

    print(json.dumps(arguments))

    # trigger the workflow
    wf_exec = wf_execution_client.create_execution(request={
        "parent": WORKFLOW_ID,
        "execution": {
            "argument": json.dumps(arguments)
        }
    })

    print(f"Execution created: {wf_exec.name}.")