Azure Functions Guide — Triggers, Bindings, Durable Functions, and Python Examples

# http_trigger.py
# Azure Function with HTTP trigger + input/output bindings
import azure.functions as func
import json
import logging
from datetime import datetime

def main(req: func.HttpRequest,
         inputblob: func.InputStream,
         outputblob: func.Out[str]) -> func.HttpResponse:
    """
    HTTP-triggered function that:
    - Accepts a name in the request body
    - Reads a template from Blob Storage (input binding)
    - Writes a personalized message (output binding)
    - Returns the result as HTTP response
    """
    logging.info('Python HTTP trigger function processed a request.')

    # Get name from request
    try:
        req_body = req.get_json()
        name = req_body.get('name', 'World')
    except ValueError:
        name = req.params.get('name', 'World')

    # Read template from blob (input binding)
    template = "Hello, {name}! This message was generated at {time}."
    if inputblob:
        template = inputblob.read().decode('utf-8')

    # Generate message
    message = template.format(name=name, time=datetime.now().isoformat())

    # Write to blob (output binding)
    outputblob.set(message)

    logging.info(f'Generated message for {name}')

    return func.HttpResponse(
        json.dumps({"message": message, "status": "success"}),
        mimetype="application/json",
        status_code=200
    )

# functions_simulator.py
# Simulate Azure Functions locally for learning
import time
import json
from datetime import datetime
from collections import defaultdict

class FunctionRuntime:
    def __init__(self, plan="Consumption", memory_mb=128):
        self.plan = plan
        self.memory = memory_mb
        self.stats = defaultdict(int)

    def http_trigger(self, method, path, body=None):
        """Simulate an HTTP-triggered function."""
        self.stats["total_invocations"] += 1
        start = time.time()

        # Simulate cold start
        if self.stats["total_invocations"] == 1:
            penalty = 0.3 if self.plan == "Consumption" else 0.05
            time.sleep(penalty)

        # Process request
        name = "World"
        if body:
            try:
                data = json.loads(body)
                name = data.get("name", name)
            except json.JSONDecodeError:
                pass

        result = {
            "message": f"Hello, {name}! Processed at {datetime.now().isoformat()}",
            "method": method,
            "path": path,
            "plan": self.plan,
        }

        duration = (time.time() - start) * 1000
        self.stats["total_duration_ms"] += duration

        print(f"[{self.plan}] HTTP {method} {path} → '{result['message']}' ({duration:.0f}ms)")
        return result

    def blob_trigger(self, container, blob_name):
        """Simulate a Blob Storage-triggered function."""
        self.stats["total_invocations"] += 1
        self.stats["blob_processed"] += 1
        print(f"[BlobTrigger] Processing {container}/{blob_name}...")
        time.sleep(0.1)
        print(f"[BlobTrigger] ✓ Resized thumbnail for {blob_name}")

    def stats_report(self):
        avg_duration = self.stats["total_duration_ms"] / max(self.stats["total_invocations"], 1)
        print(f"\n=== Function Runtime Stats ===")
        print(f"  Plan: {self.plan}")
        print(f"  Total invocations: {self.stats['total_invocations']}")
        print(f"  Average duration: {avg_duration:.0f}ms")
        print(f"  Blobs processed: {self.stats.get('blob_processed', 0)}")

# Demo
print("=== Azure Functions Simulation ===\n")

runtime = FunctionRuntime(plan="Consumption")
runtime.http_trigger("GET", "/api/hello")
runtime.http_trigger("POST", "/api/hello", '{"name": "DodaTech"}')
runtime.http_trigger("GET", "/api/health")
runtime.blob_trigger("uploads", "photos/selfie.jpg")
runtime.blob_trigger("uploads", "docs/report.pdf")
runtime.stats_report()

# durable_demo.py
# Simulate Durable Functions chaining pattern
import time
import json
from datetime import datetime

class DurableOrchestrator:
    def __init__(self):
        self.history = []

    def call_activity(self, name, input_data):
        """Simulate calling an activity function."""
        start = time.time()
        time.sleep(0.3)  # Simulate work
        result = f"{name}({input_data}) → processed at {datetime.now().strftime('%H:%M:%S')}"
        self.history.append({"activity": name, "input": input_data, "result": result, "duration": round((time.time()-start)*1000)})
        return result

    def orchestrate(self):
        """Simulate a Durable Functions orchestration."""
        print("🏁 Orchestration started\n")

        # Step 1: Validate order
        order = {"order_id": "ORD-2026-001", "amount": 150.00, "customer": "dodatech"}
        step1 = self.call_activity("ValidateOrder", order)
        print(f"  Step 1: {step1}")

        # Step 2: Process payment
        step2 = self.call_activity("ProcessPayment", {"order_id": order["order_id"], "amount": order["amount"]})
        print(f"  Step 2: {step2}")

        # Step 3: Send confirmation (fan-out)
        confirmations = ["SendEmail", "SendSMS", "UpdateCRM"]
        for activity in confirmations:
            result = self.call_activity(activity, {"order_id": order["order_id"]})
            print(f"  Step 3: {result}")

        print(f"\n🏁 Orchestration completed")
        return {"status": "completed", "steps": len(self.history), "history": self.history}

orch = DurableOrchestrator()
result = orch.orchestrate()

print(f"\n=== Orchestration Summary ===")
for step in result["history"]:
    print(f"  {step['activity']:<20} {step['duration']:>4}ms")
print(f"\nTotal steps: {result['steps']}")

# azure_workflow.py
# Simulate a multi-step Azure Functions workflow
import time
import json

class Workflow:
    def __init__(self):
        self.steps = []
        self.context = {}

    def add_step(self, name, func, depends_on=None):
        self.steps.append({"name": name, "func": func, "depends_on": depends_on or []})

    def run(self, initial_input):
        results = {}
        for step in self.steps:
            all_deps_ready = all(dep in results for dep in step["depends_on"])
            if not all_deps_ready:
                print(f"  ⏳ {step['name']}: waiting for dependencies")
                continue
            print(f"  ▶ {step['name']}: executing...")
            output = step["func"]({"input": initial_input, "context": results})
            results[step["name"]] = output
            print(f"    ✓ {output}")
        return results

wf = Workflow()
wf.add_step("ValidateOrder", lambda ctx: "Order validated")
wf.add_step("ProcessPayment", lambda ctx: f"Payment of ${ctx['input']['amount']} processed", depends_on=["ValidateOrder"])
wf.add_step("UpdateInventory", lambda ctx: "Inventory updated", depends_on=["ProcessPayment"])
wf.add_step("SendNotification", lambda ctx: "Email + SMS sent", depends_on=["UpdateInventory"])
wf.add_step("AuditLog", lambda ctx: "Audit logged", depends_on=["ProcessPayment", "UpdateInventory"])

print("=== Azure Functions Workflow ===\n")
wf.run({"order_id": "ORD-001", "amount": 129.99})

print("\n✓ Workflow complete")