Database Support¶

Back to Transactional Outbox Overview

Return to the Transactional Outbox overview page with all topics.

Back to Overview
Implementation

Repository interface, SQLAlchemy model and outbox table structure.

Read More
Usage

Event registration and publishing with at-least-once delivery guarantees.

Read More

Overview¶

The SQLAlchemy outbox implementation renders its DDL through dialect-aware column types. The same OutboxModel produces native types on every supported database: BINARY(16) on MySQL, BYTEA and UUID on PostgreSQL, and a portable BLOB fallback everywhere else.

Supported out of the box

MySQL, MariaDB and PostgreSQL are covered by built-in dialect handlers.

Support Matrix
Extensible

Any other dialect works via the portable fallback or a one-line register_dialect() call.

Add Your Database
Backward compatible

MySQL DDL is unchanged — existing deployments need no migration.

Compatibility Notes

Prerequisites

See Implementation for the repository interface and the outbox table structure, and Usage for event registration and publishing.

Support Matrix¶

Database	Dialect name	Driver (async)	Status
MySQL	`mysql`	`asyncmy`, `aiomysql`	✅ Built-in, covered by integration tests
MariaDB	`mariadb`	`asyncmy`, `aiomysql`	✅ Built-in (same handler as MySQL)
PostgreSQL	`postgresql`	`asyncpg`	✅ Built-in, covered by integration tests
SQLite	`sqlite`	`aiosqlite`	⚠️ Portable fallback (`BLOB`), useful for tests
SQL Server	`mssql`	`aioodbc`	⚠️ Portable fallback, native types via one registration
Oracle and others	`oracle`, …	driver-specific	⚠️ Portable fallback, or register a native type

SQLite and concurrent publishing

SQLite works for unit tests and local experiments, but it does not provide the row-level locking and isolation guarantees the publisher process relies on. Use MySQL/MariaDB or PostgreSQL in production.

Type Mapping¶

Three columns go through the dialect-aware layer: event_id (the idempotency UUID), event_id_bin (its 16-byte binary form used by the unique constraint) and payload (the serialized event). They are declared with the public types from cqrs.sqlalchemy_types — the same layer that provides UUIDBinary, Binary16, PayloadBinary and the saga storage's JSONType:

Model column	Python type	SQLAlchemy type	PostgreSQL	MySQL / MariaDB	Fallback
`event_id`	`uuid.UUID`	`UUIDBinary`	`UUID`	`BINARY(16)`	`BLOB` (`LargeBinary(16)`)
`event_id_bin`	`bytes`	`Binary16`	`BYTEA`	`BINARY(16)`	`BLOB` (`LargeBinary(16)`)
`payload`	`bytes`	`PayloadBinary`	`BYTEA`	`BLOB`	`BLOB` (`LargeBinary()`)
`event_status`	`EventStatus`	`Enum(EventStatus)`	`eventstatus` (native enum type)	`ENUM('NEW','PRODUCED','NOT_PRODUCED')`	`VARCHAR(12)`
`id`	`int`	`BigInteger` + `Identity()`	`BIGINT GENERATED BY DEFAULT AS IDENTITY`	`BIGINT AUTO_INCREMENT`	`BIGINT`
`created_at`	`datetime.datetime`	`DateTime`	`TIMESTAMP WITHOUT TIME ZONE`	`DATETIME`	`DATETIME`

On PostgreSQL event_id is stored as a native UUID, so asyncpg returns uuid.UUID values directly. On MySQL/MariaDB the value is converted to its 16 bytes on bind and back to uuid.UUID on read — which is exactly the behaviour the outbox had before dialect-aware types were introduced.

Enum values are member names

sqlalchemy.Enum(EventStatus) persists the names of the Python enum members, not their values. The database contains NEW, PRODUCED and NOT_PRODUCED — keep that in mind when writing raw SQL or hand-crafted migrations.

Ready-to-use DDL¶

If you create the outbox table manually instead of using Base.metadata.create_all() or Alembic, use the statements below. They match what SQLAlchemy renders for each dialect.

PostgreSQL¶

CREATE TYPE eventstatus AS ENUM ('NEW', 'PRODUCED', 'NOT_PRODUCED');

CREATE TABLE outbox (
    id BIGINT GENERATED BY DEFAULT AS IDENTITY,
    event_id UUID NOT NULL,
    event_id_bin BYTEA NOT NULL,
    event_status eventstatus NOT NULL,
    flush_counter SMALLINT NOT NULL,
    event_name VARCHAR(255) NOT NULL,
    topic VARCHAR(255) NOT NULL,
    created_at TIMESTAMP WITHOUT TIME ZONE DEFAULT now() NOT NULL,
    payload BYTEA NOT NULL,
    PRIMARY KEY (id),
    CONSTRAINT event_id_unique_index UNIQUE (event_id_bin, event_name)
);

MySQL / MariaDB¶

CREATE TABLE outbox (
    id BIGINT NOT NULL AUTO_INCREMENT,
    event_id BINARY(16) NOT NULL,
    event_id_bin BINARY(16) NOT NULL,
    event_status ENUM('NEW', 'PRODUCED', 'NOT_PRODUCED') NOT NULL,
    flush_counter SMALLINT NOT NULL,
    event_name VARCHAR(255) NOT NULL,
    topic VARCHAR(255) NOT NULL,
    created_at DATETIME NOT NULL DEFAULT now(),
    payload BLOB NOT NULL,
    PRIMARY KEY (id),
    CONSTRAINT event_id_unique_index UNIQUE (event_id_bin, event_name)
);

Defaults for flush_counter and topic

Those columns have Python-side defaults (0 and ''), so no DEFAULT clause is rendered. Add one yourself if other writers insert into the table directly.

Alembic¶

1. Register the outbox metadata¶

Add cqrs.outbox.sqlalchemy.Base.metadata to target_metadata in alembic/env.py, next to your own metadata:

import cqrs.outbox.sqlalchemy as cqrs_sqlalchemy
from myapp import orm

target_metadata = [orm.Base.metadata, cqrs_sqlalchemy.Base.metadata]

2. Make the custom types importable in migrations¶

Autogenerated migrations reference the dialect-aware types by their fully qualified names, so the migration script must import the module:

op.create_table(
    "outbox",
    sa.Column("event_id", cqrs.sqlalchemy_types.UUIDBinary(), nullable=False),
    sa.Column("event_id_bin", cqrs.sqlalchemy_types.Binary16(), nullable=False),
    ...
)

Add the import once to alembic/script.py.mako so every generated revision has it:

"""${message}

Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}

"""
from alembic import op
import sqlalchemy as sa
import cqrs.sqlalchemy_types  # <- required by the outbox columns
${imports if imports else ""}

Without this import the migration fails

A revision that uses cqrs.sqlalchemy_types.Binary16() without importing the module raises NameError: name 'cqrs' is not defined when Alembic runs it. This is why the types live at the short public path cqrs.sqlalchemy_types rather than inside cqrs.outbox.

Alternative: user_module_prefix

Instead of editing the template you can set user_module_prefix in context.configure() and import the module under that prefix in the template. Editing script.py.mako is the simpler of the two.

3. The `eventstatus` enum type on PostgreSQL¶

op.create_table() emits CREATE TYPE eventstatus AS ENUM (...) automatically before creating the table, so a fresh upgrade works as-is. Two caveats:

downgrade() does not drop the type. Alembic removes the table but leaves eventstatus behind, and the next upgrade() then fails with type "eventstatus" already exists. Drop it explicitly:
```
def downgrade() -> None:
    op.drop_table("outbox")
    sa.Enum(name="eventstatus").drop(op.get_bind(), checkfirst=True)
```
If the type already exists (for example the table was created by create_all() earlier), create it with checkfirst=True and pass create_type=False to the column type in the migration.

4. Existing MySQL deployments see no diff¶

Running alembic revision --autogenerate after upgrading python-cqrs produces an empty migration for the outbox table on MySQL, because the rendered DDL is byte-for-byte identical to the previous one.

Backward Compatibility¶

Aspect	Change
Application code	None — `SqlAlchemyOutboxedEventRepository` and its public API are unchanged
Column names	None — `event_id`, `event_id_bin` and the `event_id_unique_index` constraint are kept
MySQL / MariaDB DDL	None — `event_id_bin` is still `BINARY(16)`
PostgreSQL DDL	`event_id_bin` is now `BYTEA` instead of the invalid `BINARY(16)`
`payload` DDL	None — now declared with `PayloadBinary`, which renders the same `BYTEA`/`BLOB` as before
Imports	`BinaryUUID` still works as a deprecated alias of `UUIDBinary`

No migration needed on MySQL

Upgrading the package does not require any schema change on MySQL or MariaDB. A DDL-compilation unit test guards this guarantee.

PostgreSQL users on an older version

Before dialect-aware types, a migration generated from the CQRS metadata failed on PostgreSQL with type "binary" does not exist, so there is no legacy PostgreSQL schema to migrate from — just generate the migration again after upgrading.

Switching an application to PostgreSQL is a DSN change only:

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from cqrs import SqlAlchemyOutboxedEventRepository

session_factory = async_sessionmaker(
    create_async_engine(
        "postgresql+asyncpg://user:password@localhost/database",
        isolation_level="REPEATABLE READ",
    ),
)
outbox = SqlAlchemyOutboxedEventRepository(session=session_factory())

Adding Your Own Database¶

Out of the box the library knows MySQL, MariaDB and PostgreSQL. Every other dialect — SQL Server, Oracle, SQLite, anything else with a SQLAlchemy driver — still works: the dialect-aware columns fall back to a portable generic type (LargeBinary), the table is created and the outbox operates normally.

Registration is optional

You only need register_dialect() if the fallback type is not what you want to live with. Nothing breaks without it — you simply get a generic binary column instead of the database's native type.

Reasons to register a handler anyway:

The fallback renders to a type you consider legacy or deprecated on your database.
You want the column to carry semantics the fallback loses (e.g. a real GUID type instead of raw bytes).
You need a fixed-width column for index efficiency instead of a variable-length one.
Your DBA/schema review simply does not accept the generic type.

Public API¶

cqrs.sqlalchemy_types exposes the building blocks:

Name	Purpose
`DialectAwareType`	Base `TypeDecorator` that resolves a handler by dialect name
`DialectTypeHandler`	How a type lives in one dialect: `type_factory`, optional `bind` and `result` converters
`UUIDBinary`	`uuid.UUID` column — used by `event_id`
`Binary16`	16-byte binary column — used by `event_id_bin`
`PayloadBinary`	Binary blob of arbitrary length — used by `payload`
`JSONType`	JSON document — used by the saga storage `context` column
`register_dialect(dialect_name, handler)`	Classmethod registering a handler for one dialect
`get_handler(dialect_name)`	Classmethod returning the handler in effect (or the fallback)

JSONType belongs to the saga storage

The type layer is shared by the whole library, not just the outbox. JSONType renders as plain JSON on every dialect and exists so that a project can opt into PostgreSQL JSONB with one registration — see Saga Storage. Everything below about handlers, placement and registration order applies to it unchanged.

Registrations are per type

Each subclass keeps its own handler registry, so registering a handler for Binary16 does not affect UUIDBinary (or any custom DialectAwareType of your own). Register for every type you care about.

Worked Example: Microsoft SQL Server¶

SQL Server is a good illustration: the fallback works, but the result is not what you would design by hand.

What the fallback gives you¶

Compiled from the shipped model against the default mssql (pyodbc) dialect, before registering anything:

CREATE TABLE outbox (
    id BIGINT NOT NULL IDENTITY,
    event_id VARBINARY(16) NOT NULL,
    event_id_bin VARBINARY(16) NOT NULL,
    event_status VARCHAR(12) NOT NULL,
    flush_counter SMALLINT NOT NULL,
    event_name VARCHAR(255) NOT NULL,
    topic VARCHAR(255) NOT NULL,
    created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    payload VARBINARY(max) NOT NULL,
    PRIMARY KEY (id),
    CONSTRAINT event_id_unique_index UNIQUE (event_id_bin, event_name)
);

That is already a usable schema — note that payload lands on VARBINARY(max) with no registration at all. Two things are still worth fixing:

event_id loses its meaning. It is a UUID, but the column is raw binary. SQL Server has a first-class UNIQUEIDENTIFIER type; without it you cannot compare or join the column against GUID columns, and every report or ad-hoc query shows a hex blob.
VARBINARY is variable-length for a value that is always exactly 16 bytes. BINARY(16) is fixed-width and a better key for event_id_unique_index.

Why VARBINARY and not the deprecated IMAGE

Going through the type layer is what earns this. DialectAwareType.load_dialect_impl() resolves the fallback via dialect.type_descriptor(...), which gives the driver's colspecs a chance to map LargeBinary to _VARBINARY_pyodbc. A plain sqlalchemy.LargeBinary column skips that step during DDL compilation and renders as IMAGE, which Microsoft has deprecated.

The mapping lives in the driver, so it is driver-dependent: mssql+pyodbc and mssql+aioodbc both give VARBINARY, while a driverless mssql dialect has no such colspec and still renders IMAGE — for event_id, event_id_bin and payload alike. A registration pins the type regardless of driver.

Step 1 — Install an async driver and pick the DSN¶

The outbox repository works on an AsyncSession, so the driver has to be an async one:

pip install aioodbc  # plus the system "ODBC Driver 18 for SQL Server"

create_async_engine(
    "mssql+aioodbc://user:password@localhost:1433/database"
    "?driver=ODBC+Driver+18+for+SQL+Server",
    isolation_level="REPEATABLE READ",
)

Register by dialect name, not by driver

Handlers are keyed by dialect.name, which is mssql for both mssql+pyodbc and mssql+aioodbc (and oracle for oracle+oracledb and oracle+oracledb_async). One registration covers every driver of that database.

Step 2 — Put the registrations in one importable module¶

Create a module that does nothing but register types, for example myapp/db_types.py:

from sqlalchemy.dialects import mssql

from cqrs.sqlalchemy_types import Binary16, DialectTypeHandler, PayloadBinary, UUIDBinary

# event_id: a real GUID column instead of raw bytes.
UUIDBinary.register_dialect(
    "mssql",
    DialectTypeHandler(type_factory=lambda dialect: mssql.UNIQUEIDENTIFIER(as_uuid=True)),
)

# event_id_bin: always exactly 16 bytes, so fixed-width BINARY beats VARBINARY.
Binary16.register_dialect(
    "mssql",
    DialectTypeHandler(type_factory=lambda dialect: mssql.BINARY(16)),
)

# payload: optional — the driver already maps it to VARBINARY(max); this pins it
# so the type does not depend on which ODBC driver compiles the DDL.
PayloadBinary.register_dialect(
    "mssql",
    DialectTypeHandler(type_factory=lambda dialect: mssql.VARBINARY("max")),
)

That is the whole mapping — no bind/result converters are needed here, because SQLAlchemy's own UNIQUEIDENTIFIER(as_uuid=True) already converts between uuid.UUID and the GUID representation the driver expects, while BINARY(16) and VARBINARY(max) take the bytes the repository passes as-is. With those handlers in place the model compiles to:

CREATE TABLE outbox (
    id BIGINT NOT NULL IDENTITY,
    event_id UNIQUEIDENTIFIER NOT NULL,
    event_id_bin BINARY(16) NOT NULL,
    event_status VARCHAR(12) NOT NULL,
    flush_counter SMALLINT NOT NULL,
    event_name VARCHAR(255) NOT NULL,
    topic VARCHAR(255) NOT NULL,
    created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    payload VARBINARY(max) NOT NULL,
    PRIMARY KEY (id),
    CONSTRAINT event_id_unique_index UNIQUE (event_id_bin, event_name)
);

Step 3 — Import that module in both entry points¶

The registrations only exist in the process that imported them, so import the module in every process that touches the outbox table:

# myapp/main.py — application entry point
import myapp.db_types  # noqa: F401  registers the mssql handlers

from myapp import api

...  # create_async_engine(), bootstrap(), etc.

# alembic/env.py — migration process
import myapp.db_types  # noqa: F401  must be imported before autogenerate runs

import cqrs.outbox.sqlalchemy as cqrs_sqlalchemy
from myapp import orm

target_metadata = [orm.Base.metadata, cqrs_sqlalchemy.Base.metadata]

Forgetting alembic/env.py is the classic mistake

If only the application registers the handler, migrations are generated from the fallback types while the running application expects the native ones. The schema and the runtime then disagree — the migration creates VARBINARY(16), the app binds values as UNIQUEIDENTIFIER.

Step 4 — Register before the first use, not later¶

Put the import at the very top of the entry point, before the engine is created and before the first query or create_all(). SQLAlchemy memoizes bind/result processors per dialect instance, so a registration that happens after the engine has already processed a value is silently ignored by that engine:

When you register	DDL compilation	Value conversion
Before the engine is used	native type	native converters
After the engine processed a value	native type	still the fallback converters

The second row is the dangerous one: the table would be created with UNIQUEIDENTIFIER, but the already-cached fallback processor keeps returning the raw GUID string instead of a uuid.UUID.

Step 5 — Add `bind`/`result` when the native type needs conversion¶

A handler has three parts:

type_factory(dialect) — required. Returns the TypeEngine used for DDL and for binding.
bind(value) — optional. Converts the Python value before it goes to the database. Omit it to pass the value through unchanged.
result(value) — optional. Converts the value coming back from the database. Omit it to pass it through.

Whether you need converters depends entirely on the native type you picked:

Native type for `event_id`	Converters	Why
`mssql.UNIQUEIDENTIFIER(as_uuid=True)`	none	The SQLAlchemy type already maps `uuid.UUID` ⇄ GUID
`oracle.RAW(16)`	`bind` + `result`	A raw byte column knows nothing about UUIDs

For a byte-oriented native type you must spell the conversion out, otherwise the driver receives a uuid.UUID it cannot bind and reads back bytes where the repository expects uuid.UUID:

UUIDBinary.register_dialect(
    "oracle",
    DialectTypeHandler(
        type_factory=lambda dialect: oracle.RAW(16),
        bind=lambda value: value.bytes,
        result=lambda value: uuid.UUID(bytes=value),
    ),
)

None never reaches your converters

DialectAwareType short-circuits NULL, so bind/result are only called with non-None values.

Step 6 — Verify the mapping without a database¶

Before provisioning anything, compile the model's DDL against your dialect. This catches a missing or misplaced registration in a second:

import myapp.db_types  # noqa: F401  the registrations under test

from sqlalchemy.dialects import mssql
from sqlalchemy.schema import CreateTable

from cqrs.outbox.sqlalchemy import OutboxModel

print(CreateTable(OutboxModel.__table__).compile(dialect=mssql.dialect()))

Make it a test so a lost import cannot slip through review:

def test_outbox_uses_native_mssql_types() -> None:
    ddl = str(CreateTable(OutboxModel.__table__).compile(dialect=mssql.dialect()))

    assert "event_id UNIQUEIDENTIFIER" in ddl
    assert "event_id_bin BINARY(16)" in ddl
    assert "VARBINARY(16)" not in ddl  # the fallback is gone

Step 7 — Check the database-specific pitfalls¶

Two parts of the model are portable in SQLAlchemy but constrained by the database itself:

The unique constraint. event_id_unique_index spans event_id_bin plus event_name VARCHAR(255), and index key size limits are per-database: SQL Server caps a non-clustered key at 1700 bytes, MySQL's InnoDB at 767 bytes per column unless DYNAMIC row format is used (255 characters of utf8mb4 is 1020 bytes, which is why this can bite), and Oracle derives its limit from the block size. If your collation makes the key too wide, shorten event_name in your own migration rather than dropping the constraint — the outbox relies on it for idempotency.
Identity() on the primary key. It renders as IDENTITY on SQL Server and GENERATED BY DEFAULT AS IDENTITY on Oracle 12c and newer, but older Oracle releases have no identity columns at all and need a sequence plus trigger, and SQLite ignores it (a BIGINT primary key is not auto-incrementing there — only INTEGER PRIMARY KEY is). On such databases provide the key generation yourself in the migration.

Also keep in mind that sqlalchemy.Enum(EventStatus) only becomes a native enum on MySQL and PostgreSQL; everywhere else it is a plain 12-character string column holding NEW, PRODUCED or NOT_PRODUCED.

Compact Example: Oracle¶

The same recipe on Oracle is four lines — RAW(16) for both columns, with converters on event_id because RAW is a plain byte type:

import uuid

from sqlalchemy.dialects import oracle

from cqrs.sqlalchemy_types import Binary16, DialectTypeHandler, UUIDBinary

UUIDBinary.register_dialect(
    "oracle",
    DialectTypeHandler(
        type_factory=lambda dialect: oracle.RAW(16),
        bind=lambda value: value.bytes,
        result=lambda value: uuid.UUID(bytes=value),
    ),
)
Binary16.register_dialect(
    "oracle",
    DialectTypeHandler(type_factory=lambda dialect: oracle.RAW(16)),
)

With that module imported, the model compiles to event_id RAW(16) and event_id_bin RAW(16) instead of the BLOB columns the fallback would produce. Everything else — where to import it, the alembic/env.py requirement, the DDL self-check — is exactly as in the SQL Server walkthrough above.