Azure Functions DB

⚠️ Experimental — pattern exploration. APIs and behavior may change. Not recommended as a production dependency yet.

Part of the Azure Functions Python DX Toolkit — dogfood-tested by azure-functions-cookbook-python.

Read this in: 한국어 | 日本語 | 简体中文

SQLAlchemy-powered database integration helpers for Azure Functions Python v2 — binding-style decorators for input/output/client injection and a poll-based pseudo trigger that works with any database that ships a SQLAlchemy dialect.

Not a native Azure Functions binding extension. This package does not register native Azure Functions bindings with the Functions host. The @db.input / @db.output / @db.trigger decorators are Python function wrappers that resolve data, inject writers, or poll for changes around your handler. For runtime-native Azure SQL bindings, use the official extension (see below).

Part of the Azure Functions Python DX Toolkit → Bring FastAPI-like developer experience to Azure Functions

Why this exists

Microsoft already ships official Azure SQL bindings — including a SQL trigger backed by SQL Change Tracking — for Azure SQL Database and SQL Server. If those cover your scenario, prefer them.

This package fills a different gap. Python teams on Azure Functions still have to hand-roll integration when they need:

• No generic SQLAlchemy trigger — the official SQL bindings target Azure SQL / SQL Server only.
• No unified multi-dialect binding layer — PostgreSQL, MySQL, SQLite, Oracle, DuckDB, CockroachDB, and other SQLAlchemy dialects each require custom code.
• No SQLAlchemy-native reader/writer injection pattern for the Python v2 programming model.
• No first-class polling primitive — checkpoint, lease, batching, idempotency, and at-least-once delivery built on top of the existing timer trigger.

See ADR-006 — Python Wrapper over Native Azure Functions Extension for the long-form rationale: why this project ships as a Python decorator wrapper on top of the timer trigger instead of a native .NET Azure Functions extension, what native binding metadata and scale-controller integration are traded away, and how the choice interacts with ADR-001, ADR-002, and ADR-004.

When to use this

Use azure-functions-db when:

• You need PostgreSQL, MySQL, SQLite, Oracle, DuckDB, CockroachDB, or another SQLAlchemy dialect.
• You want Python decorator-based input / output / client injection that maps cleanly onto v2 handlers.
• You want timer-driven polling with checkpoint and lease control instead of a host-managed trigger extension.
• You want local-first, testable DB integration without depending on a native Functions extension.

Use the official Azure SQL bindings when:

• You target Azure SQL Database or SQL Server.
• You want a runtime-native SQL trigger / input binding / output binding registered with the Functions host.
• You rely on SQL Change Tracking and extension-managed scaling for your trigger source.

What it does

• Poll-based pseudo trigger — change detection via cursor-column polling, with checkpoint, lease, batching, and at-least-once delivery on top of an Azure Functions timer trigger.
• Any SQLAlchemy database — PostgreSQL, MySQL, SQL Server out of the box; Oracle, CockroachDB, DuckDB, and any other dialect with one extra pip install.
• Single pip install — one package with optional extras for the common database drivers.
• Input-style data injection — @db.input resolves query results and passes them to your handler.
• Output-style writer injection — @db.output injects a DbOut writer; you call .set(...) to write explicitly.
• Client injection — @db.inject_reader / @db.inject_writer provide imperative DbReader / DbWriter clients for multi-statement reads or transactional writes.

Compared with official Azure SQL bindings

Quick pick:

• Azure SQL / SQL Server only, and you want the Functions host to manage everything → prefer the official Azure SQL bindings.
• PostgreSQL, MySQL, SQLite, or any other SQLAlchemy dialect → this package.
• Non-SQL source (Mongo, Kafka, HTTP) via a polling model → this package via SourceAdapter.
• Native binding metadata visible to Portal / scale controllers → the official extension. This package deliberately does not register with the host — see ADR-006.

Full comparison — see azure-functions-db vs Official Azure SQL Bindings for the axis-by-axis breakdown (supported databases, trigger mechanism, scaling integration, delivery guarantee, checkpoint storage and lifecycle, local testing experience, SQLAlchemy compatibility / BYOD, and production readiness).

Choose your integration path

from azure_functions_db import DbBindings

db = DbBindings()

@db.input("rows", url="oracle+oracledb://user:pass@host:1521/mydb",
          query="SELECT * FROM orders WHERE status = :status",
          params={"status": "pending"})
def read_oracle_orders(rows: list[dict]) -> None:
    for row in rows:
        print(row)

from azure_functions_db import SqlAlchemySource

source = SqlAlchemySource(
    url="oracle+oracledb://user:pass@host:1521/mydb",
    table="orders",
    cursor_column="updated_at",
    pk_columns=["id"],
)

Installation

# Core package (pick your database)
pip install azure-functions-db[postgres]
pip install azure-functions-db[mysql]
pip install azure-functions-db[mssql]

# Multiple databases
pip install azure-functions-db[postgres,mysql]

# All drivers
pip install azure-functions-db[all]

Your Function App dependencies should include:

azure-functions
azure-functions-db[postgres]

Quick Start

The fastest path — read rows with @db.input, write rows with @db.output:

from azure_functions_db import DbBindings, DbOut

db = DbBindings()

# Read: inject query results straight into your handler
@db.input("users", url="%DB_URL%",
          query="SELECT * FROM users WHERE active = :active",
          params={"active": True})
def list_active_users(users: list[dict]) -> None:
    for user in users:
        print(user["email"])

# Write: call .set() on the injected DbOut writer
@db.output("out", url="%DB_URL%", table="orders")
def create_order(out: DbOut) -> str:
    out.set({"id": 1, "status": "pending", "total": 99.99})
    return "Created"

New here? The snippet above is the 80% case. The reference below breaks down every decorator, plus plumbing and advanced patterns.

Which decorator to use?

All decorators are Python function wrappers. They are not registered as native Azure Functions bindings with the host.

import azure.functions as func
from azure_functions_db import DbBindings, DbOut

app = func.FunctionApp()
db = DbBindings()

@app.function_name(name="create_order")
@app.route(route="orders", auth_level=func.AuthLevel.FUNCTION)
@db.output("out", url="%DB_URL%", table="orders")
def create_order(req: func.HttpRequest, out: DbOut) -> func.HttpResponse:
    out.set({"id": 1, "status": "pending"})
    return func.HttpResponse("Created", status_code=201)

Input-style data injection

@db.input injects the actual query result into your handler — no client needed.

Row lookup mode — fetch a single row by primary key:

from azure_functions_db import DbBindings

db = DbBindings()

# Static primary key
@db.input("user", url="%DB_URL%", table="users", pk={"id": 42})
def load_user(user: dict | None) -> None:
    if user:
        print(user["name"])

# Dynamic primary key — resolved from handler kwargs
@db.input("user", url="%DB_URL%", table="users",
             pk=lambda req: {"id": req.params["id"]})
def get_user(req, user: dict | None) -> None:
    print(user)

Query mode — fetch multiple rows with SQL:

# Multiple rows by SQL query
@db.input("users", url="%DB_URL%",
             query="SELECT * FROM users WHERE active = :active",
             params={"active": True})
def list_active_users(users: list[dict]) -> None:
    for user in users:
        print(user["email"])

Output-style writer injection

@db.output injects a DbOut instance into your handler — call .set() to write explicitly.

from azure_functions_db import DbBindings, DbOut

db = DbBindings()

# Insert — call .set() with a dict for single row, list[dict] for batch
@db.output("out", url="%DB_URL%", table="orders")
def create_order(out: DbOut) -> str:
    out.set({"id": 1, "status": "pending", "total": 99.99})
    return "Created"

# Upsert — set action and conflict_columns
@db.output("out", url="%DB_URL%", table="orders",
              action="upsert", conflict_columns=["id"])
def upsert_orders(out: DbOut) -> str:
    out.set([
        {"id": 1, "status": "shipped", "total": 99.99},
        {"id": 2, "status": "pending", "total": 49.99},
    ])
    return "Upserted"

The handler's return value is independent of the write — use it for HTTP responses or anything else:

import azure.functions as func
from azure_functions_db import DbBindings, DbOut

db = DbBindings()

@db.output("out", url="%DB_URL%", table="orders")
def create_order(req: func.HttpRequest, out: DbOut) -> func.HttpResponse:
    out.set({"id": 1, "status": "pending"})
    return func.HttpResponse("Created", status_code=201)

Supported upsert dialects: PostgreSQL, SQLite, MySQL.

DbOut.set([]) is an explicit no-op: it accepts an empty list (e.g. when your handler computed nothing to write) and emits no SQL. Calling set again with a non-empty payload after a no-op still writes normally.

Client injection (imperative escape hatches)

For complex operations (multiple queries, transactions, update/delete), use inject_reader / inject_writer to receive a client instance:

from azure_functions_db import DbBindings, DbReader, DbWriter

db = DbBindings()

@db.inject_reader("reader", url="%DB_URL%", table="users")
def complex_read(reader: DbReader) -> None:
    user = reader.get(pk={"id": 42})
    orders = reader.query("SELECT * FROM orders WHERE user_id = :uid", params={"uid": 42})

    total = reader.scalar("SELECT COUNT(*) FROM users")
    profile = reader.one("SELECT * FROM users WHERE id = :id", params={"id": 42})
    maybe_admin = reader.one_or_none(
        "SELECT * FROM admins WHERE user_id = :uid", params={"uid": 42}
    )

@db.inject_writer("writer", url="%DB_URL%", table="orders")
def complex_write(writer: DbWriter) -> None:
    writer.insert(data={"id": 1, "status": "pending"})
    writer.update(data={"status": "shipped"}, pk={"id": 1})
    writer.delete(pk={"id": 1})

Atomic multi-statement writes — DbWriter.transaction()

transaction() is a context manager that opens a single SQLAlchemy connection, begins a transaction, and shares it across every insert / upsert / update / delete call made on the same writer inside the with block. The transaction commits on successful exit and rolls back on any exception, including WriteError raised by individual write calls.

@db.inject_writer("writer", url="%DB_URL%", table="orders")
def transfer(writer: DbWriter) -> None:
    with writer.transaction():
        writer.insert(data={"id": 1, "status": "pending", "total": 99.99})
        writer.update(data={"status": "shipped"}, pk={"id": 1})

Notes:

• Nested transactions are not supported; calling transaction() while one is active raises WriteError.
• After the with block exits the writer returns to its default per-call autocommit behavior.
• transaction() works only on the imperative DbWriter (inject_writer). The @db.output DbOut writer commits its single .set(...) call atomically and does not need a transaction wrapper.

DbReader row-shape helpers

Trigger (poll-based pseudo trigger)

This is a pseudo trigger, not a native Azure Functions trigger. @db.trigger does not register a binding with the Functions host. It must be stacked on top of a real Azure Functions trigger (typically @app.schedule / timer) that fires the polling loop.

Delivery is at-least-once. Duplicates may occur during process crashes, lease transitions, or checkpoint commit failures. Handlers must be idempotent. See Polling Runtime & Failure Scenarios for the full operational reference (tick lifecycle, duplicate windows, lease tuning, recovery procedures), Production Checklist before going to production, and Semantics — Duplicate Windows for the formal contract.

Poll-trigger lifecycle

One PollRunner.tick() runs per timer fire, using the concrete default components (BlobCheckpointStore + SqlAlchemySource). The step-by-step sequence diagram — plus the StateStore / SourceAdapter protocol names — lives in Architecture — Trigger Flow.

import azure.functions as func
from azure.storage.blob import ContainerClient
from azure_functions_db import BlobCheckpointStore, DbBindings, RowChange, SqlAlchemySource

app = func.FunctionApp()
db = DbBindings()

source = SqlAlchemySource(
    url="%ORDERS_DB_URL%",
    table="orders",
    schema="public",
    cursor_column="updated_at",
    pk_columns=["id"],
)

checkpoint_store = BlobCheckpointStore(
    container_client=ContainerClient.from_connection_string(
        conn_str="%AzureWebJobsStorage%",
        container_name="db-state",
    ),
    source_fingerprint=source.source_descriptor.fingerprint,
)

@app.function_name(name="orders_poll")
@app.schedule(schedule="0 */1 * * * *", arg_name="timer", use_monitor=True)
@db.trigger(arg_name="events", source=source, checkpoint_store=checkpoint_store)
def orders_poll(timer: func.TimerRequest, events: list[RowChange]) -> None:
    for event in events:
        # Idempotent processing required: the same `event` may be delivered more than once.
        print(f"Order {event.pk}: {event.op}")

See Python API Spec for the full API reference.

Advanced

Combined: Trigger + writer injection

Process database changes and write results to another table. Uses EngineProvider for shared connection pooling.

import azure.functions as func
from azure.storage.blob import ContainerClient

from azure_functions_db import (
    BlobCheckpointStore,
    DbBindings,
    DbOut,
    EngineProvider,
    RowChange,
    SqlAlchemySource,
)

app = func.FunctionApp()
db = DbBindings()

engine_provider = EngineProvider()

source = SqlAlchemySource(
    url="%SOURCE_DB_URL%",
    table="orders",
    cursor_column="updated_at",
    pk_columns=["id"],
    engine_provider=engine_provider,
)

checkpoint_store = BlobCheckpointStore(
    container_client=ContainerClient.from_connection_string(
        conn_str="%AzureWebJobsStorage%",
        container_name="db-state",
    ),
    source_fingerprint=source.source_descriptor.fingerprint,
)

@app.function_name(name="orders_poll")
@app.schedule(schedule="0 */1 * * * *", arg_name="timer", use_monitor=True)
@db.trigger(arg_name="events", source=source, checkpoint_store=checkpoint_store)
@db.output(
    "out",
    url="%DEST_DB_URL%",
    table="processed_orders",
    action="upsert",
    conflict_columns=["order_id"],
    engine_provider=engine_provider,
)
def orders_poll(timer: func.TimerRequest, events: list[RowChange], out: DbOut) -> None:
    out.set([
        {
            "order_id": event.pk["id"],
            "customer": event.after["name"],
            "processed_at": str(event.cursor),
        }
        for event in events
        if event.after is not None
    ])

See examples/trigger_with_binding/ for a complete runnable sample, or examples/postgresql-poll-trigger/ for an end-to-end PostgreSQL + Azurite docker-compose setup.

Built-in Extras

These databases have pre-packaged driver dependencies. Install the matching extra and you're ready to go.

Any other database with a SQLAlchemy dialect works too — just install the driver yourself. See Choose your integration path.

Support tiers

Database support falls into two explicit tiers:

Both tiers use the identical decorator and SqlAlchemySource API — the only difference is which drivers are bundled and which paths are covered by this project's tests. See Choose your integration path for the BYOD walkthrough.

Scope

• Azure Functions Python v2 programming model
• Timer-triggered functions for poll-based change detection
• SQLAlchemy 2.0+ for database abstraction
• Checkpoint storage via Azure Blob Storage
• Read/write bindings via HTTP/Queue/Event triggers

This package does not implement a native Azure Functions trigger extension. It uses a poll-based approach on top of the existing timer trigger.

Async handlers

async def handlers are supported. SQLAlchemy operations inside azure-functions-db run synchronously; when an async handler invokes a binding (input fetch, DbOut.set(...), polling commit, etc.) the package offloads the blocking call to a worker thread via asyncio.to_thread, so the event loop is not blocked.

This package does not use SQLAlchemy AsyncEngine internally. If you need fully native asyncio drivers (e.g. asyncpg, aiomysql), drive them yourself outside the binding — azure-functions-db deliberately exposes a single sync engine path so behavior across dialects stays identical.

Exception — @db.trigger does not support async handlers. Because PollTrigger.run() is synchronous, the trigger decorator rejects an async handler at decoration time by raising ConfigurationError; PollTrigger.run() additionally raises TypeError as a defensive runtime guard if it is ever handed an async callable. Use a synchronous handler for @db.trigger.

Async writer transactions

The async writer proxy injected by @db.inject_writer into async def handlers exposes insert, insert_many, upsert, upsert_many, update, delete, close, and a transaction() async context manager for multi-statement atomicity.

Because SQLAlchemy Connection / Transaction objects are not safe to share across threads — and asyncio.to_thread does not pin work to a single OS thread — transaction() pins the entire transaction to one dedicated worker thread for the duration of the async with block. Every write issued through the yielded proxy is routed to that thread, so the underlying connection is only ever touched by a single thread. The transaction commits on normal exit and rolls back if the block raises:

from azure_functions_db import DbBindings

db = DbBindings()


@db.inject_writer("writer", url="%DB_URL%", table="orders")
async def transfer(writer) -> None:
    async with writer.transaction() as tx:
        await tx.insert(data={"id": 1, "status": "pending"})
        await tx.update(data={"status": "shipped"}, pk={"id": 1})

Notes:

• Concurrent writes inside the block (e.g. via asyncio.gather) are serialized onto the pinned thread; a SQLAlchemy connection cannot be used concurrently even on one thread.
• A commit failure surfaces as WriteError; a failure during rollback is logged and the original exception is preserved.
• If you prefer to keep everything on a synchronous DbWriter, you can still wrap the whole unit in a single asyncio.to_thread call that drives DbWriter.transaction() end-to-end.

engine_kwargs flow-through

Every binding decorator and DbConfig accept an engine_kwargs mapping that is forwarded to sqlalchemy.create_engine. Anything the underlying dialect supports — connection / query timeouts, pool sizing, isolation level, custom event listeners — flows through unchanged. Use EngineProvider when several bindings should share a single engine instance with a consistent engine_kwargs configuration.

Note: Pass driver-level connect_args via the dedicated connect_args parameter, not nested inside engine_kwargs. Nesting connect_args inside engine_kwargs raises ConfigurationError because EngineProvider already owns that argument and silently overriding it would mask user intent.

See EngineProvider Lifecycle & SQLAlchemy Pooling Guidance for engine cache-key rules, recommended pool settings on Azure Functions (pool_pre_ping, pool_recycle, pool_size / max_overflow), per-dialect snippets, and SQLite test caveats.

Observability

azure-functions-db-python exposes structured log helpers plus a lightweight MetricsCollector protocol so you can connect your own metrics backend without adding hard dependencies.

from collections.abc import Mapping

from azure_functions_db import MetricsCollector, PollTrigger


class PrintMetricsCollector:
    def increment(
        self, name: str, value: float = 1, *, labels: Mapping[str, str] | None = None
    ) -> None:
        print("increment", name, value, labels)

    def observe(
        self, name: str, value: float, *, labels: Mapping[str, str] | None = None
    ) -> None:
        print("observe", name, value, labels)

    def set_gauge(
        self, name: str, value: float, *, labels: Mapping[str, str] | None = None
    ) -> None:
        print("gauge", name, value, labels)


trigger = PollTrigger(
    name="orders",
    source=source,
    checkpoint_store=checkpoint_store,
    metrics=PrintMetricsCollector(),
)

Key Design Decisions

• Pseudo trigger — timer-based polling instead of native C# extension (ADR-001)
• SQLAlchemy-centric — single ORM layer for all databases (ADR-002)
• Blob checkpoint — Azure Blob Storage for checkpoint persistence (ADR-003)
• At-least-once — default delivery guarantee with idempotency support (ADR-004)
• Unified package — trigger + binding in one package (ADR-005)
• Python wrapper, not native extension — Python decorators over the timer trigger instead of a .NET Azure Functions extension (ADR-006)
• Async transaction via pinned worker thread — async with writer.transaction() pins the transaction to a single worker thread instead of adding native async drivers (ADR-007)

Duplicate Handling

This package provides at-least-once delivery for the polling trigger. Duplicates may occur during process crashes, lease transitions, or checkpoint commit failures. Handlers must be idempotent. Recommended patterns:

• Use the row primary key (event.pk) plus event.cursor as a deduplication key in your sink.
• Wrap downstream writes in a transaction with a unique constraint that you can swallow.
• For batch writes, prefer upsert (action="upsert" with conflict_columns=...) over plain insert.

See Semantics — Duplicate Windows for the full guarantee model and the windows in which duplicates can be observed. See Polling Runtime & Failure Scenarios for tick lifecycle, lease tuning, and recovery procedures, and the Production Checklist for pre-deployment verification.

Documentation

• Full docs: yeongseon.dev/azure-functions-python/db
• Examples: examples/
• Architecture
• Semantics
• Python API Spec
• Adapter SDK

Canonical sources. The numbered specs and ADRs (docs/00-* … docs/28-*) are canonical for design and the API contract — docs/04-python-api-spec.md is authoritative. The standard site pages and the auto-generated API reference are derivations. When behavior changes, update 04-python-api-spec.md first, then the user-facing pages.

Ecosystem

Part of the Azure Functions Python DX Toolkit:

For AI Coding Assistants

This repository includes llms.txt and llms-full.txt in the root directory. These files provide comprehensive package and API information optimized for LLM context windows.

• llms.txt — Quick reference with core API, installation, and quick-start example
• llms-full.txt — Complete reference with full signatures, patterns, design principles, and ecosystem context

Use these files to get better context when working with this package in AI-assisted coding environments.

Disclaimer

This project is an independent community project and is not affiliated with, endorsed by, or maintained by Microsoft.

Azure and Azure Functions are trademarks of Microsoft Corporation.

License

MIT