Skip to content

Adapter conformance🔗

pydapper.testing.adapter_conformance is a reusable, typed conformance suite for adapter command classes. It ships inside the installed pydapper distribution, so first-party and third-party adapter authors run the same contract without copying tests out of the pydapper repository. The helper uses only the Python standard library and pydapper core: it does not require pytest or any other test framework, it never imports an optional database driver, and importing it starts no service, container, or emulator. It is not exported from the pydapper package root; import it explicitly:

from pydapper.testing.adapter_conformance import run_core_sync

Mandatory profiles: core-sync and core-async🔗

core-sync and core-async are the stable identifiers of the two mandatory profiles.

  • Every registration that provides a sync command class promises core-sync.
  • Every registration that provides an async command class promises core-async.

The two profiles are independent. An adapter registered for both modes must pass both; passing sync conformance says nothing about async conformance, and a sync-only harness is never asked for async fields (or vice versa). Mode support is represented by which command classes a registration supplies — these mandatory profiles are not AdapterCapability flags.

Each profile covers registration and selection, the #541 cursor lifecycle (cleanup on success and active failure, error precedence, truthy-exit non-suppression, plain and context-manager cursors, unbuffered exhaustion and explicit close()/aclose()), parameter handling and execution, row mapping and RawRow mappers, zero/one/many cardinality, scalar semantics (no row vs SQL NULL vs falsey values), command options, preparation-hook ordering, the one-statement-per-call guard, and capability honesty. Hook ordering is checked on every distinct command path that owns a cursor — execute, buffered query, unbuffered query, query_first/query_single/execute_scalar, and query_multiple — because each opens its own cursor block: _prepare_cursor must run exactly once per command-owned cursor and _prepare_command exactly once per executed handler, both on the entered cursor and both before that handler reaches the driver (query_multiple therefore prepares one cursor and N handlers). These cases constrain preparation and how often a statement runs — every single-statement path must execute exactly once — but not fetching: how an adapter drains an unbuffered result set (one fetchone per row, fetchmany batches, a driver-side iterator) is its own choice and no hook-ordering case asserts it.

Cases are either live (they run through your real database resources) or instrumented (they wrap your real concrete command class around framework-owned recording and fault-injection connections, so cursor counts, call order, exception precedence, and "fails before driver work" guarantees are observed directly rather than inferred from return values).

Four of these mandatory cases certify the one-statement-per-call guard. They are core cases, not a capability profile: every adapter gets them and none can opt out. The two accept cases are deliberately instrumented rather than live, because whether a server tolerates a trailing ; is dialect-specific (Oracle rejects it) — they pin that pydapper does not refuse the SQL, never that the database accepts it.

Case id Kind Pins
sql.multi-statement-rejected live every command entry point raises MultipleStatementsError, and the connection is still usable for a normal query afterwards
sql.multi-statement-before-driver-work instrumented the raise happens with zero recorded driver events — no cursor acquired, nothing executed
sql.trailing-semicolon-accepted instrumented one trailing ; plus trailing whitespace and comments is not multi-statement, and the SQL reaches the driver unchanged
sql.separator-in-text-accepted instrumented a ; inside single-quoted text, double-quoted text, a -- comment, or a /* */ comment does not trip the guard and reaches the driver unchanged

Optional capability profiles🔗

Optional behaviors are independent profiles keyed to pydapper.AdapterCapability members — there is no linear "better adapter" tier, and no adapter is required to support any optional capability. The production catalog is returned by capability_profiles() and currently ships one profile: transactions, covering the transaction APIs in both modes. The runners automatically append the profile cases for every capability the command class under test declares, so a declaring adapter is exercised against its capability profiles in the same run_core_sync() / run_core_async() call as the core inventory — there is no separate capability runner to invoke. The framework enforces honesty in both directions:

  • A command class that declares a valid capability with no populated profile for its mode fails conformance (capabilities.declared-profile-populated). An unimplemented capability can never appear covered; zero-case profiles are rejected outright.
  • A declaration that is not a frozenset of AdapterCapability members fails conformance (capabilities.declaration-valid).
  • supports() must exactly match the class declaration (capabilities.supports-matches-declaration), and valid non-default CommandOptions for undeclared capabilities must raise UnsupportedFeatureError before cursor acquisition or driver work (options.unsupported-raises, options.unsupported-before-driver-work).

Native database support alone is never enough to declare a capability: a capability describes behavior the command class actually implements, tests through its conformance profile, and documents. Until then, the feature must fail loudly rather than silently execute as if supported.

The transactions profile🔗

These 11 cases per mode pin the commit() / rollback() / transaction() contract for every declaring adapter — the sync and async inventories share the same case ids, order, kinds, and descriptions, so both modes are held to one contract (the async cases exercise async with commands.transaction(): and the awaited coroutine twins). Every live case first commits the harness baseline (create_commands() then commit()), so scenarios start from a durable, transaction-free state even when a harness seeds inside an open transaction.

Case id Kind Pins
transactions.commit-persists live commit() makes a write durable; a later rollback() cannot remove it
transactions.rollback-discards live rollback() discards uncommitted work, committed rows stay intact
transactions.context-commits-on-exit live a transaction() block commits on clean exit
transactions.context-rolls-back-on-error live a raising block rolls back and re-raises the same exception object
transactions.commit-delegates-once instrumented one connection commit, no cursor acquired, returns None, failures propagate unchanged
transactions.rollback-delegates-once instrumented one connection rollback, no cursor acquired, returns None, failures propagate unchanged
transactions.context-lifecycle-order instrumented the block's commit comes after the command's full cursor lifecycle
transactions.context-error-precedence instrumented rollback on error, the block's error wins over a rollback failure, and a failed rollback never commits
transactions.commit-failure-propagates instrumented a failed exit commit propagates unchanged with no rollback attempt
transactions.context-base-exception-rollback instrumented a BaseException that is not an Exception rolls back and propagates unchanged too, and an interrupt raised by the rollback itself wins over the block's ordinary error, which survives as its __context__
transactions.context-not-reentrant instrumented nesting blocks on one instance raises RuntimeError (rolling the outer block back), and the guard clears so the next sequential block still commits

The harness API🔗

A harness supplies factories, isolation, and narrow dialect knobs. It never owns assertions: every behavioral check belongs to the framework runner, and harness- or adapter-provided code has no API to declare a case passed.

Sync adapters subclass SyncAdapterHarness; async adapters subclass AsyncAdapterHarness.

Every configuration field below may be declared on the subclass or assigned per instance — none of them is a ClassVar, so assigning to self in __init__ type checks without a type: ignore. That matters for the common case of a value that does not exist until a fixture has run, such as a DSN that is only known once a test container has started:

class MyHarness(SyncAdapterHarness):
    adapter_name = "acmedb"
    command_class = AcmeDbCommands

    def __init__(self, container):
        self.connect_dsn = f"acmedb://{container.host}:{container.port}/conformance"

A field that is genuinely static (table_name, column_case, strict_rowcounts, …) is still clearest as a class attribute; both forms are supported and mix freely on one harness.

Field Required Meaning
adapter_name yes The registered adapter name (exact, case-sensitive).
command_class yes The declared concrete Commands / CommandsAsync subclass for the mode.
connect_dsn yes A DSN that safely reaches an isolated test database through connect() / connect_async().
create_commands() yes Return a fresh command instance over a fresh connection with the canonical dataset freshly seeded. Called once per case.
teardown_commands(commands) yes Release everything create_commands() built. Runs after success and failure.
connect_kwargs no Extra kwargs for the connect() path (for example BigQuery's client=).
table_name no Name (optionally schema/dataset qualified) of the conformance table. Default pydapper_conformance.
column_case no "lower" (default) or "upper" for dialects that fold unquoted identifiers (Oracle).
supports_empty_strings no False only when the database cannot store '' distinctly from NULL (Oracle); the empty-string dataset value then seeds as the documented fallback "blank".
strict_rowcounts no False only when DML rowcounts are documented as unreliable (BigQuery emulator). Rowcount cases still run and must return an int; only exact equality is relaxed.
sql_overrides no Statement-id → SQL overrides for the narrow cases where portable SQL is insufficient (still pydapper ?name? placeholders).
recover_after_error(commands) no Backend recovery needed after an intentionally failed command (default no-op).
cursor_factory_style async only "awaitable" (default: connection.cursor() returns an awaitable, the CommandsAsync base contract) or "synchronous" (the psycopg3-async shape: cursor() returns the cursor directly and the command class normalizes it). Instrumented cases use this to exercise your real cursor() override.

The framework owns the canonical dataset and every query it runs. CONFORMANCE_COLUMNS is ("id", "label", "score", "note") and seed_rows(supports_empty_strings) returns the three canonical rows — including a SQL NULL note, a falsey 0 score, and an empty-string label — that create_commands() must seed. All query and DML statements are written in pydapper's portable ?name? placeholder syntax, so they run unchanged on every adapter; the harness owns only DDL, seeding, and connection construction.

Every value the framework itself binds as a parameter is non-NULL. Some drivers — BigQuery's DBAPI, for example — derive a parameter's type from its Python value and reject an untyped None, so binding NULL is not part of the mandatory core profiles. The canonical dataset still contains a real SQL NULL note and scalar.null-returns-none still asserts that a stored NULL reads back as None; producing that NULL is the harness's job, and a harness whose driver cannot bind None may seed it any way its dialect allows (the BigQuery harness binds a sentinel and then clears it with a literal UPDATE ... SET note = NULL).

A harness that does not supply a field a case needs fails that case with a structured HarnessDefinitionError carrying the profile id, the case id, and the missing field name — a missing harness field never silently skips or weakens a core case. There is deliberately no "skip this core case" option; driver limitations are expressed only through the narrow, documented knobs above.

Where to seed: under the adapter or through it🔗

create_commands() must return a freshly seeded dataset, and there are two reasonable ways to produce it. Both are used in this repository, so it is worth choosing deliberately:

  • Driver-level (recommended default). Insert the rows with the raw driver, underneath the adapter — the connection.executemany(...) in both examples below, and what pydapper's own sqlite3 harness does. Setup then shares no code path with the thing under test, so a broken seed cannot be confused with broken adapter behavior, and reading the harness tells you exactly what is in the table.
  • Through the adapter. Insert the rows with commands.execute(...) and pydapper's portable ?name? placeholders, as pydapper's service-backed harnesses do through the shared tests/conformance_support.py::seed_through_adapter helper. One seeding statement then works unchanged across every dialect, which is why harnesses for many backends share it; the cost is that setup runs through the adapter it is about to judge.

Prefer driver-level seeding unless you are maintaining harnesses for several dialects at once and want a single portable seeding path. This is a readability and blast-radius preference, not a correctness one: failures inside create_commands() are attributed to the harness either way (see below), so seeding through the adapter no longer risks a broken seed being reported as adapter misbehavior.

Harness setup failures are attributed to the harness🔗

If the harness's own create_commands() raises — a connection factory that cannot connect, a CREATE TABLE against the wrong dataset, a seeding statement the driver rejects — the resulting CaseResult is failed with harness_setup_failed=True and a message that names create_commands() as the culprit, while cause remains the original exception object. ConformanceReport.harness_setup_failed is True if any case in the run has it, and ConformanceFailureError's summary gains a [HARNESS SETUP FAILED: n of m failure(s) …] segment, so the traceback CI prints already says the harness never got off the ground.

This exists because one broken setup call fails every case that needs a fixture: without attribution, a single bad seeding statement reads as dozens of identical behavioral failures against a perfectly good adapter. A run with harness_setup_failed set is not a verdict on the adapter — fix the harness and run again.

The boundaries are narrow and deliberate:

  • Only the harness's own create_commands() call is wrapped, never the framework's validation or bookkeeping around it, and never the adapter behavior a case exercises afterwards. Genuine behavioral failures keep their existing shape and report harness_setup_failed=False.
  • KeyboardInterrupt, SystemExit, and asyncio.CancelledError still propagate unconverted; cancellation and shutdown are never turned into case results.
  • A harness that omits a required field or override is still a HarnessDefinitionError with its missing_field — a missing override is a definition error, not a setup failure, and reporting for it is unchanged.
  • The internal exception type used to carry this across the runner boundary is private. Branch on the two structured attributes (CaseResult.harness_setup_failed, ConformanceReport.harness_setup_failed), not on an exception class and not on message text.

Case isolation and cleanup🔗

Every case gets a fresh command instance, connection, and dataset through create_commands(); case ordering and earlier failures cannot make later cases pass. teardown_commands() runs after every case, on success and on failure. If both a case and its teardown fail, the case failure is preserved and the teardown failure is retained as structured secondary information on the result (CaseResult.cleanup_error); a teardown failure after an otherwise passing case fails that case.

Deterministic, structured results🔗

run_core_sync() / run_core_async() return a ConformanceReport whose results follow the declared profile/case order exactly, run after run. Each CaseResult carries the profile id, case id, pass/fail, a human-readable message, the original cause exception when one exists, the missing_field for harness validation failures, any cleanup_error, and harness_setup_failed. report.raise_for_failures() raises a ConformanceFailureError that exposes the same structured failures — nothing requires parsing exception prose.

Two flags that make a run mean less than it looks🔗

report.passed answers "did every case that ran pass?" and nothing more. Two separate booleans say whether the run was a conformance run at all, and both are structured attributes so no caller has to read messages:

Flag True means What it invalidates
report.covers_full_inventory every planned case ran (the default; False only after a case_ids filter) a passing run that covered a subset is not conformance
report.harness_setup_failed at least one case failed inside the harness's own create_commands() the failures describe your harness, not the adapter

Anything that claims conformance — a CI gate, a published matrix row, a release note — must therefore assert report.passed and report.covers_full_inventory. raise_for_failures() deliberately ignores completeness: it is about failures only, so a filtered run that passes everything it ran raises nothing.

report = run_core_sync(MyAdapterHarness())
assert report.passed and report.covers_full_inventory

Running one adapter🔗

from pydapper.testing.adapter_conformance import run_core_sync

report = run_core_sync(MyAdapterHarness())
for failure in report.failures:
    print(failure.profile_id, failure.case_id, failure.message)
report.raise_for_failures()

run_core_async is a plain awaitable and is awaited inside your own event loop — the framework never calls asyncio.run() itself:

report = await run_core_async(MyAsyncAdapterHarness())

Debugging one case with case_ids🔗

Both runners accept an optional case_ids collection that narrows the run to the named cases. It exists purely to shorten the debug loop: against a container-backed backend, a full profile costs minutes per iteration — mostly the client's connection retry backoff — which is a miserable way to chase one failing case.

report = run_core_sync(MyAdapterHarness(), case_ids=["rows.query-buffered", "scalar.null-returns-none"])
assert report.covers_full_inventory is False  # by construction: this is a debug run, not a conformance run

The rules are chosen so a filtered run can never be mistaken for a real one:

  • A filtered run is never a conformance result. covers_full_inventory is False whenever a filter narrowed the run, and ConformanceFailureError's summary gains a [PARTIAL RUN: …] segment. Naming every planned case is still full coverage, and case_ids=None (the default) runs the full inventory exactly as before.
  • Unknown ids fail loudly, before anything runs. A typo raises CaseSelectionError — carrying profile_id, requested_case_ids, and unknown_case_ids — rather than quietly running fewer cases. An empty selection raises the same error with an empty unknown_case_ids, because "no cases ran and none failed" is not a pass.
  • A bare string is a TypeError. case_ids="rows.query-buffered" would otherwise be a collection of characters; pass a list.
  • Ids come from core_sync_profile().sync_cases / core_async_profile().async_cases (plus any declared-capability profile cases, which are planned before filtering and so can also be selected). Duplicates are de-duplicated, and results still follow declared case order, not the order you requested.

Complete third-party sync example🔗

This one computes connect_dsn per instance in __init__ — the shape a harness needs when the DSN only exists after a fixture has run — and seeds at the driver level.

"""A complete third-party sync-only adapter running the reusable conformance suite.

The "acmedb" driver below is stdlib sqlite3 wearing a third-party hat: the point is
the shape of the harness, not the database. No pytest is required.
"""

import sqlite3
import tempfile

import pydapper
from pydapper.commands import BaseSqlParamHandler
from pydapper.commands import Commands
from pydapper.testing.adapter_conformance import SyncAdapterHarness
from pydapper.testing.adapter_conformance import run_core_sync
from pydapper.testing.adapter_conformance import seed_rows


class AcmeDbCommands(Commands):
    class SqlParamHandler(BaseSqlParamHandler):
        def get_param_placeholder(self, param_name: str) -> str:
            return "?"

    @classmethod
    def connect(cls, parsed_dsn, **connect_kwargs) -> "AcmeDbCommands":
        return cls(sqlite3.connect(parsed_dsn.database, **connect_kwargs))


pydapper.register_adapter(
    "acmedb",
    commands=AcmeDbCommands,
    using_connection_predicate=lambda connection: isinstance(connection, sqlite3.Connection),
)


class AcmeDbConformanceHarness(SyncAdapterHarness):
    adapter_name = "acmedb"
    command_class = AcmeDbCommands

    def __init__(self) -> None:
        # no harness field is a ClassVar, so configuration a real harness only learns at
        # runtime -- a container's host and port, a temp directory -- is assigned on self
        workdir = tempfile.mkdtemp()
        # four slashes make the sqlite path absolute; the +acmedb suffix selects this adapter
        self.connect_dsn = f"sqlite+acmedb:///{workdir}/example.db"

    def create_commands(self) -> Commands:
        # a fresh, isolated database and dataset for every conformance case
        connection = sqlite3.connect(":memory:")
        connection.execute(
            "CREATE TABLE pydapper_conformance (id INTEGER PRIMARY KEY, label TEXT, score INTEGER, note TEXT)"
        )
        connection.executemany(
            "INSERT INTO pydapper_conformance (id, label, score, note) VALUES (?, ?, ?, ?)",
            seed_rows(self.supports_empty_strings),
        )
        connection.commit()
        return AcmeDbCommands(connection)

    def teardown_commands(self, commands: Commands) -> None:
        commands.connection.close()


report = run_core_sync(AcmeDbConformanceHarness())
for failure in report.failures:
    print(f"{failure.profile_id}/{failure.case_id}: {failure.message}")
print(f"{report.profile_id} for {report.adapter_name!r}: {len(report.results)} cases, passed={report.passed}")
report.raise_for_failures()
# claiming conformance takes both: every case passed *and* every planned case ran
assert report.passed and report.covers_full_inventory
# core-sync for 'acmedb': 58 cases, passed=True
(This script is complete, it should run "as is")

Complete third-party async example🔗

This one declares connect_dsn on the class instead, which is the simpler form when the value is known up front. Both forms are supported.

"""A complete third-party async-only adapter running the reusable conformance suite.

The "acmeaio" driver below is a tiny async facade over stdlib sqlite3, standing in
for a real async driver. The conformance API itself never calls ``asyncio.run()`` —
``run_core_async`` is awaited inside whatever event loop the caller owns, which here
is the one this script starts at the bottom.
"""

import asyncio
import sqlite3
import tempfile

import pydapper
from pydapper.commands import BaseSqlParamHandler
from pydapper.commands import CommandsAsync
from pydapper.testing.adapter_conformance import AsyncAdapterHarness
from pydapper.testing.adapter_conformance import run_core_async
from pydapper.testing.adapter_conformance import seed_rows

WORKDIR = tempfile.mkdtemp()


class AcmeAioCursor:
    """An async cursor facade over a sqlite3 cursor."""

    def __init__(self, cursor: sqlite3.Cursor) -> None:
        self._cursor = cursor

    @property
    def description(self):
        return self._cursor.description

    @property
    def rowcount(self) -> int:
        return self._cursor.rowcount

    async def execute(self, sql, parameters=None) -> None:
        if parameters is None:
            self._cursor.execute(sql)
        else:
            self._cursor.execute(sql, parameters)

    async def executemany(self, sql, seq_of_parameters=None) -> None:
        self._cursor.executemany(sql, seq_of_parameters or [])

    async def fetchone(self):
        return self._cursor.fetchone()

    async def fetchall(self):
        return self._cursor.fetchall()

    async def fetchmany(self, size=1):
        return self._cursor.fetchmany(size)

    async def close(self) -> None:
        self._cursor.close()


class AcmeAioConnection:
    """An async connection facade whose cursor() returns an awaitable."""

    def __init__(self, connection: sqlite3.Connection) -> None:
        self._connection = connection

    async def cursor(self, *args, **kwargs) -> AcmeAioCursor:
        return AcmeAioCursor(self._connection.cursor())

    def close(self) -> None:
        self._connection.close()


class AcmeAioCommands(CommandsAsync):
    class SqlParamHandler(BaseSqlParamHandler):
        def get_param_placeholder(self, param_name: str) -> str:
            return "?"

    @classmethod
    async def connect_async(cls, parsed_dsn, **connect_kwargs) -> "AcmeAioCommands":
        return cls(AcmeAioConnection(sqlite3.connect(parsed_dsn.database, **connect_kwargs)))


pydapper.register_adapter(
    "acmeaio",
    async_commands=AcmeAioCommands,
    using_connection_predicate=lambda connection: isinstance(connection, AcmeAioConnection),
)


class AcmeAioConformanceHarness(AsyncAdapterHarness):
    adapter_name = "acmeaio"
    command_class = AcmeAioCommands
    # four slashes make the sqlite path absolute; the +acmeaio suffix selects this adapter
    connect_dsn = f"sqlite+acmeaio:///{WORKDIR}/example.db"
    # our cursor() returns an awaitable, the CommandsAsync base contract; a driver whose
    # cursor() returns the cursor synchronously (like psycopg3 async) declares "synchronous"
    cursor_factory_style = "awaitable"

    async def create_commands(self) -> CommandsAsync:
        connection = sqlite3.connect(":memory:")
        connection.execute(
            "CREATE TABLE pydapper_conformance (id INTEGER PRIMARY KEY, label TEXT, score INTEGER, note TEXT)"
        )
        connection.executemany(
            "INSERT INTO pydapper_conformance (id, label, score, note) VALUES (?, ?, ?, ?)",
            seed_rows(self.supports_empty_strings),
        )
        connection.commit()
        return AcmeAioCommands(AcmeAioConnection(connection))

    async def teardown_commands(self, commands: CommandsAsync) -> None:
        commands.connection.close()


async def main() -> None:
    report = await run_core_async(AcmeAioConformanceHarness())
    for failure in report.failures:
        print(f"{failure.profile_id}/{failure.case_id}: {failure.message}")
    print(f"{report.profile_id} for {report.adapter_name!r}: {len(report.results)} cases, passed={report.passed}")
    report.raise_for_failures()


asyncio.run(main())
# core-async for 'acmeaio': 58 cases, passed=True
(This script is complete, it should run "as is")

Reporting service-backed runs you could not execute🔗

Service-free evidence (mock harnesses and instrumented concrete-class cases) and live service-backed evidence are different things, and reports must keep them distinct. A live suite that was skipped, or merely collected, is a test-run status — never a pass. When you cannot run a live suite, report:

  • whether the optional database driver was installed;
  • whether the service or emulator was reachable;
  • whether credentials were required and available; and
  • which service-free conformance cases cover the same shared behavior.

First-party driver conformance matrix🔗

One row per first-party command class. "Verification" describes what is actually executed where — service-free coverage runs everywhere, live suites run only under their backend marker when their Docker/testcontainers service or emulator is available.

Adapter name Command class Mode Declared capabilities Conformance entry Verification Driver limits
sqlite3 Sqlite3Commands sync transactions tests/test_sqlite/test_conformance.py live, service-free (runs in the core and sqlite suites) SQLite; the connection context manager commits on clean exit and rolls back on error but never closes, and legacy isolation_level opens implicit transactions on DML only (DDL issued outside an open transaction is autocommitted)
psycopg2 Psycopg2Commands sync transactions tests/test_postgresql/test_conformance.py service-free instrumented + mock coverage; live suite under the postgresql marker when Docker is available PostgreSQL; the connection context manager commits on clean exit and rolls back on error but never closes the connection
psycopg Psycopg3Commands sync transactions tests/test_postgresql/test_conformance.py service-free instrumented + mock coverage; live suite under the postgresql marker when Docker is available PostgreSQL; the connection context manager commits on clean exit, rolls back on error, and then closes the connection
psycopg Psycopg3CommandsAsync async transactions tests/test_postgresql/test_conformance.py service-free instrumented + mock coverage (including its synchronous cursor-factory normalization); live suite under the postgresql marker when Docker is available PostgreSQL; the connection context manager commits on clean exit, rolls back on error, and then closes the connection
aiopg AiopgCommands async (none) tests/test_postgresql/test_conformance.py service-free instrumented + mock coverage; live suite under the postgresql marker when Docker is available PostgreSQL; aiopg is autocommit-only and emulates executemany by looping execute; its connection-level commit/rollback raise, so transactions is not declared even though the async APIs exist; its async context manager closes on exit (everything was already durable)
mysql MySqlConnectorPythonCommands sync transactions tests/test_mysql/test_conformance.py service-free instrumented + mock coverage (including its query_first unread-result drain); live suite under the mysql marker when Docker is available MySQL; unread results must be drained before a cursor closes; the connection context manager only closes — exit never commits, uncommitted DML is discarded, and MySQL implicitly commits DDL statements (temporary tables excepted); the driver enables CLIENT_MULTI_STATEMENTS by default, so pydapper clears it at connect time on connections it opens
pymssql PymssqlCommands sync transactions tests/test_mssql/test_conformance.py service-free instrumented + mock coverage; live suite under the mssql marker when Docker is available Microsoft SQL Server; non-autocommit connections hold an always-open BEGIN TRAN, the context manager only closes, and close() implicitly rolls back all uncommitted work
oracledb OracledbCommands sync transactions tests/test_oracle/test_conformance.py service-free instrumented + mock coverage; live suite under the oracle marker when Docker is available Oracle; unquoted identifiers fold to uppercase (column_case="upper"), and '' is stored as NULL (supports_empty_strings=False); the connection context manager rolls back uncommitted work and closes — exit never commits — and Oracle implicitly commits DDL statements
google GoogleBigqueryClientCommands sync (none) tests/test_bigquery/test_conformance.py service-free instrumented + mock coverage; live suite under the bigquery marker when the emulator is available Google BigQuery; the emulator's DML rowcounts are unreliable (strict_rowcounts=False), and the DBAPI cannot bind an untyped None, so the harness seeds the canonical SQL NULL note through a sentinel and a literal UPDATE; the DBAPI's commit() is a no-op and there is no rollback(), so transactions is not declared; the DBAPI connection has no context manager, so with-block exit performs no driver call

Extending the suite in future capability work🔗

Every PR that adds or changes an optional capability must, in the same change:

  • update the AdapterCapability enum/declaration when necessary;
  • add a real, reusable capability profile to the production catalog (zero-case profiles are rejected, so an empty placeholder cannot ship);
  • test at least one supported and one unsupported adapter against the new profile;
  • update the affected first-party capability declaration;
  • update driver limitations and documentation, including the matrix above; and
  • update the type tests when public typing changes.

This keeps supports(), the declarations, the profiles, and the documentation honest together: a capability is declared only when its implementation, its profile, and its documentation all exist.