Intro
Most Python tutorials for Azure Functions feel like a single notebook cell: parse JSON, run a calculation, return. That’s perfect for a demo and exactly the wrong shape once you want to trust it. In production, payloads drift, retries happen, and a “quick function” becomes the piece of code you’re reluctant to touch.
We learned that the hard way building ESG pipelines. The fix wasn’t a heavy framework. It was a small amount of structure that keeps Azure Functions concerns separate from your domain logic and makes your Python code testable locally—fast.
What Are Azure Functions? (and why quants should care)
Azure Functions are Microsoft’s serverless offering: Python code that runs when events happen. An HTTP request arrives, a message appears on a queue, a timer ticks—your function executes, no servers to manage. You only pay for what runs.
For quants and modellers, this is ideal for workflows like recording simulation results, starting post-processing, moving artifacts, and updating a dashboard. Treat each Function as a one-off script and it will bite you. Give it some design and it behaves.
The Architecture: How to Structure Python Azure Functions
The pattern is deliberately small:
- The Function is just an adapter. It reads input, wraps it in an
EventEnvelope, and hands it to a handler. - The Handler is a small Python class that does one job, with a typed payload validated by Pydantic.
- A Router chooses the right handler based on
(event_type, version). - A Composition module builds shared clients (database, storage, HTTP) once and injects them into handlers.
This “clean code” approach for Azure Functions gives you three things immediately: bad data fails at the edge, retries don’t corrupt state, and handlers are trivial to unit test without Azure.
Example: ESG Simulation Results and VaR Post-Processing
Imagine your ESG engine has just finished a simulation for a portfolio. The next step is post-processing: fetch the simulation artifact, compute a 99% VaR, persist the result, and update a status so dashboards can consume it.
Instead of a single mega-payload full of optional fields, we define an explicit event: esg.simulation.completed (v1). That event is the only thing the Function needs to recognise; the handler takes care of the rest.
The envelope (everything passes through this):
from pydantic import BaseModel
from typing import Any
class EventEnvelope(BaseModel):
type: str # e.g. "esg.simulation.completed"
version: str = "v1"
correlation_id: str
payload: dict[str, Any]
message_id: str | None = None
The payload (typed, no optional soup):
from pydantic import BaseModel, constr
class SimulationCompleted(BaseModel):
run_id: constr(min_length=8, max_length=64)
portfolio_id: constr(min_length=3, max_length=32)
as_of: constr(min_length=10, max_length=25)
artifact_uri: str
correlation_id: constr(min_length=8, max_length=64)
The handler (business logic):
import numpy as np
from common.service.base import BaseEventHandler
class ComputeVarHandler(BaseEventHandler[SimulationCompleted]):
def __init__(self, artifacts, store, idem):
super().__init__(schema=SimulationCompleted, idem=idem)
self.artifacts, self.store = artifacts, store
def process(self, m: SimulationCompleted, env: EventEnvelope) -> None:
df = self.artifacts.read_parquet(m.artifact_uri)
losses = -(df["pnl"].to_numpy())
var_99 = float(np.quantile(losses, 0.99))
self.store.upsert_var_result(
run_id=m.run_id,
portfolio_id=m.portfolio_id,
as_of=m.as_of,
var_99=var_99,
method="historical",
corr=env.correlation_id,
)
self.store.update_status(model="esg", as_of=m.as_of, status="var_ready")
The handler is the function that performs the actual work that we want our function to perform. Note that it is inherited from a baseclass and there are type annotations and - more importantly - the expected message type is passed into the constructor. This way , the base class can validate the payload with Pydantic before processing and can run an idempotency check to avoid double-processing on retries.
Service Bus dispatcher (adapter):
from azure.functions import ServiceBusMessage
def main(msg: ServiceBusMessage) -> None:
env = EventEnvelope.model_validate_json(msg.get_body().decode())
env.message_id = getattr(msg, "message_id", None)
resolve(env)(get_services()).handle(env)
HTTP endpoint (adapter):
import azure.functions as func
def main(req: func.HttpRequest) -> func.HttpResponse:
payload = SimulationCompleted.model_validate_json(req.get_body())
env = EventEnvelope(
type="esg.simulation.completed",
version="v1",
correlation_id=payload.correlation_id,
payload=payload.model_dump(),
)
svc = get_services()
ComputeVarHandler(svc.artifacts, svc.store, svc.idem).handle(env)
return func.HttpResponse(status_code=202)
Ultimately, these are just our entrypoints for calling functions. In the Http endpoint example, we're directly calling our handler. In our servicebus version, we're using a small registry that keeps track of what messages types on the envelope get mapped to what schemas and to what functions they are routed.
Why This Azure Functions Pattern Works
This architecture keeps concerns separate: Functions handle triggers and bindings; handlers contain domain logic and are type-safe; the router is a small mapping; the composition module wires up shared infrastructure once. If somebody wants you to add Expected Shortfall tomorrow, you add a new schema and handler — nothing else changes.
Testability & Performance
Because Azure code is confined to adapters, testing is just Python. Handlers take explicit dependencies, so you can inject fakes. A minimal unit test could read: “Given a known P&L DataFrame, when ComputeVarHandler.handle runs, then the store receives a result with the correct VaR.” Tests run in milliseconds—no emulators, no deployment — which makes you actually write them.
Pydantic v2 validation is fast, router lookups are instant. The overhead is a couple of milliseconds. Real latency is in storage or database calls you already make. Clean architecture costs almost nothing at runtime and saves hours in operations.
Conclusion
Azure Functions are excellent for running code; they won’t organise it for you. By introducing just enough structure - an envelope, typed payloads, small handlers, and a composition module — you gain confidence (bad data fails early, retries don’t corrupt state), speed (tests run locally), and clarity (every Function looks the same).
For an ESG pipeline that just finished a simulation and needs a VaR, this turns a fragile script into a dependable component — without turning your Python into a framework. Start with one handler and one envelope. When you add the second, you’ll be glad the rails are already there.
Appendix: Idempotency in Azure Functions
Cloud queues and HTTP endpoints often deliver events at least once. If a worker fails mid-way, the same event can arrive again. An idempotency store is a simple ledger of “I’ve seen this key before,” so retries don’t redo the same effect.
We use a key — by default message_id from Service Bus or a domain key like run_id — to decide whether to proceed. For small setups, an in-memory store per worker is enough. For cross-worker dedupe, swap in Redis or Table Storage. The difference between “compute VaR once” and “compute and persist it twice” is the difference between a reliable system and corrupted history. Idempotency makes the former easy.
