API Reference

The following page outlines the donphan API.

Connection Helpers

await donphan.create_pool(dsn, codecs={}, *, set_as_default=False, **kwargs)

This function is a coroutine.

Creates the database connection pool.

Parameters:
  • dsn (str) – A database connection string.

  • codecs (Dict[str, TypeCodec]) – A mapping of type to encoder and decoder for custom type codecs. A pre-defined set of codecs is provided in TYPE_CODECS is used by default. As well as a set of OPTIONAL_CODECS is provided.

  • set_as_default (bool) – Sets whether the pool should be used as the default connection pool for database operations. Defaults to False.

  • **kwargs (Any) – Extra keyword arguments to pass to asyncpg.create_pool

Returns:

The new pool which was created.

Return type:

asyncpg.Pool

class donphan.MaybeAcquire(connection=None, /, *, pool=None)

Async helper for acquiring a connection to the database.

async with x as c

Yields a database connection to use, if a new one was created it will be closed on exit.

Parameters:
  • connection (Optional[asyncpg.Connection]) – A database connection to use. If none is supplied a connection will be acquired from the pool.

  • pool (Optional[asyncpg.Pool]) – A connection pool to use.

connection

The supplied database connection, if provided.

Type:

Optional[asyncpg.Connection]

pool

The connection pool used to acquire new connections.

Type:

Optional[asyncpg.Pool]

Type Codecs

class donphan.TYPE_CODECS

A dictionary of pre-defined custom type-codecs.

property json

Automatically conveters a dictionary to and from json data to be stored in the database.

property jsonb

Automatically conveters a dictionary to and from json data to be stored in the database.

class donphan.OPTIONAL_CODECS

A dictionary of optional custom type-codecs.

property timestamp

Ensures all timestamps are timezone aware relative to UTC.

Database Creation Helpers

await donphan.create_db(connection, if_not_exists=True, automatic_migrations=False, with_transaction=True)

This function is a coroutine.

A helper function to create all objects in the database.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • if_not_exists (bool) – Sets whether creation should continue if the object already exists. Defaults to True.

  • automatic_migrations (bool) – Sets whether automatic migrations should be run. Defaults to False.

  • with_transaction (bool) – Sets whether the database should be wrapped in a transaction. Defaults to True.

SQL Types

SQLType

class donphan.SQLType

A representation of an SQL type

py_type

The python type associated with the column.

Type:

Type[Any]

sql_type

The SQL type associated with the column.

Type:

str

Supported Types

The following types are supported by donphan.

Numeric Types

SQLType.Integer

Represents the SQL INTEGER type. Python class int, can be used as a substitute.

SQLType.SmallInt

Represents the SQL SMALLINT type.

SQLType.BigInt

Represents the SQL BIGINT type.

SQLType.Serial

Represents the SQL SERIAL type.

SQLType.Float

Represents the SQL FLOAT type. Python class float, can be used as a substitute.

SQLType.DoublePrecision

Represents the SQL DOUBLE PRECISION type.

SQLType.Numeric

Represents the SQL NUMERIC type. Python class decimal.Decimal, can be used as a substitute.

Monetary Types

SQLType.Money

Represents the SQL MONEY type.

Character Types

SQLType.Character

Represents the SQL CHARACTER type.

SQLType.Text

Represents the SQL TEXT type. Python class str, can be used as a substitute.

Binary Type

SQLType.Bytea

Represents the SQL BYTEA type. Python class bytes, can be used as a substitute.

Date/Time Types

SQLType.Timestamp

Represents the SQL TIMESTAMP type. Python class datetime.datetime, can be used as a substitute.

SQLType.AwareTimestamp

Represents the SQL TIMESTAMP WITH TIME ZONE type. Python class datetime.datetime, can be used as a substitute.

SQLType.Date

Represents the SQL DATE type. Python class datetime.date, can be used as a substitute.

SQLType.Interval

Represents the SQL INTERVAL type. Python class datetime.timedelta, can be used as a substitute.

Boolean Type

SQLType.Boolean

Represents the SQL BOOLEAN type. Python class bool, can be used as a substitute.

Network Adress Types

SQLType.CIDR

Represents the SQL CIDR type.

SQLType.Inet

Represents the SQL INET type.

SQLType.MACAddr

Represents the SQL MACADDR type.

UUID Type

SQLType.UUID

Represents the SQL UUID type. Python class uuid.UUID, can be used as a substitute.

JSON Types

SQLType.JSON

Represents the SQL JSON type.

SQLType.JSONB

Represents the SQL JSONB type. Python class dict, can be used as a substitute.

Database Objects

Below are types which are used to interact with the dtabase.

BaseColumn

class donphan.BaseColumn

A Base class which all Column types inherit.

name

The column’s name.

Type:

str

inner_join(other)

A shortcut for creating an INNER JOIN between two columns.

Parameters:

other (BaseColumn) – The other column to join to.

Returns:

A representation of a join of which fetch methods can be applied to.

Return type:

Join

left_join(other)

A shortcut for creating an LEFT JOIN between two columns.

Parameters:

other (BaseColumn) – The other column to join to.

Returns:

A representation of a join of which fetch methods can be applied to.

Return type:

Join

right_join(other)

A shortcut for creating an RIGHT JOIN between two columns.

Parameters:

other (BaseColumn) – The other column to join to.

Returns:

A representation of a join of which fetch methods can be applied to.

Return type:

Join

full_outer_join(other)

A shortcut for creating an FULL OUTER JOIN between two columns.

Parameters:

other (BaseColumn) – The other column to join to.

Returns:

A representation of a join of which fetch methods can be applied to.

Return type:

Join

Column

class donphan.Column(primary_key=False, index=False, nullable=True, unique=False, cascade=False, default=MISSING, references=None)

A representation of a SQL Database Table Column

Parameters:
  • primary_key (bool) – Whether the column is a primary key column, defaults to False.

  • index (bool) – Whether to create an index for the given column, defaults to False.

  • nullable (bool) – Whether the column is nullable.

  • unique (bool) – Whether the column has a unique constraint.

  • default (Any) – The default value of the column.

  • references (Optional[Column]) – The column which this column references.

  • cascade (bool) – Whether deletions / updates the referenced column should cascace. Defaults to False.

name

The column’s name.

Type:

str

table

The Table the column is a part of.

Type:

Table

py_type

The python type associated with the column.

Type:

Type

sql_type

The SQL type associated with the column.

Type:

Type[SQLType]

primary_key

Whether the column is a primary key column.

Type:

bool

index

Whether the column is indexed.

Type:

bool

nullable

Whether the column is nullable.

Type:

bool

unique

Whether the column has a unique constraint.

Type:

bool

default

The default value of the column.

Type:

Any

references

The column which this column references, if set.

Type:

Optional[Column]

cascade

Whether deletions / updates the referenced column should cascace.

Type:

bool

classmethod create(name, type, **options)

A shortcut for creating a column with a given type.

Parameters:
  • name (str) – The name of the column.

  • type (type[SQLType]) – The type of the column.

Table

class donphan.Table

Base class for creating representations of SQL Database Tables.

_name

The name of the table.

Type:

str

_schema

The tables schema.

Type:

str

_columns

The columns contained in the table.

Type:

Iterable[Column]

_columns_dict

A mapping between a column’s name and itself.

Type:

Dict[str, Column]

_primary_keys

The primary key columns of the table.

Type:

Iterable[Column]

classmethod await create(connection, *, if_not_exists=True, create_schema=True, automatic_migrations=False)

This function is a coroutine.

Creates this database object.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • if_not_exists (bool) – Sets whether creation should continue if the object already exists. Defaults to True.

  • create_schema (bool) – Sets whether the database schema should also be created. Defaults to True.

  • automatic_migrations (bool) – Sets whether migrations should be automatically run. Defaults to False.

classmethod await drop_column(connection, /, column)

Drops a column from the table.

Parameters:
  • connection (Connection) – The connection to use.

  • column (Column) – The column to drop.

classmethod await add_column(connection, /, column)

Adds a new column to the table.

Parameters:
  • connection (Connection) – The connection to use.

  • column (Column) – The column to add. This column must not be associated with another table.

classmethod await migrate(connection, /, columns, with_transaction=True)

This function is a coroutine.

Migrates the table to the given columns.

Parameters:
  • connection (Connection) – The connection to use.

  • columns (Iterable[Column]) – The new list of columns for the table.

  • with_transaction (bool) – Sets whether the database should be wrapped in a transaction. Defaults to True.

classmethod await migrate_to(connection, /, table, migration=MISSING, *, create_new_table=False, drop_table=False, with_transaction=True)

This function is a coroutine.

Helper function for migrating data in a table to another.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • table (Type[Table]) – The new table to migrate to.

  • migration (Callable[[asyncpg.Record], Dict[str, Any]]) – The function used to migrate data between tables. By default this passes the record unchanged.

  • create_new_table (bool) – Sets whether the table to migrate to should be created. Defaults to False.

  • drop_table (bool) – Sets whether this table should be dropped after migrating. Defaults to False.

  • with_transaction (bool) – Sets whether the database should be wrapped in a transaction. Defaults to True.

classmethod await create_all(connection, /, *, if_not_exists=True, create_schema=True, automatic_migrations=False, with_transaction=True)

This function is a coroutine.

Creates all subclasses of this database object.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • if_not_exists (bool) – Sets whether creation should continue if the object already exists. Defaults to True.

  • create_schema (bool) – Sets whether the database schema should also be created. Defaults to True.

  • automatic_migrations (bool) – Sets whether migrations should be automatically performed. Defaults to False.

  • with_transaction (bool) – Sets whether the database should be wrapped in a transaction. Defaults to True.

classmethod await create_schema(connection, *, if_not_exists=True)

This function is a coroutine.

Creates the schema this database object uses.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • if_not_exists (bool) – Sets whether creation should continue if the schema already exists. Defaults to True.

classmethod await delete(connection, /, returning=None, **values)

This function is a coroutine.

Deletes records in the database which contain the given values.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • **values (Any) – The column to value mapping to filter records with.

  • returning (Optional[Union[Iterable[Union[Column, str]], str]]) – An optional list of or string representing columns to return from the deleted records.

Returns:

A list of records containing information from the deleted records.

Return type:

Optional[List[asyncpg.Record]]

classmethod await delete_record(connection, /, record, *, returning=None)

This function is a coroutine.

Deletes a given record from the database.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • record (asyncpg.Record) – The record to delete.

  • returning (Optional[Union[Iterable[Union[Column, str]], str]]) – An optional list of or string representing columns to return from the deleted record.

Returns:

A record containing information from the deleted record.

Return type:

Optional[asyncpg.Record]

classmethod await delete_where(connection, /, where, *values, returning=None)

This function is a coroutine.

Deletes records in the database which match the given WHERE clause.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • where (str) – An SQL WHERE clause.

  • *values (Any) – Values to be substituted into the WHERE clause.

  • returning (Optional[Union[Iterable[Union[Column, str]], str]]) – An optional list of or string representing columns to return from the deleted records.

Returns:

A list of records containing information from the deleted records.

Return type:

Optional[List[asyncpg.Record]]

classmethod await drop(connection, /, *, if_exists=True, cascade=False)

This function is a coroutine.

Drops this object from the database.

Parameters:
  • if_exists (bool) – Sets whether dropping should not error if the object does not exist. Defaults to True.

  • cascade (bool) – Sets whether dropping should cascade. Defaults to False.

classmethod await exists(connection)

Check if the table exists.

Parameters:

conn – The connection to use.

Returns:

True if the table exists, False otherwise.

classmethod export(*, if_not_exists=False, export_schema=True, fp=None)

A function which exports this database object.

Parameters:
  • if_not_exists (bool) – Sets whether the if_not_exists clause should be set on exported objects. Defaults to False.

  • export_schema (bool) – Sets whether to additionally export the schema used by this object. Defaults to True.

  • fp (Optional[os.PathLike, io.TextIOBase]) –

    A file-like object opened in text mode and write mode. or a filename representing a file on disk to write to.

    Note

    If the file-like object passed is opened via open() ensure the object is in a text-writing mode such as "w".

Returns:

The file-like object which was provided or a string containing the exported database object.

Return type:

Union[io.TextIOBase, str]

classmethod export_all(*, if_not_exists=False, export_schema=True, fp=None)

A function which exports all database objects that subclass this object.

Parameters:
  • if_not_exists (bool) – Sets whether the if_not_exists clause should be set on exported objects. Defaults to False.

  • export_schema (bool) – Sets whether to additionally export any schemas used by these objects. Defaults to True.

  • fp (Optional[os.PathLike, io.TextIOBase]) –

    A file-like object opened in text mode and write mode. or a filename representing a file on disk to write to.

    Note

    If the file-like object passed is opened via open() ensure the object is in a text-writing mode such as "w".

Returns:

The file-like object which was provided or a string containing the exported database objects.

Return type:

Union[io.TextIOBase, str]

classmethod export_schema(*, if_not_exists=False, fp=None)

This function is a coroutine.

A function which exports this database object’s schema.

Parameters:
  • if_not_exists (bool) – Sets whether the if_not_exists clause should be set on the exported database schema. Defaults to False.

  • fp (Optional[os.PathLike, io.TextIOBase]) –

    A file-like object opened in text mode and write mode. or a filename representing a file on disk to write to.

    Note

    If the file-like object passed is opened via open() ensure the object is in a text-writing mode such as "w".

Returns:

The file-like object which was provided or a string containing the exported database schema.

Return type:

Union[io.TextIOBase, str]

classmethod await fetch(connection, /, *, limit=None, order_by=None, **values)

This function is a coroutine.

Fetches records in the database which contain the given values.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • limit (Optional[int]) – If provided, sets the maxmimum number of records to be returned.

  • order_by (Optional[Union[Tuple[BaseColumn, Literal[“ASC”, “DESC”]], str]) – Sets the ORDER BY condition on the database query. Takes a tuple concisting of the column and direction, or a string defining the condition.

  • **values (Any) – The column to value mapping to filter records with.

Returns:

Records which which contain the given values.

Return type:

Iterable[asyncpg.Record]

classmethod await fetch_row(connection, /, *, order_by=None, **values)

This function is a coroutine.

Fetches a records in the database which contains the given values.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • order_by (Optional[Union[Tuple[BaseColumn, Literal[“ASC”, “DESC”]], str]) – Sets the ORDER BY condition on the database query. Takes a tuple concisting of the column and direction, or a string defining the condition.

  • **values (Any) – The column to value mapping to filter records with.

Returns:

A record which contains the given values if found.

Return type:

Optional[asyncpg.Record]

classmethod await fetch_row_where(connection, /, where, *values, order_by=None)

This function is a coroutine.

Fetches a record in the database which match a given WHERE clause.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • where (str) – An SQL WHERE clause.

  • *values (Any) – Values to be substituted into the WHERE clause.

  • order_by (Optional[Union[Tuple[BaseColumn, Literal[“ASC”, “DESC”]], str]) – Sets the ORDER BY condition on the database query. Takes a tuple concisting of the column and direction, or a string defining the condition.

Returns:

A record which matches the WHERE clause if found.

Return type:

Optional[asyncpg.Record]

classmethod await fetch_value(connection, /, column, *, order_by=None, **values)

This function is a coroutine.

Fetches a records in the database which contains the given values.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • order_by (Optional[Union[Tuple[BaseColumn, Literal[“ASC”, “DESC”]], str]) – Sets the ORDER BY condition on the database query. Takes a tuple concisting of the column and direction, or a string defining the condition.

  • **values (Any) – The column to value mapping to filter records with.

Returns:

A record which contains the given values if found.

Return type:

Optional[asyncpg.Record]

classmethod await fetch_value_where(connection, /, column, where, *values, order_by=None)

This function is a coroutine.

Fetches a record in the database which match a given WHERE clause.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • where (str) – An SQL WHERE clause.

  • *values (Any) – Values to be substituted into the WHERE clause.

  • order_by (Optional[Union[Tuple[BaseColumn, Literal[“ASC”, “DESC”]], str]) – Sets the ORDER BY condition on the database query. Takes a tuple concisting of the column and direction, or a string defining the condition.

Returns:

A record which matches the WHERE clause if found.

Return type:

Optional[asyncpg.Record]

classmethod await fetch_where(connection, /, where, *values, limit=None, order_by=None)

This function is a coroutine.

Fetches records in the database which match a given WHERE clause.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • where (str) – An SQL WHERE clause.

  • *values (Any) – Values to be substituted into the WHERE clause.

  • limit (Optional[int]) – If provided, sets the maxmimum number of records to be returned.

  • order_by (Optional[Union[Tuple[BaseColumn, Literal[“ASC”, “DESC”]], str]) – Sets the ORDER BY condition on the database query. Takes a tuple concisting of the column and direction, or a string defining the condition.

Returns:

Records which match the WHERE clause.

Return type:

Iterable[asyncpg.Record]

classmethod full_outer_join(other, on)

A chainable method to join with another database object utilising a FULL OUTER JOIN.

Parameters:
  • other (Union[Table, View, Join]) – The other database object to join with.

  • on (Union[OnClause, Iterable[BaseColumn]]) – The column or columns to join on

Returns:

A representation of a join of which fetch methods can be applied to.

Return type:

Join

classmethod inner_join(other, on)

A chainable method to join with another database object utilising an INNER JOIN.

Parameters:
  • other (Union[Table, View, Join]) – The other database object to join with.

  • on (Union[OnClause, Iterable[OnClause]]) – The column or columns to join on

Returns:

A representation of a join of which fetch methods can be applied to.

Return type:

Join

classmethod await insert(connection, /, *, ignore_on_conflict=False, update_on_conflict=None, returning=None, **values)

This function is a coroutine.

Inserts a new record into the database.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • ignore_on_conflict (bool) – Sets whether to ignore errors when inserting, defaults to False.

  • update_on_conflict (Optional[Union[Iterable[Union[Column, str]], str]]) – An Optional list of or string representing columns to update with new data if a conflict occurs.

  • returning (Optional[Union[Iterable[Union[Column, str]], str]]) – An optional list of or string representing columns to return from the inserted record.

  • **values (Any) – The column to value mapping for the record to insert.

Returns:

A record containing information from the inserted record.

Return type:

Optional[asyncpg.Record]

classmethod await insert_many(connection, /, columns, *values, ignore_on_conflict=False, update_on_conflict=None)

This function is a coroutine.

Inserts a set of new records into the database.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • *values (Dict[str, Any]) – The column to value mappings for each record to insert.

  • ignore_on_conflict (bool) – Sets whether to ignore errors when inserting, defaults to False.

  • update_on_conflict (Optional[Union[Iterable[Union[Column, str]], str]]) – An Optional list of or string representing columns to update with new data if a conflict occurs.

classmethod left_join(other, on)

A chainable method to join with another database object utilising a LEFT JOIN.

Parameters:
  • other (Union[Table, View, Join]) – The other database object to join with.

  • on (Union[OnClause, Iterable[BaseColumn]]) – The column or columns to join on

Returns:

A representation of a join of which fetch methods can be applied to.

Return type:

Join

classmethod right_join(other, on)

A chainable method to join with another database object utilising a RIGHT JOIN.

Parameters:
  • other (Union[Table, View, Join]) – The other database object to join with.

  • on (Union[OnClause, Iterable[BaseColumn]]) – The column or columns to join on

Returns:

A representation of a join of which fetch methods can be applied to.

Return type:

Join

classmethod set_pool(pool)

A function which sets the default connection pool to use on this database onject and subclasses of it.

Parameters:

pool (asyncpg.Pool) – The connection pool to use.

classmethod await update_record(connection, /, record, returning=None, **values)

This function is a coroutine.

Updates a record in the database.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • record (asyncpg.Record) – The record to update.

  • **values (Any) – The column to value mapping to assign to updated record.

Returns:

A record containing information from the updated record.

Return type:

Optional[asyncpg.Record]

classmethod await update_where(connection, /, where, *values, returning=None, **_values)

This function is a coroutine.

Updates records in the database which match a given WHERE clause.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • where (str) – An SQL WHERE clause.

  • *values (Any) – Values to be substituted into the WHERE clause.

  • **values (Any) – The column to value mapping to assign to updated records.

Returns:

A list of records containing information from the updated records.

Return type:

Optional[List[asyncpg.Record]]

ViewColumn

class donphan.ViewColumn(select=MISSING)

A representation of a column in an SQL View object.

Parameters:

select (str) – the attribute to SELECT from the view QUERY for this column. Defaults to the column name.

name

The column’s name.

Type:

str

view

The View the column is a part of.

Type:

View

select

the attribute to SELECT from the view QUERY for this column.

Type:

str

View

class donphan.View(*args, **kwargs)

Base class for creating representations of SQL Database Views.

_name

The name of the table.

Type:

str

_schema

The tables schema.

Type:

str

_columns

The columns contained in the table.

Type:

Iterable[ViewColumn]

_columns_dict

A mapping between a column’s name and itself.

Type:

Dict[str, ViewColumn]

_query

The query used to create the view.

Type:

str

classmethod await create(connection, *, if_not_exists=True, create_schema=True, automatic_migrations=False)

This function is a coroutine.

Creates this database object.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • if_not_exists (bool) – Sets whether creation should continue if the object already exists. Defaults to True.

  • create_schema (bool) – Sets whether the database schema should also be created. Defaults to True.

  • automatic_migrations (bool) – Sets whether migrations should be automatically run. Defaults to False.

classmethod await create_all(connection, /, *, if_not_exists=True, create_schema=True, automatic_migrations=False, with_transaction=True)

This function is a coroutine.

Creates all subclasses of this database object.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • if_not_exists (bool) – Sets whether creation should continue if the object already exists. Defaults to True.

  • create_schema (bool) – Sets whether the database schema should also be created. Defaults to True.

  • automatic_migrations (bool) – Sets whether migrations should be automatically performed. Defaults to False.

  • with_transaction (bool) – Sets whether the database should be wrapped in a transaction. Defaults to True.

classmethod await create_schema(connection, *, if_not_exists=True)

This function is a coroutine.

Creates the schema this database object uses.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • if_not_exists (bool) – Sets whether creation should continue if the schema already exists. Defaults to True.

classmethod await drop(connection, /, *, if_exists=True, cascade=False)

This function is a coroutine.

Drops this object from the database.

Parameters:
  • if_exists (bool) – Sets whether dropping should not error if the object does not exist. Defaults to True.

  • cascade (bool) – Sets whether dropping should cascade. Defaults to False.

classmethod await exists(connection)

Check if the table exists.

Parameters:

conn – The connection to use.

Returns:

True if the table exists, False otherwise.

classmethod export(*, if_not_exists=False, export_schema=True, fp=None)

A function which exports this database object.

Parameters:
  • if_not_exists (bool) – Sets whether the if_not_exists clause should be set on exported objects. Defaults to False.

  • export_schema (bool) – Sets whether to additionally export the schema used by this object. Defaults to True.

  • fp (Optional[os.PathLike, io.TextIOBase]) –

    A file-like object opened in text mode and write mode. or a filename representing a file on disk to write to.

    Note

    If the file-like object passed is opened via open() ensure the object is in a text-writing mode such as "w".

Returns:

The file-like object which was provided or a string containing the exported database object.

Return type:

Union[io.TextIOBase, str]

classmethod export_all(*, if_not_exists=False, export_schema=True, fp=None)

A function which exports all database objects that subclass this object.

Parameters:
  • if_not_exists (bool) – Sets whether the if_not_exists clause should be set on exported objects. Defaults to False.

  • export_schema (bool) – Sets whether to additionally export any schemas used by these objects. Defaults to True.

  • fp (Optional[os.PathLike, io.TextIOBase]) –

    A file-like object opened in text mode and write mode. or a filename representing a file on disk to write to.

    Note

    If the file-like object passed is opened via open() ensure the object is in a text-writing mode such as "w".

Returns:

The file-like object which was provided or a string containing the exported database objects.

Return type:

Union[io.TextIOBase, str]

classmethod export_schema(*, if_not_exists=False, fp=None)

This function is a coroutine.

A function which exports this database object’s schema.

Parameters:
  • if_not_exists (bool) – Sets whether the if_not_exists clause should be set on the exported database schema. Defaults to False.

  • fp (Optional[os.PathLike, io.TextIOBase]) –

    A file-like object opened in text mode and write mode. or a filename representing a file on disk to write to.

    Note

    If the file-like object passed is opened via open() ensure the object is in a text-writing mode such as "w".

Returns:

The file-like object which was provided or a string containing the exported database schema.

Return type:

Union[io.TextIOBase, str]

classmethod await fetch(connection, /, *, limit=None, order_by=None, **values)

This function is a coroutine.

Fetches records in the database which contain the given values.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • limit (Optional[int]) – If provided, sets the maxmimum number of records to be returned.

  • order_by (Optional[Union[Tuple[BaseColumn, Literal[“ASC”, “DESC”]], str]) – Sets the ORDER BY condition on the database query. Takes a tuple concisting of the column and direction, or a string defining the condition.

  • **values (Any) – The column to value mapping to filter records with.

Returns:

Records which which contain the given values.

Return type:

Iterable[asyncpg.Record]

classmethod await fetch_row(connection, /, *, order_by=None, **values)

This function is a coroutine.

Fetches a records in the database which contains the given values.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • order_by (Optional[Union[Tuple[BaseColumn, Literal[“ASC”, “DESC”]], str]) – Sets the ORDER BY condition on the database query. Takes a tuple concisting of the column and direction, or a string defining the condition.

  • **values (Any) – The column to value mapping to filter records with.

Returns:

A record which contains the given values if found.

Return type:

Optional[asyncpg.Record]

classmethod await fetch_row_where(connection, /, where, *values, order_by=None)

This function is a coroutine.

Fetches a record in the database which match a given WHERE clause.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • where (str) – An SQL WHERE clause.

  • *values (Any) – Values to be substituted into the WHERE clause.

  • order_by (Optional[Union[Tuple[BaseColumn, Literal[“ASC”, “DESC”]], str]) – Sets the ORDER BY condition on the database query. Takes a tuple concisting of the column and direction, or a string defining the condition.

Returns:

A record which matches the WHERE clause if found.

Return type:

Optional[asyncpg.Record]

classmethod await fetch_value(connection, /, column, *, order_by=None, **values)

This function is a coroutine.

Fetches a records in the database which contains the given values.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • order_by (Optional[Union[Tuple[BaseColumn, Literal[“ASC”, “DESC”]], str]) – Sets the ORDER BY condition on the database query. Takes a tuple concisting of the column and direction, or a string defining the condition.

  • **values (Any) – The column to value mapping to filter records with.

Returns:

A record which contains the given values if found.

Return type:

Optional[asyncpg.Record]

classmethod await fetch_value_where(connection, /, column, where, *values, order_by=None)

This function is a coroutine.

Fetches a record in the database which match a given WHERE clause.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • where (str) – An SQL WHERE clause.

  • *values (Any) – Values to be substituted into the WHERE clause.

  • order_by (Optional[Union[Tuple[BaseColumn, Literal[“ASC”, “DESC”]], str]) – Sets the ORDER BY condition on the database query. Takes a tuple concisting of the column and direction, or a string defining the condition.

Returns:

A record which matches the WHERE clause if found.

Return type:

Optional[asyncpg.Record]

classmethod await fetch_where(connection, /, where, *values, limit=None, order_by=None)

This function is a coroutine.

Fetches records in the database which match a given WHERE clause.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • where (str) – An SQL WHERE clause.

  • *values (Any) – Values to be substituted into the WHERE clause.

  • limit (Optional[int]) – If provided, sets the maxmimum number of records to be returned.

  • order_by (Optional[Union[Tuple[BaseColumn, Literal[“ASC”, “DESC”]], str]) – Sets the ORDER BY condition on the database query. Takes a tuple concisting of the column and direction, or a string defining the condition.

Returns:

Records which match the WHERE clause.

Return type:

Iterable[asyncpg.Record]

classmethod full_outer_join(other, on)

A chainable method to join with another database object utilising a FULL OUTER JOIN.

Parameters:
  • other (Union[Table, View, Join]) – The other database object to join with.

  • on (Union[OnClause, Iterable[BaseColumn]]) – The column or columns to join on

Returns:

A representation of a join of which fetch methods can be applied to.

Return type:

Join

classmethod inner_join(other, on)

A chainable method to join with another database object utilising an INNER JOIN.

Parameters:
  • other (Union[Table, View, Join]) – The other database object to join with.

  • on (Union[OnClause, Iterable[OnClause]]) – The column or columns to join on

Returns:

A representation of a join of which fetch methods can be applied to.

Return type:

Join

classmethod left_join(other, on)

A chainable method to join with another database object utilising a LEFT JOIN.

Parameters:
  • other (Union[Table, View, Join]) – The other database object to join with.

  • on (Union[OnClause, Iterable[BaseColumn]]) – The column or columns to join on

Returns:

A representation of a join of which fetch methods can be applied to.

Return type:

Join

classmethod right_join(other, on)

A chainable method to join with another database object utilising a RIGHT JOIN.

Parameters:
  • other (Union[Table, View, Join]) – The other database object to join with.

  • on (Union[OnClause, Iterable[BaseColumn]]) – The column or columns to join on

Returns:

A representation of a join of which fetch methods can be applied to.

Return type:

Join

classmethod set_pool(pool)

A function which sets the default connection pool to use on this database onject and subclasses of it.

Parameters:

pool (asyncpg.Pool) – The connection pool to use.

JoinColumn

class donphan.JoinColumn

A representation of a column in an Join.

name

The column’s name.

Type:

str

join

The Join the column is a part of.

Type:

Join

Join

class donphan.Join

Base class for representations of SQL Database Tables.

Note

Subclasses of this class should never be created, instead join methods such as Table.inner_join() should be used to create this model.

_name

The name of the table.

Type:

str

_schema

The tables schema.

Type:

str

_columns

The columns contained in the table.

Type:

Iterable[ViewColumn]

_columns_dict

A mapping between a column’s name and itself.

Type:

Dict[str, ViewColumn]

_query

The query used to create the view.

Type:

str

classmethod await fetch(connection, /, *, limit=None, order_by=None, **values)

This function is a coroutine.

Fetches records in the database which contain the given values.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • limit (Optional[int]) – If provided, sets the maxmimum number of records to be returned.

  • order_by (Optional[Union[Tuple[BaseColumn, Literal[“ASC”, “DESC”]], str]) – Sets the ORDER BY condition on the database query. Takes a tuple concisting of the column and direction, or a string defining the condition.

  • **values (Any) – The column to value mapping to filter records with.

Returns:

Records which which contain the given values.

Return type:

Iterable[asyncpg.Record]

classmethod await fetch_row(connection, /, *, order_by=None, **values)

This function is a coroutine.

Fetches a records in the database which contains the given values.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • order_by (Optional[Union[Tuple[BaseColumn, Literal[“ASC”, “DESC”]], str]) – Sets the ORDER BY condition on the database query. Takes a tuple concisting of the column and direction, or a string defining the condition.

  • **values (Any) – The column to value mapping to filter records with.

Returns:

A record which contains the given values if found.

Return type:

Optional[asyncpg.Record]

classmethod await fetch_row_where(connection, /, where, *values, order_by=None)

This function is a coroutine.

Fetches a record in the database which match a given WHERE clause.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • where (str) – An SQL WHERE clause.

  • *values (Any) – Values to be substituted into the WHERE clause.

  • order_by (Optional[Union[Tuple[BaseColumn, Literal[“ASC”, “DESC”]], str]) – Sets the ORDER BY condition on the database query. Takes a tuple concisting of the column and direction, or a string defining the condition.

Returns:

A record which matches the WHERE clause if found.

Return type:

Optional[asyncpg.Record]

classmethod await fetch_value(connection, /, column, *, order_by=None, **values)

This function is a coroutine.

Fetches a records in the database which contains the given values.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • order_by (Optional[Union[Tuple[BaseColumn, Literal[“ASC”, “DESC”]], str]) – Sets the ORDER BY condition on the database query. Takes a tuple concisting of the column and direction, or a string defining the condition.

  • **values (Any) – The column to value mapping to filter records with.

Returns:

A record which contains the given values if found.

Return type:

Optional[asyncpg.Record]

classmethod await fetch_value_where(connection, /, column, where, *values, order_by=None)

This function is a coroutine.

Fetches a record in the database which match a given WHERE clause.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • where (str) – An SQL WHERE clause.

  • *values (Any) – Values to be substituted into the WHERE clause.

  • order_by (Optional[Union[Tuple[BaseColumn, Literal[“ASC”, “DESC”]], str]) – Sets the ORDER BY condition on the database query. Takes a tuple concisting of the column and direction, or a string defining the condition.

Returns:

A record which matches the WHERE clause if found.

Return type:

Optional[asyncpg.Record]

classmethod await fetch_where(connection, /, where, *values, limit=None, order_by=None)

This function is a coroutine.

Fetches records in the database which match a given WHERE clause.

Parameters:
  • connection (asyncpg.Connection) – The database connection to use for transactions.

  • where (str) – An SQL WHERE clause.

  • *values (Any) – Values to be substituted into the WHERE clause.

  • limit (Optional[int]) – If provided, sets the maxmimum number of records to be returned.

  • order_by (Optional[Union[Tuple[BaseColumn, Literal[“ASC”, “DESC”]], str]) – Sets the ORDER BY condition on the database query. Takes a tuple concisting of the column and direction, or a string defining the condition.

Returns:

Records which match the WHERE clause.

Return type:

Iterable[asyncpg.Record]

classmethod full_outer_join(other, on)

A chainable method to join with another database object utilising a FULL OUTER JOIN.

Parameters:
  • other (Union[Table, View, Join]) – The other database object to join with.

  • on (Union[OnClause, Iterable[BaseColumn]]) – The column or columns to join on

Returns:

A representation of a join of which fetch methods can be applied to.

Return type:

Join

classmethod inner_join(other, on)

A chainable method to join with another database object utilising an INNER JOIN.

Parameters:
  • other (Union[Table, View, Join]) – The other database object to join with.

  • on (Union[OnClause, Iterable[OnClause]]) – The column or columns to join on

Returns:

A representation of a join of which fetch methods can be applied to.

Return type:

Join

classmethod left_join(other, on)

A chainable method to join with another database object utilising a LEFT JOIN.

Parameters:
  • other (Union[Table, View, Join]) – The other database object to join with.

  • on (Union[OnClause, Iterable[BaseColumn]]) – The column or columns to join on

Returns:

A representation of a join of which fetch methods can be applied to.

Return type:

Join

classmethod right_join(other, on)

A chainable method to join with another database object utilising a RIGHT JOIN.

Parameters:
  • other (Union[Table, View, Join]) – The other database object to join with.

  • on (Union[OnClause, Iterable[BaseColumn]]) – The column or columns to join on

Returns:

A representation of a join of which fetch methods can be applied to.

Return type:

Join

classmethod set_pool(pool)

A function which sets the default connection pool to use on this database onject and subclasses of it.

Parameters:

pool (asyncpg.Pool) – The connection pool to use.

Utility Functions / Classes

Below are utility some utility functions and classes which interact with donphan.

donphan.export_db(*, if_not_exists=False, fp=None)

A helper function which exports all objects in the database.

Parameters:
  • if_not_exists (bool) – Sets whether the if_not_exists clause should be set on exported objects. Defaults to False.

  • fp (Optional[os.PathLike, io.TextIOBase]) –

    A file-like object opened in text mode and write mode. or a filename representing a file on disk to write to.

    Note

    If the file-like object passed is opened via open() ensure the object is in a text-writing mode such as "w".

    If the file-like object passed is a path it will be opened in write-mode "w".

Returns:

The file-like object which was provided or a string containing the exported database.

Return type:

Union[io.TextIOBase, str]

donphan.utils.not_creatable(cls)

Marks a type as non-creatable.

Enum

class donphan.Enum

A fully-featured Enum class with support for custom types, see enum.Enum for documentation.

Usage Example:

from donphan import Column, Enum, Table


class MyEnum(Enum):
    A = 1
    B = 2
    C = 3


class MyTable(Table):
    key: Column[str] = Column(primary_key=True)
    value: Column[MyEnum]

Type Validation Helpers

class donphan.OnClause(a, b, operator='eq')

A NamedTuple defining an on clause for a join.

Parameters:
  • a (BaseColumn) – the primary column to join to.

  • b (BaseColumn) – the secondary column to join to.

  • operator (Literal["eq", "lt", "le", "ne", "ge", "gt"]) – An opterator to use to compare columns, defaults to eq.

class donphan.TypeCodec(format, encoder, decoder)

A NamedTuple defining a custom type codec.

See asyncpg.Connection.set_type_codec for more information.

Parameters:
  • format (Literal["text", "binary", "tuple"]) – The type of decoder/encoder to use.

  • encoder (collections.abc.Callable[..., Any]) – A callable which given a python object returns an encoded value.

  • decoder (collections.abc.Callable[..., Any]) – A callable which given an encoded value returns a python object.