Application Development - GCP Professional Cloud Developer¶

# requirements.txt
flask==2.3.0
google-cloud-storage==2.10.0
google-cloud-secret-manager==2.16.0

# Use virtual environments
python -m venv venv
source venv/bin/activate
pip install -r requirements.txt

# config.py
import os

class Config:
    PROJECT_ID = os.environ.get('GCP_PROJECT_ID')
    DB_CONNECTION = os.environ.get('DATABASE_URL')
    SECRET_KEY = os.environ.get('SECRET_KEY')
    DEBUG = os.environ.get('DEBUG', 'False') == 'True'

    # For Cloud Run
    PORT = int(os.environ.get('PORT', 8080))

# database.py - treat database as attached resource
import os
import sqlalchemy

def get_db_connection():
    db_url = os.environ['DATABASE_URL']
    # Connection can be swapped by changing environment variable
    engine = sqlalchemy.create_engine(db_url)
    return engine

# Stateless application example
from flask import Flask, session
import redis

app = Flask(__name__)
# Don't store session in process memory
redis_client = redis.Redis(host=os.environ['REDIS_HOST'])

@app.route('/cart', methods=['POST'])
def add_to_cart():
    user_id = request.json['user_id']
    item = request.json['item']
    # Store in Redis, not in-memory
    redis_client.lpush(f'cart:{user_id}', item)
    return {'status': 'added'}

# app.py
if __name__ == '__main__':
    port = int(os.environ.get('PORT', 8080))
    app.run(host='0.0.0.0', port=port)

# graceful_shutdown.py
import signal
import sys

def signal_handler(sig, frame):
    print('Shutting down gracefully...')
    # Close database connections
    db.close()
    # Finish current requests
    server.shutdown()
    sys.exit(0)

signal.signal(signal.SIGTERM, signal_handler)

# logging_config.py
import logging
import json

# Write structured logs to stdout (Cloud Logging captures)
def setup_logging():
    logging.basicConfig(
        level=logging.INFO,
        format='%(message)s',
        handlers=[logging.StreamHandler()]
    )

def log_structured(message, severity='INFO', **kwargs):
    log_entry = {
        'severity': severity,
        'message': message,
        **kwargs
    }
    print(json.dumps(log_entry))

# Usage
log_structured('User logged in', user_id=123, ip='1.2.3.4')

# 1. Synchronous REST API
import requests

def call_user_service(user_id):
    response = requests.get(
        f'https://user-service.run.app/users/{user_id}',
        headers={'Authorization': f'Bearer {get_token()}'}
    )
    return response.json()

# 2. Asynchronous Pub/Sub
from google.cloud import pubsub_v1

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path('project-id', 'order-created')

def publish_order_event(order_data):
    message_json = json.dumps(order_data)
    future = publisher.publish(
        topic_path,
        message_json.encode('utf-8'),
        event_type='order.created'
    )
    return future.result()

# 3. gRPC for internal services
import grpc
import user_service_pb2
import user_service_pb2_grpc

def get_user_grpc(user_id):
    with grpc.insecure_channel('user-service:50051') as channel:
        stub = user_service_pb2_grpc.UserServiceStub(channel)
        response = stub.GetUser(
            user_service_pb2.GetUserRequest(user_id=user_id)
        )
        return response

# event_store.py
from google.cloud import firestore
from datetime import datetime

class EventStore:
    def __init__(self):
        self.db = firestore.Client()

    def append_event(self, aggregate_id, event_type, event_data):
        event = {
            'aggregate_id': aggregate_id,
            'event_type': event_type,
            'event_data': event_data,
            'timestamp': datetime.utcnow(),
            'version': self.get_next_version(aggregate_id)
        }
        self.db.collection('events').add(event)
        return event

    def get_events(self, aggregate_id):
        events = self.db.collection('events')\
            .where('aggregate_id', '==', aggregate_id)\
            .order_by('version')\
            .stream()
        return [e.to_dict() for e in events]

    def replay_events(self, aggregate_id, aggregate_class):
        events = self.get_events(aggregate_id)
        aggregate = aggregate_class()
        for event in events:
            aggregate.apply_event(event)
        return aggregate

# order_aggregate.py
class Order:
    def __init__(self):
        self.order_id = None
        self.items = []
        self.status = 'PENDING'

    def apply_event(self, event):
        if event['event_type'] == 'OrderCreated':
            self.order_id = event['event_data']['order_id']
        elif event['event_type'] == 'ItemAdded':
            self.items.append(event['event_data']['item'])
        elif event['event_type'] == 'OrderCompleted':
            self.status = 'COMPLETED'

# commands.py - Write model
class OrderCommandService:
    def __init__(self, event_store, publisher):
        self.event_store = event_store
        self.publisher = publisher

    def create_order(self, order_data):
        # Validate command
        if not order_data.get('customer_id'):
            raise ValueError('Customer ID required')

        # Create event
        event = self.event_store.append_event(
            aggregate_id=order_data['order_id'],
            event_type='OrderCreated',
            event_data=order_data
        )

        # Publish event for read model update
        self.publisher.publish(
            'order-events',
            json.dumps(event).encode()
        )

# queries.py - Read model
class OrderQueryService:
    def __init__(self):
        self.db = firestore.Client()

    def get_order(self, order_id):
        # Read from optimized read model
        doc = self.db.collection('orders_read_model')\
            .document(order_id).get()
        return doc.to_dict()

    def get_customer_orders(self, customer_id):
        # Denormalized for fast queries
        orders = self.db.collection('orders_read_model')\
            .where('customer_id', '==', customer_id)\
            .order_by('created_at', direction=firestore.Query.DESCENDING)\
            .limit(10).stream()
        return [o.to_dict() for o in orders]

# event_handler.py - Update read model
def handle_order_event(event, context):
    """Cloud Function to update read model"""
    event_data = json.loads(event['data'])

    if event_data['event_type'] == 'OrderCreated':
        # Update read model
        db = firestore.Client()
        db.collection('orders_read_model').document(
            event_data['aggregate_id']
        ).set({
            'order_id': event_data['aggregate_id'],
            'customer_id': event_data['event_data']['customer_id'],
            'items': event_data['event_data']['items'],
            'status': 'PENDING',
            'created_at': event_data['timestamp']
        })

# saga_orchestrator.py
from google.cloud import pubsub_v1
from enum import Enum

class SagaState(Enum):
    STARTED = 'STARTED'
    PAYMENT_PROCESSING = 'PAYMENT_PROCESSING'
    INVENTORY_RESERVED = 'INVENTORY_RESERVED'
    COMPLETED = 'COMPLETED'
    COMPENSATING = 'COMPENSATING'
    FAILED = 'FAILED'

class OrderSaga:
    def __init__(self):
        self.publisher = pubsub_v1.PublisherClient()
        self.db = firestore.Client()

    def start_order_saga(self, order_data):
        saga_id = order_data['order_id']

        # Save saga state
        self.save_saga_state(saga_id, SagaState.STARTED, order_data)

        # Step 1: Process payment
        self.publish_command('process-payment', {
            'saga_id': saga_id,
            'amount': order_data['total'],
            'customer_id': order_data['customer_id']
        })

    def handle_payment_success(self, event):
        saga_id = event['saga_id']
        saga = self.get_saga(saga_id)

        # Update state
        self.save_saga_state(saga_id, SagaState.PAYMENT_PROCESSING, event)

        # Step 2: Reserve inventory
        self.publish_command('reserve-inventory', {
            'saga_id': saga_id,
            'items': saga['data']['items']
        })

    def handle_inventory_reserved(self, event):
        saga_id = event['saga_id']

        # Update state
        self.save_saga_state(saga_id, SagaState.INVENTORY_RESERVED, event)

        # Step 3: Complete order
        self.publish_command('complete-order', {'saga_id': saga_id})
        self.save_saga_state(saga_id, SagaState.COMPLETED, event)

    def handle_failure(self, event):
        """Compensating transactions"""
        saga_id = event['saga_id']
        saga = self.get_saga(saga_id)

        self.save_saga_state(saga_id, SagaState.COMPENSATING, event)

        # Rollback based on current state
        if saga['state'] in [SagaState.PAYMENT_PROCESSING, SagaState.INVENTORY_RESERVED]:
            self.publish_command('refund-payment', {'saga_id': saga_id})

        if saga['state'] == SagaState.INVENTORY_RESERVED:
            self.publish_command('release-inventory', {'saga_id': saga_id})

        self.save_saga_state(saga_id, SagaState.FAILED, event)

# main.py - Flask application
from flask import Flask, request, jsonify
from google.cloud import datastore
from google.cloud import storage
from google.cloud import tasks_v2
import os

app = Flask(__name__)

# Configuration
PROJECT_ID = os.environ['GOOGLE_CLOUD_PROJECT']
ds_client = datastore.Client()

@app.route('/')
def index():
    return 'App Engine Application'

@app.route('/api/orders', methods=['POST'])
def create_order():
    """Create order and enqueue processing task"""
    order_data = request.get_json()

    # Save to Datastore
    key = ds_client.key('Order')
    entity = datastore.Entity(key=key)
    entity.update({
        'customer_id': order_data['customer_id'],
        'items': order_data['items'],
        'status': 'PENDING'
    })
    ds_client.put(entity)

    # Enqueue task for async processing
    enqueue_order_processing(entity.key.id)

    return jsonify({
        'order_id': entity.key.id,
        'status': 'PENDING'
    }), 201

@app.route('/tasks/process-order', methods=['POST'])
def process_order():
    """Background task handler"""
    # Verify request is from Task Queue
    if request.headers.get('X-AppEngine-QueueName') is None:
        return 'Unauthorized', 401

    order_id = request.get_json()['order_id']

    # Process order
    key = ds_client.key('Order', int(order_id))
    order = ds_client.get(key)
    order['status'] = 'PROCESSING'

    # ... process order logic ...

    order['status'] = 'COMPLETED'
    ds_client.put(order)

    return '', 200

def enqueue_order_processing(order_id):
    """Enqueue task to Cloud Tasks"""
    client = tasks_v2.CloudTasksClient()
    parent = client.queue_path(PROJECT_ID, 'us-central1', 'default')

    task = {
        'app_engine_http_request': {
            'http_method': tasks_v2.HttpMethod.POST,
            'relative_uri': '/tasks/process-order',
            'headers': {'Content-Type': 'application/json'},
            'body': json.dumps({'order_id': order_id}).encode()
        }
    }

    client.create_task(request={'parent': parent, 'task': task})

@app.route('/_ah/warmup')
def warmup():
    """Warmup handler to initialize resources"""
    # Initialize database connections, load data, etc.
    return '', 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=int(os.environ.get('PORT', 8080)))

# app.py - Optimized for Cloud Run concurrency
from flask import Flask
import threading
import queue

app = Flask(__name__)

# Thread-safe request processing
request_queue = queue.Queue(maxsize=100)

# Database connection pool (thread-safe)
from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

engine = create_engine(
    os.environ['DATABASE_URL'],
    poolclass=QueuePool,
    pool_size=10,  # Max connections
    max_overflow=20,  # Extra connections when needed
    pool_pre_ping=True,  # Verify connections
    pool_recycle=3600  # Recycle connections hourly
)

@app.route('/api/data', methods=['GET'])
def get_data():
    """Handle request with connection pooling"""
    with engine.connect() as conn:
        result = conn.execute("SELECT * FROM data LIMIT 10")
        return {'data': [dict(row) for row in result]}

# Graceful shutdown
import signal
import sys

def signal_handler(sig, frame):
    print('SIGTERM received, shutting down...')
    # Finish current requests
    # Close database connections
    engine.dispose()
    sys.exit(0)

signal.signal(signal.SIGTERM, signal_handler)

if __name__ == '__main__':
    # Listen on PORT environment variable
    port = int(os.environ.get('PORT', 8080))
    # Use production WSGI server
    from waitress import serve
    serve(app, host='0.0.0.0', port=port, threads=80)

# cloudsql_connection.py
import os
import sqlalchemy
from google.cloud.sql.connector import Connector

def get_connection_pool():
    """Create connection pool for Cloud SQL"""
    # Cloud Run provides Unix socket access
    instance_connection_name = os.environ['INSTANCE_CONNECTION_NAME']

    connector = Connector()

    def getconn():
        return connector.connect(
            instance_connection_name,
            "pg8000",
            user=os.environ['DB_USER'],
            password=os.environ['DB_PASS'],
            db=os.environ['DB_NAME']
        )

    pool = sqlalchemy.create_engine(
        "postgresql+pg8000://",
        creator=getconn,
        pool_size=5,
        max_overflow=2,
        pool_timeout=30,
        pool_recycle=1800,
    )
    return pool

# Usage
db_pool = get_connection_pool()

@app.route('/users/<int:user_id>')
def get_user(user_id):
    with db_pool.connect() as conn:
        result = conn.execute(
            sqlalchemy.text("SELECT * FROM users WHERE id = :id"),
            {"id": user_id}
        )
        user = result.fetchone()
        return jsonify(dict(user))

# Python HTTP function
import functions_framework
from flask import Request, jsonify

@functions_framework.http
def hello_http(request: Request):
    """HTTP Cloud Function.
    Args:
        request (flask.Request): The request object
    Returns:
        Response object
    """
    request_json = request.get_json(silent=True)
    request_args = request.args

    if request_json and 'name' in request_json:
        name = request_json['name']
    elif request_args and 'name' in request_args:
        name = request_args['name']
    else:
        name = 'World'

    return jsonify({'message': f'Hello {name}!'}), 200

# Deploy
# gcloud functions deploy hello_http \
#   --gen2 \
#   --runtime=python311 \
#   --region=us-central1 \
#   --source=. \
#   --entry-point=hello_http \
#   --trigger-http \
#   --allow-unauthenticated

# Python Pub/Sub function
import functions_framework
import base64
import json

@functions_framework.cloud_event
def process_pubsub(cloud_event):
    """Triggered from a message on a Cloud Pub/Sub topic.
    Args:
        cloud_event: CloudEvent containing Pub/Sub message
    """
    # Decode Pub/Sub message
    pubsub_message = base64.b64decode(
        cloud_event.data["message"]["data"]
    ).decode('utf-8')

    data = json.loads(pubsub_message)

    # Access message attributes
    attributes = cloud_event.data["message"].get("attributes", {})

    print(f"Processing message: {data}")
    print(f"Attributes: {attributes}")

    # Process the message
    process_order(data)

def process_order(order_data):
    """Process order logic"""
    print(f"Processing order: {order_data['order_id']}")
    # Business logic here

# Deploy
# gcloud functions deploy process_pubsub \
#   --gen2 \
#   --runtime=python311 \
#   --region=us-central1 \
#   --source=. \
#   --entry-point=process_pubsub \
#   --trigger-topic=order-events \
#   --retry  # Enable retries on failure

# Python Storage function
import functions_framework
from google.cloud import storage
from google.cloud import vision

@functions_framework.cloud_event
def process_image(cloud_event):
    """Triggered by Cloud Storage when file is uploaded.
    Args:
        cloud_event: CloudEvent containing Storage event data
    """
    data = cloud_event.data

    bucket_name = data["bucket"]
    file_name = data["name"]
    event_type = cloud_event["type"]  # google.cloud.storage.object.v1.finalized

    print(f"Event type: {event_type}")
    print(f"File: gs://{bucket_name}/{file_name}")

    # Ignore deletion events
    if event_type == "google.cloud.storage.object.v1.deleted":
        return

    # Process only images
    if not file_name.lower().endswith(('.png', '.jpg', '.jpeg', '.gif')):
        print(f"Skipping non-image file: {file_name}")
        return

    # Analyze image with Vision API
    analyze_image(bucket_name, file_name)

def analyze_image(bucket_name, file_name):
    """Analyze image and save results"""
    vision_client = vision.ImageAnnotatorClient()
    storage_client = storage.Client()

    # Get image from Storage
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(file_name)

    image = vision.Image()
    image.source.image_uri = f"gs://{bucket_name}/{file_name}"

    # Detect labels
    response = vision_client.label_detection(image=image)
    labels = [label.description for label in response.label_annotations]

    print(f"Labels: {labels}")

    # Save results to metadata
    blob.metadata = {'labels': ','.join(labels)}
    blob.patch()

# Deploy
# gcloud functions deploy process_image \
#   --gen2 \
#   --runtime=python311 \
#   --region=us-central1 \
#   --source=. \
#   --entry-point=process_image \
#   --trigger-bucket=my-upload-bucket \
#   --retry

# Python Firebase Auth function
import functions_framework
from google.cloud import firestore

@functions_framework.cloud_event
def on_user_create(cloud_event):
    """Triggered when new user is created in Firebase Auth.
    Args:
        cloud_event: CloudEvent with user data
    """
    data = cloud_event.data

    uid = data.get('uid')
    email = data.get('email')
    display_name = data.get('displayName')
    photo_url = data.get('photoURL')

    print(f"New user created: {uid}")
    print(f"Email: {email}")

    # Create user profile in Firestore
    db = firestore.Client()
    user_ref = db.collection('users').document(uid)
    user_ref.set({
        'email': email,
        'displayName': display_name or '',
        'photoURL': photo_url or '',
        'createdAt': firestore.SERVER_TIMESTAMP,
        'role': 'user',
        'status': 'active'
    })

    # Send welcome email
    send_welcome_email(email, display_name)

    print(f"User profile created for {uid}")

def send_welcome_email(email, name):
    """Send welcome email to new user"""
    # Use SendGrid, Mailgun, or other email service
    print(f"Sending welcome email to {email}")

# Deploy with Eventarc
# gcloud functions deploy on_user_create \
#   --gen2 \
#   --runtime=python311 \
#   --region=us-central1 \
#   --source=. \
#   --entry-point=on_user_create \
#   --trigger-event-filters="type=google.firebase.auth.user.v1.created"

# error_handling.py
import functions_framework
from google.cloud import pubsub_v1
import traceback
import time

@functions_framework.cloud_event
def resilient_function(cloud_event):
    """Function with comprehensive error handling"""
    try:
        # Process event
        process_event(cloud_event)

    except RetryableError as e:
        # Transient error - allow retry
        print(f"Retryable error: {e}")
        raise  # Re-raise to trigger retry

    except FatalError as e:
        # Permanent error - don't retry
        print(f"Fatal error: {e}")
        # Log to error reporting
        log_error(e, cloud_event)
        # Send to DLQ
        send_to_dlq(cloud_event)
        # Don't raise - prevent retry
        return

    except Exception as e:
        # Unknown error - log and decide
        print(f"Unknown error: {e}")
        print(traceback.format_exc())

        # Send to DLQ after multiple attempts
        attempt = cloud_event.data.get("message", {}).get("attributes", {}).get("retry_count", "0")
        if int(attempt) >= 3:
            send_to_dlq(cloud_event)
            return

        raise  # Retry

class RetryableError(Exception):
    """Transient errors that should be retried"""
    pass

class FatalError(Exception):
    """Permanent errors that should not be retried"""
    pass

def process_event(cloud_event):
    """Process event with validation"""
    data = cloud_event.data

    # Idempotency check
    message_id = data.get("message", {}).get("messageId")
    if is_already_processed(message_id):
        print(f"Message {message_id} already processed, skipping")
        return

    # Validate data
    if not validate_data(data):
        raise FatalError("Invalid data format")

    # Process
    result = do_processing(data)

    # Mark as processed
    mark_as_processed(message_id)

    return result

def send_to_dlq(cloud_event):
    """Send failed message to Dead Letter Queue"""
    publisher = pubsub_v1.PublisherClient()
    topic_path = publisher.topic_path('project-id', 'dlq-topic')

    message_data = str(cloud_event.data).encode('utf-8')
    future = publisher.publish(topic_path, message_data)
    print(f"Sent to DLQ: {future.result()}")

# Deploy with retry configuration
# gcloud functions deploy resilient_function \
#   --gen2 \
#   --runtime=python311 \
#   --region=us-central1 \
#   --source=. \
#   --entry-point=resilient_function \
#   --trigger-topic=events \
#   --retry \
#   --max-retry-attempts=5 \
#   --min-retry-delay=10s \
#   --max-retry-delay=300s