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
frozensetofAdapterCapabilitymembers fails conformance (capabilities.declaration-valid). supports()must exactly match the class declaration (capabilities.supports-matches-declaration), and valid non-defaultCommandOptionsfor undeclared capabilities must raiseUnsupportedFeatureErrorbefore 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 sharedtests/conformance_support.py::seed_through_adapterhelper. 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 reportharness_setup_failed=False. KeyboardInterrupt,SystemExit, andasyncio.CancelledErrorstill propagate unconverted; cancellation and shutdown are never turned into case results.- A harness that omits a required field or override is still a
HarnessDefinitionErrorwith itsmissing_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_inventoryisFalsewhenever a filter narrowed the run, andConformanceFailureError's summary gains a[PARTIAL RUN: …]segment. Naming every planned case is still full coverage, andcase_ids=None(the default) runs the full inventory exactly as before. - Unknown ids fail loudly, before anything runs. A typo raises
CaseSelectionError— carryingprofile_id,requested_case_ids, andunknown_case_ids— rather than quietly running fewer cases. An empty selection raises the same error with an emptyunknown_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
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
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
AdapterCapabilityenum/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.