deev.common#

class ConnectionString(connection_str=None)#

Bases: object

A type-safe “Connection String” representation that can parse/build constituent parts.

property server: str | None#

property database: str | None#

property user: str | None#

property password: str | None#

property provider: str | None#

property connect_timeout: int | None#

property command_timeout: int | None#

parse(connectionstring)#

Return type:: ConnectionString

class DbConnection(*args, **kwargs)#

Bases: Protocol

DB-API 2.0 Connection proto.

cursor(*args, **kwargs)#

Return type:: DbCursor

commit()#

Return type:: None

rollback()#

Return type:: None

close()#

Return type:: None

class DbCursor(*args, **kwargs)#

Bases: Protocol

DB-API 2.0 Cursor proto.

property description: Sequence[tuple[Any, Any, int | None, int | None, int | None, int | None, bool]] | None#

property rowcount: int#

execute(operation, params=Ellipsis)#

Return type:: None

executemany(operation, seq_of_params)#

Return type:: None

fetchone()#

Return type:: tuple[Any, ...] | None

fetchmany(size=Ellipsis)#

Return type:: list[tuple[Any, ...]]

fetchall()#

Return type:: list[tuple[Any, ...]]

close()#

Return type:: None

final exception DbError(reason)#

Bases: Exception

Exception raised when a DB operation fails.

class DbMigrator(connectionstring)#

Bases: object

Performs database changes from a set of “migration scripts.”

Migration scripts are scanned from a filesystem directory.

apply(migrations_path, stop_at=None)#

Return type:: None

undo(migrations_path, stop_at=None)#

Return type:: None

class DbTableAdapter(*args, **kwargs)#

Bases: Protocol[TEntity]

create_table()#

Utility method for creating the target table.

Return type:: None

commit()#

Return type:: None

rollback()#

Return type:: None

create(entity)#

Creates a new record in the specified table with the provided keyword arguments.

Parameters:: kwargs (dict[str, Any]) – A dictionary containing the column names and their corresponding values for the new record.
Returns:: The newly created record’s ID if successful, otherwise None.
Return type:: dict[str, Any]

read(**kwargs)#

Reads a record from the specified table with the primary key represented by kwargs.

Parameters:: kwargs (Any) – The primary key.
Returns:: A dictionary containing the column names and their corresponding values for the specified record, or None if no such record exists.
Return type:: Optional[TypeVar(TEntity)]

update(entity)#

Return type:: None

delete(**kwargs)#

Return type:: None

exists(**kwargs)#

Return type:: bool

upsert(entity)#

Return type:: dict[str, Any]

query(where=Ellipsis, params=Ellipsis, orderby=Ellipsis, limit=Ellipsis)#

Return type:: Generator[TypeVar(TEntity), None, None]

class DbTransactionContext(connection)#

Bases: Protocol

property connection: DbConnection#

cursor()#

Return type:: DbCursor

commit()#

Return type:: None

execute(sql, params=Ellipsis)#

Return type:: DbCursor

execute_script(sql)#

Return type:: None

execute_nonquery(sql, params=Ellipsis)#

Return type:: None

execute_reader(sql, params=Ellipsis)#

Return type:: Generator[tuple[Any, ...], None, None]

execute_scalar(sql, params=Ellipsis)#

Return type:: Any

rollback()#

Return type:: None

class DbTypeMapper(*args, **kwargs)#

Bases: Protocol

get_provider_type(field_name)#

Get the SQL type (string) needed to represent an entity field in the underlying table.

Parameters:: field_spec – The “Entity Field Spec”.
Return type:: str
Returns:: The SQL type string.