Skip to content

Database API

The Database API provides utilities for database connections, ORM integration, and repository patterns.

Core Components

DatabaseEngine

__init__ 

__init__(orm: ORMEnum, configuration: Any) -> None

Initialize the DatabaseEngine.

Parameters:

Name Type Description Default
orm ORMEnum

The ORM to use (Tortoise or SQLAlchemy).

 required
configuration Any

The configuration for the ORM.

 required

generate_context 

generate_context()

Generate a database context for SQLAlchemy ORM.

Raises:

Type Description
WrongORMException

If the ORM is not SQLAlchemy.

Returns:

Name Type Description
AppDBContext

The database context that carries database session for SQLAlchemy.

Types & Objects

AppDBContext

Application Database Context for SQLAlchemy.

Usage example: 

from ascender.core.database import DatabaseEngine
from .entities.users import UserEntity


@Injectable(provided_in="root")
class MyService:
    def __init__(self, engine: DatabaseEngine):
        self.context = engine.generate_context() # It's AppDBContext

    async def create_user(self, name: str):
        # creates user
        async with self.context() as db:
            entity = UserEntity(name)
            db.add(entity) # sqlalchemy session calls
            await db.commit() # sqlalchemy session calls

    async def fetch_user(self, id: int):
        query = self.context.construct(UserEntity) # After you call construct, it returns `Construction[UserEntity]` where you can apply query constrains (e.g. filter, where, limits and etc).

        query = await query.filter(UserEntity.id == id) # as soon as we `await` the `Construction[UserEntity]` it's no longer `Construction[UserEntity]` anymore, it's SQLAlchemy's `Result` object.

        return query.scalar() # Returns UserEntity or None, refer to SQLAlchemy's official documentation for more information about `Result` object.

construct 

construct(*entities: type[T]) -> Constructor[T]

Creates a Constructor for the given entities.

As in given example above, after you call construct method, it returns Constructor[T] where you can apply query constrains (e.g. filter, where, limits and etc). As soon as you await the Constructor[T] it's no longer Constructor[T] anymore, it's SQLAlchemy's Result object, calling await is similar to doing await session.execute(...).

Returns:

Type Description
Constructor[T]

Constructor[T]: The constructed query builder.

__call__ 

__call__() -> AsyncSession

Provides a new database session.

Returns:

Name Type Description
AsyncSession AsyncSession

The new database session.

ORMEnum

Bases: Enum

Source code in ascender/core/database/types/orm_enum.py 
4
5
6
7
8
class ORMEnum(Enum):
    TORTOISE = "tortoise"
    """Tortoise ORM"""
    SQLALCHEMY = "sqlalchemy"
    """SQLAlchemy ORM"""

TORTOISE class-attribute instance-attribute 

TORTOISE = 'tortoise'

Tortoise ORM

SQLALCHEMY class-attribute instance-attribute 

SQLALCHEMY = 'sqlalchemy'

SQLAlchemy ORM

provideDatabase 

provideDatabase(orm_mode: ORMEnum, orm_configuration: Any) -> Provider | Sequence[Provider]

Provides the DatabaseEngine based on the ORM mode.

Usage example: 

# src/bootstrap.py

from ascender.core.database import provideDatabase, ORMEnum

providers = [
    provideDatabase(ORMEnum.SQLALCHEMY, {
        "database_url": "sqlite+aiosqlite:///./test.db",
        "entities": [...], # specify path using python's import style path (e.g. "entities.users")
    }),
]

Parameters:

Name Type Description Default
orm_mode ORMEnum

The ORM mode to use.

 required
orm_configuration Any

The configuration for the ORM.

 required

Returns:

Type Description
Provider | Sequence[Provider]

Provider | Sequence[Provider]: The provider for the DatabaseEngine.

See Also