Direct API to Workflows

Welcome to the final lesson of this course on building serverless applications! So far, you have learned how to use Google Cloud Functions, API Gateway, and Workflows to create and deploy serverless workflows. In this lesson, we will take your skills a step further by showing you how to use API Gateway to start Workflows directly — without needing an extra Cloud Function just to trigger a workflow.

This approach can make your applications faster and more cost-effective, since you avoid the extra step and cost of running a function just to start a workflow. You will learn how to use Python code to control the request and response, validate input, handle errors, and even route to different workflows — all within your serverless application.

By the end of this lesson, you will be able to build a robust API that starts Workflow executions with dynamic input and strong validation, all using Python. In the practice exercises, we'll use an emulated API Gateway environment to focus on the Python logic without requiring full GCP deployment.

When starting a Workflow execution, you often want each execution to have a unique name. This helps you track and manage each workflow run.

In Python, you can generate a unique name using the uuid module. For example, you can combine a prefix like "order-" with a random string to create a unique execution name.

Here's how you can do this in a Cloud Function or as part of your workflow logic:

import uuid

def generate_execution_name(prefix="order"):
    unique_id = uuid.uuid4().hex[:8]  # 8-character unique string
    return f"{prefix}-{unique_id}"

# Example usage:
execution_name = generate_execution_name()
print(execution_name)  # e.g., "order-7f3b2c1a"

Explanation:

• uuid.uuid4().hex[:8] generates a random 8-character string.
• The function combines this with a prefix to create a unique execution name.

This ensures every execution has a unique name, which is important for tracking and debugging.

It's important to make sure the incoming request has all the required fields before starting a workflow. For example, you might require a customerId in the request body.

You can perform this validation in Python code before starting the workflow. Here's how you might do it in a Cloud Function that receives a request from API Gateway:

def validate_input(request):
    request_json = request.get_json()
    
    if not request_json or not request_json.get("customerId"):
        return {"error": "Missing required field: customerId"}, 400
    
    return request_json, 200

# Example usage in a Cloud Function:
def start_workflow(request):
    data, status = validate_input(request)
    if status != 200:
        return data, status
    # Proceed to start the workflow...

Explanation:

• The function checks if customerId is present and not empty.
• If missing, it returns a 400 error with a helpful message.
• If valid, it returns the data for further processing.

This helps prevent invalid requests from starting workflows, saving resources and making your API more reliable.

When a Workflow execution starts, the response from the API may not always be user-friendly. You can use Python code to transform the response and add helpful information before returning it to the client.

For example, you might want to include:

• The execution name
• A timestamp for when the execution started
• A status field

Here's how you can do this in Python:

import datetime

def transform_response(execution_name, execution_id):
    response = {
        "execution_id": execution_id,
        "execution_name": execution_name,
        "started_at": datetime.datetime.utcnow().isoformat() + "Z",
        "status": "RUNNING"
    }
    return response, 202

# Example usage:
# return transform_response(execution_name, execution_id)

Explanation:

• The function builds a dictionary response with the execution details.
• The timestamp is in ISO 8601 format.
• The status is set to "RUNNING" for clarity.

This makes your API responses easier to understand and use.

Sometimes, starting a Workflow execution can fail. For example, the workflow might not exist, or you might try to use a duplicate execution name.

You can handle these errors in your Python code by returning appropriate HTTP status codes and messages.

Here's how you can set up error handling in your Cloud Function:

def start_workflow(request):
    try:
        # ... input validation and workflow start logic ...
        # For example, call the Workflows API here
        execution_name = generate_execution_name()
        # Simulate starting workflow and getting an execution_id
        execution_id = "projects/demo/locations/us-central1/workflows/OrderProcessing/executions/" + execution_name
        return transform_response(execution_name, execution_id)
    except ValueError as e:
        return {"error": "Invalid request parameters"}, 400
    except Exception as e:
        return {
            "error": "Internal server error",
            "message": str(e)
        }, 500

Explanation:

• If the input is invalid, return a 400 error.
• For unexpected errors, return a 500 error with the error message.

Example Output (for a duplicate execution name):

{
  "error": "Invalid request parameters"
}

Example Output (for an internal error):

{
  "error": "Internal server error",
  "message": "ExecutionAlreadyExists: An execution with the same name already exists."
}

This makes your API more robust and user-friendly by providing clear error messages.

Sometimes, you want your API to start different workflows based on a parameter in the request. For example, you might have different workflows for orders, refunds, or inventory updates.

You can read a workflow_type parameter from the request body, validate it, and use it to customize the workflow input and execution name.

Here's how you can do this in Python:

def start_workflow(request):
    request_json = request.get_json()
    
    if not request_json or not request_json.get("customerId"):
        return {"error": "Missing required field: customerId"}, 400

    workflow_type = request_json.get("workflow_type", "order")
    if workflow_type not in ["order", "refund", "inventory"]:
        workflow_type = "order"

    execution_name = generate_execution_name(prefix=workflow_type)
    workflow_input = {
        "order": request_json,
        "workflow_type": workflow_type
    }

    # Simulate starting the workflow and getting an execution_id
    execution_id = f"projects/demo/locations/us-central1/workflows/OrderProcessing/executions/{execution_name}"

    return transform_response(execution_name, execution_id)

Explanation:

• Sets a default workflow_type to "order".
• If the request includes a valid workflow_type (like "refund" or "inventory"), it uses that instead.
• The execution name and input are customized based on the workflow type.
• It still validates that customerId is present.

Example Output (for a refund workflow):

{
  "execution_id": "projects/demo/locations/us-central1/workflows/OrderProcessing/executions/refund-8a7b6c5d",
  "execution_name": "refund-8a7b6c5d",
  "started_at": "2024-06-01T12:34:56Z",
  "status": "RUNNING"
}

This lets you use a single API endpoint to start different workflows, making your API flexible and powerful.

In this lesson, you learned how to build a Python API that starts Workflows, implementing validation, error handling, and dynamic routing. You saw how to:

• Generate unique execution names using Python
• Validate input and return helpful error messages
• Transform Workflow responses to be more user-friendly
• Handle errors gracefully with custom status codes and messages
• Route requests dynamically based on workflow type

You have now reached the end of this course on building serverless applications with Google Cloud. Congratulations on making it this far! You are now ready to put these concepts into practice with the hands-on exercises that follow. Great work, and keep building!