Database class

edgy.Database

Database(url=None, *, force_rollback=None, config=None, full_isolation=None, poll_interval=None, **options)

An abstraction on the top of the EncodeORM databases.Database object.

This object allows to pass also a configuration dictionary in the format of

DATABASEZ_CONFIG = { "connection": { "credentials": { "scheme": 'sqlite', "postgres"... "host": ..., "port": ..., "user": ..., "password": ..., "database": ..., "options": { "driver": ... "ssl": ... } } } }

PARAMETER	DESCRIPTION
url	TYPE: Union[str, DatabaseURL, URL, Database, None] DEFAULT: None
force_rollback	TYPE: Union[bool, None] DEFAULT: None
config	TYPE: Optional['DictAny'] DEFAULT: None
full_isolation	TYPE: Union[bool, None] DEFAULT: None
poll_interval	TYPE: Union[float, None] DEFAULT: None
**options	TYPE: Any DEFAULT: {}

Source code in .venv/lib/python3.11/site-packages/databasez/core/database.py

def __init__(
    self,
    url: typing.Union[str, DatabaseURL, URL, Database, None] = None,
    *,
    force_rollback: typing.Union[bool, None] = None,
    config: typing.Optional["DictAny"] = None,
    full_isolation: typing.Union[bool, None] = None,
    # for
    poll_interval: typing.Union[float, None] = None,
    **options: typing.Any,
):
    init()
    assert config is None or url is None, "Use either 'url' or 'config', not both."
    if isinstance(url, Database):
        assert not options, "Cannot specify options when copying a Database object."
        self.backend = url.backend.__copy__()
        self.url = url.url
        self.options = url.options
        self._call_hooks = url._call_hooks
        if poll_interval is None:
            poll_interval = url.poll_interval
        if force_rollback is None:
            force_rollback = bool(url.force_rollback)
        if full_isolation is None:
            full_isolation = bool(url._full_isolation)
    else:
        url = DatabaseURL(url)
        if config and "connection" in config:
            connection_config = config["connection"]
            if "credentials" in connection_config:
                connection_config = connection_config["credentials"]
                url = url.replace(**connection_config)
        self.backend, self.url, self.options = self.apply_database_url_and_options(
            url, **options
        )
        if force_rollback is None:
            force_rollback = False
        if full_isolation is None:
            full_isolation = False
        if poll_interval is None:
            poll_interval = DATABASEZ_POLL_INTERVAL
    self.poll_interval = poll_interval
    self._full_isolation = full_isolation
    self._force_rollback = ForceRollback(force_rollback)
    self.backend.owner = self
    self._connection_map = weakref.WeakKeyDictionary()
    self._databases_map = {}

    # When `force_rollback=True` is used, we use a single global
    # connection, within a transaction that always rolls back.
    self._global_connection: typing.Optional[Connection] = None

    self.ref_counter: int = 0
    self.ref_lock: asyncio.Lock = asyncio.Lock()

_connection_map instance-attribute

_connection_map = WeakKeyDictionary()

_databases_map instance-attribute

_databases_map = {}

_loop class-attribute instance-attribute

_loop = None

backend instance-attribute

backend

url instance-attribute

url

options instance-attribute

options

is_connected class-attribute instance-attribute

is_connected = False

_call_hooks class-attribute instance-attribute

_call_hooks = True

_remove_global_connection class-attribute instance-attribute

_remove_global_connection = True

_full_isolation class-attribute instance-attribute

_full_isolation = full_isolation

poll_interval instance-attribute

poll_interval = poll_interval

_force_rollback instance-attribute

_force_rollback = ForceRollback(force_rollback)

force_rollback class-attribute instance-attribute

force_rollback = ForceRollbackDescriptor()

async_helper class-attribute instance-attribute

async_helper = AsyncHelperDatabase

_global_connection instance-attribute

_global_connection = None

ref_counter instance-attribute

ref_counter = 0

ref_lock instance-attribute

ref_lock = Lock()

_current_task property

_current_task

_connection property writable

_connection

engine property

engine

__copy__

__copy__()

Source code in .venv/lib/python3.11/site-packages/databasez/core/database.py

def __copy__(self) -> Database:
    return self.__class__(self)

inc_refcount async

inc_refcount()

Internal method to bump the ref_count.

Return True if ref_count is 0, False otherwise.

Should not be used outside of tests. Use connect and hooks instead. Not multithreading safe!

Source code in .venv/lib/python3.11/site-packages/databasez/core/database.py

async def inc_refcount(self) -> bool:
    """
    Internal method to bump the ref_count.

    Return True if ref_count is 0, False otherwise.

    Should not be used outside of tests. Use connect and hooks instead.
    Not multithreading safe!
    """
    async with self.ref_lock:
        self.ref_counter += 1
        # on the first call is count is 1 because of the former +1
        if self.ref_counter == 1:
            return True
    return False

decr_refcount async

decr_refcount()

Internal method to decrease the ref_count.

Return True if ref_count drops to 0, False otherwise.

Should not be used outside of tests. Use disconnect and hooks instead. Not multithreading safe!

Source code in .venv/lib/python3.11/site-packages/databasez/core/database.py

async def decr_refcount(self) -> bool:
    """
    Internal method to decrease the ref_count.

    Return True if ref_count drops to 0, False otherwise.

    Should not be used outside of tests. Use disconnect and hooks instead.
    Not multithreading safe!
    """
    async with self.ref_lock:
        self.ref_counter -= 1
        # on the last call, the count is 0
        if self.ref_counter == 0:
            return True
    return False

connect_hook async

connect_hook()

Refcount protected connect hook. Executed begore engine and global connection setup.

Source code in .venv/lib/python3.11/site-packages/databasez/core/database.py

async def connect_hook(self) -> None:
    """Refcount protected connect hook. Executed begore engine and global connection setup."""

connect async

connect()

Establish the connection pool.

Source code in .venv/lib/python3.11/site-packages/databasez/core/database.py

async def connect(self) -> bool:
    """
    Establish the connection pool.
    """
    loop = asyncio.get_running_loop()
    if self._loop is not None and loop != self._loop:
        if self.poll_interval < 0:
            raise RuntimeError("Subdatabases and polling are disabled")
        # copy when not in map
        if loop not in self._databases_map:
            assert (
                self._global_connection is not None
            ), "global connection should have been set"
            # correctly initialize force_rollback with parent value
            database = self.__class__(
                self, force_rollback=bool(self.force_rollback), full_isolation=False
            )
            # prevent side effects of connect_hook
            database._call_hooks = False
            database._global_connection = self._global_connection
            self._databases_map[loop] = database
        # forward call
        return await self._databases_map[loop].connect()

    if not await self.inc_refcount():
        assert self.is_connected, "ref_count < 0"
        return False
    if self._call_hooks:
        try:
            await self.connect_hook()
        except BaseException as exc:
            await self.decr_refcount()
            raise exc
    self._loop = asyncio.get_event_loop()

    await self.backend.connect(self.url, **self.options)
    logger.info("Connected to database %s", self.url.obscure_password, extra=CONNECT_EXTRA)
    self.is_connected = True

    if self._global_connection is None:
        connection = Connection(self, force_rollback=True, full_isolation=self._full_isolation)
        self._global_connection = connection
    else:
        self._remove_global_connection = False
    return True

disconnect_hook async

disconnect_hook()

Refcount protected disconnect hook. Executed after connection, engine cleanup.

Source code in .venv/lib/python3.11/site-packages/databasez/core/database.py

async def disconnect_hook(self) -> None:
    """Refcount protected disconnect hook. Executed after connection, engine cleanup."""

disconnect async

disconnect(force=False, *, parent_database=None)

Close all connections in the connection pool.

PARAMETER	DESCRIPTION
force	TYPE: bool DEFAULT: False
parent_database	TYPE: Optional[Database] DEFAULT: None

Source code in .venv/lib/python3.11/site-packages/databasez/core/database.py

@multiloop_protector(True, inject_parent=True)
async def disconnect(
    self, force: bool = False, *, parent_database: typing.Optional[Database] = None
) -> bool:
    """
    Close all connections in the connection pool.
    """
    # parent_database is injected and should not be specified manually
    if not await self.decr_refcount() or force:
        if not self.is_connected:
            logger.debug("Already disconnected, skipping disconnection")
            return False
        if force:
            logger.warning("Force disconnect, despite refcount not 0")
        else:
            return False
    if parent_database is not None:
        loop = asyncio.get_running_loop()
        del parent_database._databases_map[loop]
    if force:
        if self._databases_map:
            assert not self._databases_map, "sub databases still active, force terminate them"
            for sub_database in self._databases_map.values():
                await arun_coroutine_threadsafe(
                    sub_database.disconnect(True), sub_database._loop, self.poll_interval
                )
            self._databases_map = {}
    assert not self._databases_map, "sub databases still active"

    try:
        assert self._global_connection is not None
        if self._remove_global_connection:
            await self._global_connection.__aexit__()
            self._global_connection = None
        self._connection = None
    finally:
        logger.info(
            "Disconnected from database %s",
            self.url.obscure_password,
            extra=DISCONNECT_EXTRA,
        )
        self.is_connected = False
        await self.backend.disconnect()
        self._loop = None
        if self._call_hooks:
            await self.disconnect_hook()
    return True

fetch_all async

fetch_all(query, values=None, timeout=None)

PARAMETER	DESCRIPTION
query	TYPE: Union[ClauseElement, str]
values	TYPE: Optional[dict] DEFAULT: None
timeout	TYPE: Optional[float] DEFAULT: None

Source code in .venv/lib/python3.11/site-packages/databasez/core/database.py

async def fetch_all(
    self,
    query: typing.Union[ClauseElement, str],
    values: typing.Optional[dict] = None,
    timeout: typing.Optional[float] = None,
) -> typing.List[interfaces.Record]:
    async with self.connection() as connection:
        return await connection.fetch_all(query, values, timeout=timeout)

fetch_one async

fetch_one(query, values=None, pos=0, timeout=None)

PARAMETER	DESCRIPTION
query	TYPE: Union[ClauseElement, str]
values	TYPE: Optional[dict] DEFAULT: None
pos	TYPE: int DEFAULT: 0
timeout	TYPE: Optional[float] DEFAULT: None

Source code in .venv/lib/python3.11/site-packages/databasez/core/database.py

async def fetch_one(
    self,
    query: typing.Union[ClauseElement, str],
    values: typing.Optional[dict] = None,
    pos: int = 0,
    timeout: typing.Optional[float] = None,
) -> typing.Optional[interfaces.Record]:
    async with self.connection() as connection:
        return await connection.fetch_one(query, values, pos=pos, timeout=timeout)

fetch_val async

fetch_val(query, values=None, column=0, pos=0, timeout=None)

PARAMETER	DESCRIPTION
query	TYPE: Union[ClauseElement, str]
values	TYPE: Optional[dict] DEFAULT: None
column	TYPE: Any DEFAULT: 0
pos	TYPE: int DEFAULT: 0
timeout	TYPE: Optional[float] DEFAULT: None

Source code in .venv/lib/python3.11/site-packages/databasez/core/database.py

async def fetch_val(
    self,
    query: typing.Union[ClauseElement, str],
    values: typing.Optional[dict] = None,
    column: typing.Any = 0,
    pos: int = 0,
    timeout: typing.Optional[float] = None,
) -> typing.Any:
    async with self.connection() as connection:
        return await connection.fetch_val(
            query,
            values,
            column=column,
            pos=pos,
            timeout=timeout,
        )

execute async

execute(query, values=None, timeout=None)

PARAMETER	DESCRIPTION
query	TYPE: Union[ClauseElement, str]
values	TYPE: Any DEFAULT: None
timeout	TYPE: Optional[float] DEFAULT: None

Source code in .venv/lib/python3.11/site-packages/databasez/core/database.py

async def execute(
    self,
    query: typing.Union[ClauseElement, str],
    values: typing.Any = None,
    timeout: typing.Optional[float] = None,
) -> typing.Union[interfaces.Record, int]:
    async with self.connection() as connection:
        return await connection.execute(query, values, timeout=timeout)

execute_many async

execute_many(query, values=None, timeout=None)

PARAMETER	DESCRIPTION
query	TYPE: Union[ClauseElement, str]
values	TYPE: Any DEFAULT: None
timeout	TYPE: Optional[float] DEFAULT: None

Source code in .venv/lib/python3.11/site-packages/databasez/core/database.py

async def execute_many(
    self,
    query: typing.Union[ClauseElement, str],
    values: typing.Any = None,
    timeout: typing.Optional[float] = None,
) -> typing.Union[typing.Sequence[interfaces.Record], int]:
    async with self.connection() as connection:
        return await connection.execute_many(query, values, timeout=timeout)

iterate async

iterate(query, values=None, chunk_size=None, timeout=None)

PARAMETER	DESCRIPTION
query	TYPE: Union[ClauseElement, str]
values	TYPE: Optional[dict] DEFAULT: None
chunk_size	TYPE: Optional[int] DEFAULT: None
timeout	TYPE: Optional[float] DEFAULT: None

Source code in .venv/lib/python3.11/site-packages/databasez/core/database.py

async def iterate(
    self,
    query: typing.Union[ClauseElement, str],
    values: typing.Optional[dict] = None,
    chunk_size: typing.Optional[int] = None,
    timeout: typing.Optional[float] = None,
) -> typing.AsyncGenerator[interfaces.Record, None]:
    async with self.connection() as connection:
        async for record in connection.iterate(query, values, chunk_size, timeout=timeout):
            yield record

batched_iterate async

batched_iterate(query, values=None, batch_size=None, batch_wrapper=tuple, timeout=None)

PARAMETER	DESCRIPTION
query	TYPE: Union[ClauseElement, str]
values	TYPE: Optional[dict] DEFAULT: None
batch_size	TYPE: Optional[int] DEFAULT: None
batch_wrapper	TYPE: BatchCallable DEFAULT: tuple
timeout	TYPE: Optional[float] DEFAULT: None

Source code in .venv/lib/python3.11/site-packages/databasez/core/database.py

async def batched_iterate(
    self,
    query: typing.Union[ClauseElement, str],
    values: typing.Optional[dict] = None,
    batch_size: typing.Optional[int] = None,
    batch_wrapper: BatchCallable = tuple,
    timeout: typing.Optional[float] = None,
) -> typing.AsyncGenerator[BatchCallableResult, None]:
    async with self.connection() as connection:
        async for batch in typing.cast(
            typing.AsyncGenerator["BatchCallableResult", None],
            connection.batched_iterate(
                query,
                values,
                batch_wrapper=batch_wrapper,
                batch_size=batch_size,
                timeout=timeout,
            ),
        ):
            yield batch

transaction

transaction(*, force_rollback=False, **kwargs)

PARAMETER	DESCRIPTION
force_rollback	TYPE: bool DEFAULT: False
**kwargs	TYPE: Any DEFAULT: {}

Source code in .venv/lib/python3.11/site-packages/databasez/core/database.py

def transaction(self, *, force_rollback: bool = False, **kwargs: typing.Any) -> "Transaction":
    return Transaction(self.connection, force_rollback=force_rollback, **kwargs)

run_sync async

run_sync(fn, *args, timeout=None, **kwargs)

PARAMETER	DESCRIPTION
fn	TYPE: Callable[..., Any]
*args	TYPE: Any DEFAULT: ()
timeout	TYPE: Optional[float] DEFAULT: None
**kwargs	TYPE: Any DEFAULT: {}

Source code in .venv/lib/python3.11/site-packages/databasez/core/database.py

async def run_sync(
    self,
    fn: typing.Callable[..., typing.Any],
    *args: typing.Any,
    timeout: typing.Optional[float] = None,
    **kwargs: typing.Any,
) -> typing.Any:
    async with self.connection() as connection:
        return await connection.run_sync(fn, *args, **kwargs, timeout=timeout)

create_all async

create_all(meta, timeout=None, **kwargs)

PARAMETER	DESCRIPTION
meta	TYPE: MetaData
timeout	TYPE: Optional[float] DEFAULT: None
**kwargs	TYPE: Any DEFAULT: {}

Source code in .venv/lib/python3.11/site-packages/databasez/core/database.py

async def create_all(
    self, meta: MetaData, timeout: typing.Optional[float] = None, **kwargs: typing.Any
) -> None:
    async with self.connection() as connection:
        await connection.create_all(meta, **kwargs, timeout=timeout)

drop_all async

drop_all(meta, timeout=None, **kwargs)

PARAMETER	DESCRIPTION
meta	TYPE: MetaData
timeout	TYPE: Optional[float] DEFAULT: None
**kwargs	TYPE: Any DEFAULT: {}

Source code in .venv/lib/python3.11/site-packages/databasez/core/database.py

async def drop_all(
    self, meta: MetaData, timeout: typing.Optional[float] = None, **kwargs: typing.Any
) -> None:
    async with self.connection() as connection:
        await connection.drop_all(meta, **kwargs, timeout=timeout)

_non_global_connection

_non_global_connection(timeout=None)

PARAMETER	DESCRIPTION
timeout	TYPE: Optional[float] DEFAULT: None

Source code in .venv/lib/python3.11/site-packages/databasez/core/database.py

@multiloop_protector(False)
def _non_global_connection(
    self,
    timeout: typing.Optional[
        float
    ] = None,  # stub for type checker, multiloop_protector handles timeout
) -> Connection:
    if self._connection is None:
        _connection = self._connection = Connection(self)
        return _connection
    return self._connection

connection

connection(timeout=None)

PARAMETER	DESCRIPTION
timeout	TYPE: Optional[float] DEFAULT: None

Source code in .venv/lib/python3.11/site-packages/databasez/core/database.py

def connection(self, timeout: typing.Optional[float] = None) -> Connection:
    if not self.is_connected:
        raise RuntimeError("Database is not connected")
    if self.force_rollback:
        return typing.cast(Connection, self._global_connection)
    return self._non_global_connection(timeout=timeout)

get_backends classmethod

get_backends(scheme='', *, overwrite_paths=['databasez.overwrites'], database_name='Database', connection_name='Connection', transaction_name='Transaction', database_class=None, connection_class=None, transaction_class=None)

PARAMETER	DESCRIPTION
scheme	TYPE: str DEFAULT: ''
overwrite_paths	TYPE: Sequence[str] DEFAULT: ['databasez.overwrites']
database_name	TYPE: str DEFAULT: 'Database'
connection_name	TYPE: str DEFAULT: 'Connection'
transaction_name	TYPE: str DEFAULT: 'Transaction'
database_class	TYPE: Optional[Type[DatabaseBackend]] DEFAULT: None
connection_class	TYPE: Optional[Type[ConnectionBackend]] DEFAULT: None
transaction_class	TYPE: Optional[Type[TransactionBackend]] DEFAULT: None

Source code in .venv/lib/python3.11/site-packages/databasez/core/database.py

@classmethod
def get_backends(
    cls,
    # let scheme empty for direct imports
    scheme: str = "",
    *,
    overwrite_paths: typing.Sequence[str] = ["databasez.overwrites"],
    database_name: str = "Database",
    connection_name: str = "Connection",
    transaction_name: str = "Transaction",
    database_class: typing.Optional[typing.Type[interfaces.DatabaseBackend]] = None,
    connection_class: typing.Optional[typing.Type[interfaces.ConnectionBackend]] = None,
    transaction_class: typing.Optional[typing.Type[interfaces.TransactionBackend]] = None,
) -> typing.Tuple[
    typing.Type[interfaces.DatabaseBackend],
    typing.Type[interfaces.ConnectionBackend],
    typing.Type[interfaces.TransactionBackend],
]:
    module: typing.Any = None
    for overwrite_path in overwrite_paths:
        imp_path = f"{overwrite_path}.{scheme.replace('+', '_')}" if scheme else overwrite_path
        try:
            module = importlib.import_module(imp_path)
        except ImportError as exc:
            logging.debug(
                f'Import of "{imp_path}" failed. This is not an error.', exc_info=exc
            )
            if "+" in scheme:
                imp_path = f"{overwrite_path}.{scheme.split('+', 1)[0]}"
                try:
                    module = importlib.import_module(imp_path)
                except ImportError as exc:
                    logging.debug(
                        f'Import of "{imp_path}" failed. This is not an error.', exc_info=exc
                    )
        if module is not None:
            break
    database_class = getattr(module, database_name, database_class)
    assert database_class is not None and issubclass(
        database_class, interfaces.DatabaseBackend
    )
    connection_class = getattr(module, connection_name, connection_class)
    assert connection_class is not None and issubclass(
        connection_class, interfaces.ConnectionBackend
    )
    transaction_class = getattr(module, transaction_name, transaction_class)
    assert transaction_class is not None and issubclass(
        transaction_class, interfaces.TransactionBackend
    )
    return database_class, connection_class, transaction_class

apply_database_url_and_options classmethod

apply_database_url_and_options(url, *, overwrite_paths=['databasez.overwrites'], **options)

PARAMETER	DESCRIPTION
url	TYPE: Union[DatabaseURL, str]
overwrite_paths	TYPE: Sequence[str] DEFAULT: ['databasez.overwrites']
**options	TYPE: Any DEFAULT: {}

Source code in .venv/lib/python3.11/site-packages/databasez/core/database.py

@classmethod
def apply_database_url_and_options(
    cls,
    url: typing.Union[DatabaseURL, str],
    *,
    overwrite_paths: typing.Sequence[str] = ["databasez.overwrites"],
    **options: typing.Any,
) -> typing.Tuple[interfaces.DatabaseBackend, DatabaseURL, typing.Dict[str, typing.Any]]:
    url = DatabaseURL(url)
    database_class, connection_class, transaction_class = cls.get_backends(
        url.scheme,
        database_class=default_database,
        connection_class=default_connection,
        transaction_class=default_transaction,
        overwrite_paths=overwrite_paths,
    )

    backend = database_class(
        connection_class=connection_class, transaction_class=transaction_class
    )
    url, options = backend.extract_options(url, **options)
    # check against transformed url
    assert url.sqla_url.get_dialect(True).is_async, f'Dialect: "{url.scheme}" is not async.'

    return backend, url, options