"""PEP 249 database adapter protocols using structural subtyping (PEP 544).""" from __future__ import annotations from enum import IntEnum from typing import Any, ContextManager, Protocol, Sequence, runtime_checkable class ColumnDescriptionIndex(IntEnum): """Index positions for column description tuple fields.""" NAME = 0 TYPE_CODE = 1 DISPLAY_SIZE = 2 INTERNAL_SIZE = 3 PRECISION = 4 SCALE = 5 NULL_OK = 6 ColumnDescription = tuple[ str, # name: column name Any, # type_code: database-specific type code int | None, # display_size: display size (may be None) int | None, # internal_size: internal size (may be None) int | None, # precision: numeric precision (may be None) int | None, # scale: numeric scale (may be None) bool | None, # null_ok: whether NULL values are allowed (may be None) ] @runtime_checkable class Cursor(Protocol): """PEP 249 Cursor Object Protocol.""" @property def description(self) -> Sequence[ColumnDescription] | None: """Read-only attribute providing the column descriptions. Returns a sequence of 7-item sequences describing each result column: - name: str - Column name - type_code: Any - Database-specific type code - display_size: int | None - Display size - internal_size: int | None - Internal size - precision: int | None - Numeric precision - scale: int | None - Numeric scale - null_ok: bool | None - Whether NULL values are allowed Returns None if the cursor has not executed an operation that produces a result set. """ ... @property def rowcount(self) -> int | None: """Read-only attribute specifying the number of rows affected by the last operation.""" ... @property def lastrowid(self) -> int: """ID generated by the last INSERT statement.""" ... def execute(self, sql: str, params: Any = (), /) -> Any: """Prepare and execute a database operation (query or command).""" ... def fetchall(self) -> list[dict[str, Any]]: """Fetch all (remaining) rows of a query result.""" ... def fetchmany(self, size: int) -> list[dict[str, Any]]: """Fetch the next set of rows of a query result.""" ... def fetchone(self) -> dict[str, Any] | None: """Fetch the next row of a query result.""" ... def close(self) -> Any: """Close the cursor now.""" ... @runtime_checkable class Connection(Protocol): """PEP 249 Connection Object Protocol.""" def close(self) -> None: """Close the connection now.""" ... def commit(self) -> None: """Commit the current transaction to the database.""" ... def cursor(self) -> ContextManager[Cursor]: """Return a new Cursor Object using the connection.""" ... def rollback(self) -> None: """Roll back the current transaction, discarding changes.""" ...