ushka.orm.database
This module provides an asynchronous Object-Relational Mapper (ORM) for the Ushka framework, built on top of SQLAlchemy 2.0.
It includes the UshkaDB manager for engine and session handling, a dynamically
generated BaseModel with Active Record-style CRUD operations, and helper
methods for defining common SQLAlchemy column types.
class UshkaDB
Asynchronous Database Manager for the Ushka Framework.
Handles engine creation, session management, and provides the declarative Model base with built-in async Active Record methods.
Attributes
CASCADE | str | SQL ON DELETE CASCADE action. |
SET_NULL | str | SQL ON DELETE SET NULL action. |
RESTRICT | str | SQL ON DELETE RESTRICT action. |
url | str | The database connection URL. |
engine | sqlalchemy.ext.asyncio.AsyncEngine | The asynchronous SQLAlchemy engine. |
session_factory | async_sessionmaker | Factory for creating new async sessions. |
Session | async_sessionmaker | Factory of independent async sessions ( |
metadata | sqlalchemy.MetaData | The metadata object used by the declarative base. |
Model | Type[_Base] | The dynamic BaseModel class with CRUD methods. |
__init__(self, url: Optional[str] = None)
Initializes the database manager.
The engine is created lazily, on first use, from url or, when it is
omitted, from the application config ([database] url in
ushka.toml, USHKA_DATABASE_URL or DATABASE_URL). This is
the same setting the ushka db migration commands read, so the app
and its migrations always target the same database.
to_async_url(cls, url: str) -> str
Returns url with an async driver (sqlite:// → sqlite+aiosqlite://).
configure(self, url: str) -> None
Points the manager at another database (call before first use).
text(sql: str, **params: Any)
Guarded sqlalchemy.text(): refuses SQL built from request data.
sqlalchemy.text() copies the string through re.sub, which drops
the Untrusted mark, so the check has to happen before the call. Use
this (or pass the string to execute) instead of importing text.
Parameters
sql | str | The SQL with |
**params | Any | Values bound to the placeholders (never pasted into the SQL). |
execute(self, stmt)
Executes a raw SQL statement or SQLAlchemy construct using a session.
Parameters
stmt | Any | A SQLAlchemy executable construct (e.g., Select, Insert, raw text). |
Returns
sqlalchemy.engine.Result | The result object from the execution. |
Raises
UnsafeValueError | If the SQL text is |
String(self, length = 255, unique = False, nullable = False, default = None, choices: list = None) -> Column
Defines a SQLAlchemy String column.
Parameters
length | int | Maximum length of the string. By default 255. |
unique | bool | If True, enforces uniqueness constraint. By default False. |
nullable | bool | If True, allows NULL values. By default False. |
default | Any | Default value for the column. By default None. |
choices | list | If provided, converts the column to a SQLAlchemy Enum type using the list elements as valid values. By default None. |
Returns
sqlalchemy.Column | The configured Column object. |
UUID(self, unique = False, nullable = False, primary_key = False, default = None) -> Column
Defines a SQLAlchemy UUID column.
If primary_key is True and default is None, uuid.uuid4 is used
as the default generator.
Parameters
unique | bool | If True, enforces uniqueness constraint. By default False. |
nullable | bool | If True, allows NULL values. By default False. |
primary_key | bool | If True, sets this column as the primary key. By default False. |
default | Any | Default value or callable for the column. By default None. |
Returns
sqlalchemy.Column | The configured Column object. |
Text(self, nullable = False, default = None) -> Column
Defines a SQLAlchemy Text column (for long strings).
Parameters
nullable | bool | If True, allows NULL values. By default False. |
default | Any | Default value for the column. By default None. |
Returns
sqlalchemy.Column | The configured Column object. |
Int(self, unique = False, nullable = False, default = None) -> Column
Defines a SQLAlchemy Integer column.
Parameters
unique | bool | If True, enforces uniqueness constraint. By default False. |
nullable | bool | If True, allows NULL values. By default False. |
default | Any | Default value for the column. By default None. |
Returns
sqlalchemy.Column | The configured Column object. |
BigInt(self, unique = False, nullable = False, default = None) -> Column
Defines a SQLAlchemy BigInteger column.
Parameters
unique | bool | If True, enforces uniqueness constraint. By default False. |
nullable | bool | If True, allows NULL values. By default False. |
default | Any | Default value for the column. By default None. |
Returns
sqlalchemy.Column | The configured Column object. |
Float(self, nullable = False, default = 0.0) -> Column
Defines a SQLAlchemy Float column.
Parameters
nullable | bool | If True, allows NULL values. By default False. |
default | float | Default value for the column. By default 0.0. |
Returns
sqlalchemy.Column | The configured Column object. |
Decimal(self, precision = 10, scale = 2, nullable = False, default = 0) -> Column
Defines a SQLAlchemy Numeric/Decimal column.
Parameters
precision | int | Total number of digits. By default 10. |
scale | int | Number of digits after the decimal point. By default 2. |
nullable | bool | If True, allows NULL values. By default False. |
default | int or float | Default value for the column. By default 0. |
Returns
sqlalchemy.Column | The configured Column object. |
Bool(self, default = False, nullable = False) -> Column
Defines a SQLAlchemy Boolean column.
Parameters
default | bool | Default value for the column. By default False. |
nullable | bool | If True, allows NULL values. By default False. |
Returns
sqlalchemy.Column | The configured Column object. |
DateTime(self, auto_now = False, nullable = False) -> Column
Defines a SQLAlchemy DateTime column.
Parameters
auto_now | bool | If True, sets the default value to the current database time (func.now()). By default False. |
nullable | bool | If True, allows NULL values. By default False. |
Returns
sqlalchemy.Column | The configured Column object. |
Date(self, auto_now = False, nullable = False) -> Column
Defines a SQLAlchemy Date column.
Parameters
auto_now | bool | If True, sets the default value to the current database date (func.current_date()). By default False. |
nullable | bool | If True, allows NULL values. By default False. |
Returns
sqlalchemy.Column | The configured Column object. |
Time(self, nullable = False) -> Column
Defines a SQLAlchemy Time column.
Parameters
nullable | bool | If True, allows NULL values. By default False. |
Returns
sqlalchemy.Column | The configured Column object. |
JSON(self, nullable = False, default = None) -> Column
Defines a SQLAlchemy JSON column.
Parameters
nullable | bool | If True, allows NULL values. By default False. |
default | Any | Default value for the column. If None, defaults to an empty dictionary {}. By default None. |
Returns
sqlalchemy.Column | The configured Column object. |
Enum(self, enum_cls: Type[enum.Enum], nullable = False, default = None) -> Column
Defines a SQLAlchemy Enum column based on a Python Enum class.
Parameters
enum_cls | Type[enum.Enum] | The Python Enum class defining the valid values. |
nullable | bool | If True, allows NULL values. By default False. |
default | Any | Default value for the column (should be an Enum member or its value). By default None. |
Returns
sqlalchemy.Column | The configured Column object. |
Binary(self, length: int = None, nullable = False) -> Column
Defines a SQLAlchemy LargeBinary column.
Parameters
length | int | Maximum length of the binary data. By default None. |
nullable | bool | If True, allows NULL values. By default False. |
Returns
sqlalchemy.Column | The configured Column object. |
ForeignKey(self, target: str, dtype: Any = None, on_delete: str = CASCADE, nullable: bool = False) -> Column
Defines a SQLAlchemy Column that acts as a Foreign Key.
Automatically sets nullable=True if on_delete is SET NULL.
Parameters
target | str | The target table and column name (e.g., 'users.id'). |
dtype | Any | The SQLAlchemy type of the foreign key column. Defaults to BigInteger. By default None. |
on_delete | str | The ON DELETE SQL action (e.g., 'CASCADE', 'SET NULL', 'RESTRICT'). By default CASCADE. |
nullable | bool | If True, allows NULL values. Overridden if on_delete='SET NULL'. By default False. |
Returns
sqlalchemy.Column | The configured Column object with ForeignKey constraint. |
Relationship(self, model: str, back_populates: str = None) -> Any
Defines a SQLAlchemy relationship property.
Sets passive_deletes=True to allow the database to handle cascading
deletes efficiently when used with ondelete='CASCADE' in the ForeignKey.
Parameters
model | str | The target model class name or string name. |
back_populates | str | The name of the relationship property on the target model. By default None. |
Returns
sqlalchemy.orm.RelationshipProperty | The configured relationship object. |