Skip to content

Database

commons.database

SQLModel/SQLAlchemy building blocks shared across apps.

Reusable persistence primitives that sit below any one app's ORM models:

  • Ids (:func:new_id, from :mod:commons.id_operators) — time-ordered UUIDv7 strings for new rows.
  • Column length bounds (:data:ID_LENGTH, :data:CODE_LENGTH, :data:NAME_LENGTH) — pick a bound so String columns stay indexable on SQL Server.
  • Column types (:class:EnumString, :class:PydanticJSON, :class:UTCDateTime, :class:UnicodeCode, :class:UnicodeName) — TypeDecorators handed to Field(sa_type=...).
  • Table mixins (:class:EntityMixin, :class:HistoryMixin) — a UUIDv7-PK operational entity base and its temporal history-table companion.
  • Unit-of-work ports (:class:UnitOfWork, :class:UnitOfWorkFactory) — abstract transactional-scope contracts, generic over an app-defined repository bundle T.
  • SQL unit of work (:class:SqlUnitOfWork, :class:SqlUnitOfWorkFactory) — a Session-per-scope implementation of those ports; the request actor is injected (current_user_provider) so the library stays free of any web-framework dependency.

These are engine-portable (SQLite for tests, SQL Server / Azure SQL in production); the SQL-Server-only pieces (filtered indexes) are gated by dialect.

Text columns on SQL Server. SQLModel makes str columns VARCHAR(n). With the default collation of Azure SQL (code page 1252), VARCHAR keeps German umlauts but stores other characters (ł, ş, Cyrillic, Greek, emoji) as ?, silently. Use VARCHAR (plain Field(max_length=...)) for ASCII values: ids, codes, hashes, e-mail addresses, URLs, keys. Use NVARCHAR (:class:UnicodeCode, :class:UnicodeName, or Unicode(n) in a Column) for names and free text people enter. create_all doesn't change existing tables; convert a column with e.g. ALTER TABLE registered_users ALTER COLUMN display_name NVARCHAR(128) NOT NULL (keep the column's length and nullability, and drop and recreate an index on it first).

Modules:

Name Description
column_lengths

Shared max_length bounds for String columns.
entity
history
types

SQLModel-compatible column types (TypeDecorator subclasses).
unit_of_work
unit_of_work_sql

Classes:

Name Description
EntityMixin

Base for an operational entity table: a generated id primary key and a derived table name.
HistoryMixin

Base for a history table, generic over the entity ORM E it mirrors (HistoryMixin[CountryORM]).
EnumString

A length-bounded String for Literal-valued columns (status / kind / mode / license).
PydanticJSON

JSON column that round-trips any Pydantic-validatable type T via a TypeAdapter.
UnicodeCode

A length-bounded Unicode column (CODE_LENGTH) for short text people enter, e.g. a case number.
UnicodeName

A length-bounded Unicode column (NAME_LENGTH) for names and titles people enter.
UTCDateTime

Store tz-aware datetimes as naive UTC (engine-portable). Read back naive (repo convention).
UnitOfWork

Port for a single transactional scope (one with block).
UnitOfWorkFactory

Port that mints a fresh :class:UnitOfWork per call - safe to share process-wide.
SqlUnitOfWork

One Session and one transaction for a single with block.
SqlUnitOfWorkFactory

Mints a fresh :class:SqlUnitOfWork per with block.

Functions:

Name Description
new_id

Return a new id: a time-ordered UUIDv7 as a 36-character string.

Attributes:

Name Type Description
CODE_LENGTH
ID_LENGTH
NAME_LENGTH
CommitFunction

CODE_LENGTH module-attribute 

CODE_LENGTH = 64

ID_LENGTH module-attribute 

ID_LENGTH = 36

NAME_LENGTH module-attribute 

NAME_LENGTH = 255

CommitFunction module-attribute 

CommitFunction = Callable[[], Awaitable[None]]

__all__ module-attribute 

__all__ = [
    "CODE_LENGTH",
    "ID_LENGTH",
    "NAME_LENGTH",
    "CommitFunction",
    "EntityMixin",
    "EnumString",
    "HistoryMixin",
    "PydanticJSON",
    "SqlUnitOfWork",
    "SqlUnitOfWorkFactory",
    "UTCDateTime",
    "UnicodeCode",
    "UnicodeName",
    "UnitOfWork",
    "UnitOfWorkFactory",
    "new_id",
]

EntityMixin

Bases: SQLModel

Base for an operational entity table: a generated id primary key and a derived table name.

Inheriting it (with table=True) gives the table a UUIDv7-string id PK (commons.id_operators.new_id) and a name derived from the class (CountryORM -> countries) via declared_attr (so it runs at class-creation time). Override the name by setting __tablename__ explicitly on the subclass.

Methods:

Name Description
__tablename__

Attributes:

Name Type Description
id str

id class-attribute instance-attribute 

id: str = Field(
    default_factory=new_id,
    primary_key=True,
    min_length=36,
    max_length=36,
)

__tablename__ 

__tablename__() -> str
Source code in src/commons/database/entity.py 
27
28
29
@declared_attr.directive
def __tablename__(cls) -> str:  # type: ignore  # declared_attr vs SQLModel's __tablename__ class var
    return _entity_tablename(cls.__name__)

HistoryMixin

Bases: SQLModel

Base for a history table, generic over the entity ORM E it mirrors (HistoryMixin[CountryORM]).

Inheriting it (with table=True) derives the table name from the class and adds the open-interval index — both via declared_attr so they run at class-creation time. Override the name by setting __tablename__ explicitly on the subclass.

Methods:

Name Description
from_entity

Open a new history row mirroring entity's columns, stamped with the actor and time.
close_open_interval

Close this entity's open interval (history_valid_to IS NULL), if any, at timestamp.
at_timestamp

The version of entity_id valid at timestamp (half-open [from, to)), or None.
all_at_timestamp

Every entity's version valid at timestamp (half-open [from, to)).
record

Append a new version: close the previous open interval, then add an open row mirroring entity.
__tablename__
__table_args__

Attributes:

Name Type Description
id str
history_valid_from datetime
history_valid_to datetime | None
history_changed_by_user_displayname str
history_changed_by_user_id str
history_deleted bool

id class-attribute instance-attribute 

id: str = Field(primary_key=True, max_length=ID_LENGTH)

history_valid_from class-attribute instance-attribute 

history_valid_from: datetime = Field(
    primary_key=True, sa_type=UTCDateTime
)

history_valid_to class-attribute instance-attribute 

history_valid_to: datetime | None = Field(
    default=None, sa_type=UTCDateTime
)

history_changed_by_user_displayname class-attribute instance-attribute 

history_changed_by_user_displayname: str = Field(
    sa_type=UnicodeName
)

history_changed_by_user_id class-attribute instance-attribute 

history_changed_by_user_id: str = Field(
    max_length=NAME_LENGTH
)

history_deleted class-attribute instance-attribute 

history_deleted: bool = Field(default=False)

from_entity classmethod 

from_entity(
    user: User,
    entity: E,
    timestamp: datetime,
    *,
    deleted: bool = False
) -> Self

Open a new history row mirroring entity's columns, stamped with the actor and time.

Copies the columns the entity and this history table share (by name); entity-only columns (e.g. an operational soft-delete flag) are ignored.

Source code in src/commons/database/history.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
@classmethod
def from_entity(cls, user: User, entity: E, timestamp: datetime, *, deleted: bool = False) -> Self:
    """Open a new history row mirroring ``entity``'s columns, stamped with the actor and time.

    Copies the columns the entity and this history table share (by name); entity-only columns
    (e.g. an operational soft-delete flag) are ignored.
    """
    shared = {key: value for key, value in entity.model_dump().items() if key in cls.model_fields}
    return cls(
        **shared,
        history_valid_from=timestamp,
        history_changed_by_user_displayname=user.display_name,
        history_changed_by_user_id=user.id,
        history_deleted=deleted,
    )

close_open_interval classmethod 

close_open_interval(
    session: Session,
    entity_id: str,
    timestamp: datetime | None = None,
) -> datetime

Close this entity's open interval (history_valid_to IS NULL), if any, at timestamp.

Returns the timestamp used, so the caller can reuse it as the new version's history_valid_from (giving contiguous half-open intervals [from, to)).

Source code in src/commons/database/history.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@classmethod
def close_open_interval(cls, session: Session, entity_id: str, timestamp: datetime | None = None) -> datetime:
    """Close this entity's open interval (``history_valid_to IS NULL``), if any, at ``timestamp``.

    Returns the timestamp used, so the caller can reuse it as the new version's ``history_valid_from``
    (giving contiguous half-open intervals ``[from, to)``).
    """
    timestamp = timestamp or utc_now()
    open_row = session.exec(
        select(cls).where(
            col(cls.id) == entity_id,
            col(cls.history_valid_to).is_(None),
        )
    ).first()
    if open_row is not None:
        open_row.history_valid_to = timestamp
    return timestamp

at_timestamp classmethod 

at_timestamp(
    session: Session,
    entity_id: str,
    timestamp: datetime,
    *,
    include_deleted: bool = False
) -> Self | None

The version of entity_id valid at timestamp (half-open [from, to)), or None.

Intervals never overlap, so at most one row matches. A tombstone (history_deleted) counts as "did not exist" and yields None by default; pass include_deleted=True to get the tombstone row itself (e.g. to show "deleted at …").

Source code in src/commons/database/history.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
@classmethod
def at_timestamp(
    cls,
    session: Session,
    entity_id: str,
    timestamp: datetime,
    *,
    include_deleted: bool = False,
) -> Self | None:
    """The version of ``entity_id`` valid at ``timestamp`` (half-open ``[from, to)``), or ``None``.

    Intervals never overlap, so at most one row matches. A tombstone (``history_deleted``) counts
    as "did not exist" and yields ``None`` by default; pass ``include_deleted=True`` to get the
    tombstone row itself (e.g. to show "deleted at …").
    """
    stmt = select(cls).where(
        col(cls.id) == entity_id,
        col(cls.history_valid_from) <= timestamp,
        or_(
            col(cls.history_valid_to).is_(None),
            col(cls.history_valid_to) > timestamp,
        ),
    )
    if not include_deleted:
        stmt = stmt.where(col(cls.history_deleted) == false())
    return session.exec(stmt).one_or_none()

all_at_timestamp classmethod 

all_at_timestamp(
    session: Session,
    timestamp: datetime,
    *,
    include_deleted: bool = False
) -> Sequence[Self]

Every entity's version valid at timestamp (half-open [from, to)).

The set-valued companion to :meth:at_timestamp: one row per entity that existed at timestamp (tombstones excluded unless include_deleted). Backs cross-aggregate reverse lookups that must run against a historical snapshot rather than the live tables.

Source code in src/commons/database/history.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
@classmethod
def all_at_timestamp(
    cls,
    session: Session,
    timestamp: datetime,
    *,
    include_deleted: bool = False,
) -> Sequence[Self]:
    """Every entity's version valid at ``timestamp`` (half-open ``[from, to)``).

    The set-valued companion to :meth:`at_timestamp`: one row per entity that
    existed at ``timestamp`` (tombstones excluded unless ``include_deleted``).
    Backs cross-aggregate reverse lookups that must run against a historical
    snapshot rather than the live tables.
    """
    stmt = select(cls).where(
        col(cls.history_valid_from) <= timestamp,
        or_(
            col(cls.history_valid_to).is_(None),
            col(cls.history_valid_to) > timestamp,
        ),
    )
    if not include_deleted:
        stmt = stmt.where(col(cls.history_deleted) == false())
    return session.exec(stmt).all()

record classmethod 

record(
    session: Session,
    user: User,
    entity: E,
    *,
    timestamp: datetime | None = None,
    deleted: bool = False
) -> Self

Append a new version: close the previous open interval, then add an open row mirroring entity.

The single per-entity write path — it always closes before opening, so the "at most one open interval per id" invariant can't be broken by forgetting a step. timestamp defaults to now; pass one to backdate or to align several entities to the same instant. Returns the added row.

Source code in src/commons/database/history.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
@classmethod
def record(
    cls,
    session: Session,
    user: User,
    entity: E,
    *,
    timestamp: datetime | None = None,
    deleted: bool = False,
) -> Self:
    """Append a new version: close the previous open interval, then add an open row mirroring ``entity``.

    The single per-entity write path — it always closes before opening, so the "at most one open
    interval per id" invariant can't be broken by forgetting a step. ``timestamp`` defaults to now;
    pass one to backdate or to align several entities to the same instant. Returns the added row.
    """
    timestamp = cls.close_open_interval(session, entity.id, timestamp)
    row = cls.from_entity(user, entity, timestamp, deleted=deleted)
    session.add(row)
    return row

__tablename__ 

__tablename__() -> str
Source code in src/commons/database/history.py 
147
148
149
@declared_attr.directive
def __tablename__(cls) -> str:  # type: ignore  # declared_attr vs SQLModel's __tablename__ class var
    return _history_tablename(cls.__name__)

__table_args__ 

__table_args__() -> tuple[Index, ...]
Source code in src/commons/database/history.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
@declared_attr.directive
def __table_args__(cls) -> tuple[Index, ...]:
    # at most one open interval (history_valid_to IS NULL) per id; name follows the resolved
    # table name, whether derived above or overridden on the subclass.
    name = cls.__tablename__
    return (
        Index(
            f"uq_{name}_open",
            "id",
            unique=True,
            sqlite_where=text("history_valid_to IS NULL"),
            mssql_where=text("history_valid_to IS NULL"),
            postgresql_where=text("history_valid_to IS NULL"),
        ),
    )

EnumString

Bases: TypeDecorator[str]

A length-bounded String for Literal-valued columns (status / kind / mode / license).

A TypeDecorator rather than String(ENUM_LENGTH) so it can be handed to SQLModel's sa_type as a type (which SQLModel instantiates into a VARCHAR(ENUM_LENGTH) column) instead of an instance — SQLModel types sa_type as type[Any], so passing an instance fails type checking.

Attributes:

Name Type Description
impl
cache_ok

impl class-attribute instance-attribute 

impl = String(ENUM_LENGTH)

cache_ok class-attribute instance-attribute 

cache_ok = True

PydanticJSON 

PydanticJSON(data_type: type[T], **kwargs: Any)

Bases: TypeDecorator[T], Generic[T]

JSON column that round-trips any Pydantic-validatable type T via a TypeAdapter.

T can be a model, list[Model], dict[str, Model], etc. Writes dump to JSON-native values (so the JSON impl can encode them); reads rebuild validated instances rather than handing back raw dicts.

Methods:

Name Description
process_bind_param
process_result_value

Attributes:

Name Type Description
impl
cache_ok
data_type
Source code in src/commons/database/types.py 
66
67
68
69
70
def __init__(self, data_type: type[T], **kwargs: Any) -> None:
    super().__init__(**kwargs)
    # Stored under the ``__init__`` arg name so SQLAlchemy's cache key picks it up.
    self.data_type = data_type
    self._adapter: TypeAdapter[T] = TypeAdapter(data_type)

impl class-attribute instance-attribute 

impl = JSON

cache_ok class-attribute instance-attribute 

cache_ok = True

data_type instance-attribute 

data_type = data_type

process_bind_param 

process_bind_param(value: Any, dialect: Dialect) -> Any
Source code in src/commons/database/types.py 
72
73
74
75
76
77
78
79
def process_bind_param(self, value: Any, dialect: Dialect) -> Any:
    if value is None:
        return None
    # ``validate_python`` first so we serialise canonically whether handed model
    # instances (the normal ORM path) or raw dicts (e.g. a history mixin that
    # snapshots rows via ``model_dump``); table models skip validation on init,
    # so dicts would otherwise reach here and trip ``dump_python``'s warnings.
    return self._adapter.dump_python(self._adapter.validate_python(value), mode="json")

process_result_value 

process_result_value(
    value: Any, dialect: Dialect
) -> T | None
Source code in src/commons/database/types.py 
81
82
83
84
def process_result_value(self, value: Any, dialect: Dialect) -> T | None:
    if value is None:
        return None
    return self._adapter.validate_python(value)

UnicodeCode

Bases: TypeDecorator[str]

A length-bounded Unicode column (CODE_LENGTH) for short text people enter, e.g. a case number.

NVARCHAR(64) on SQL Server, where VARCHAR stores only code page 1252 and turns other characters into ?; VARCHAR(64) elsewhere. Hand it to Field(sa_type=...) as a type, like EnumString (sa_type=Unicode would ignore max_length and give an unindexable NVARCHAR(max)).

Attributes:

Name Type Description
impl
cache_ok

impl class-attribute instance-attribute 

impl = Unicode(CODE_LENGTH)

cache_ok class-attribute instance-attribute 

cache_ok = True

UnicodeName

Bases: TypeDecorator[str]

A length-bounded Unicode column (NAME_LENGTH) for names and titles people enter.

NVARCHAR(255) on SQL Server (510 bytes, still indexable), VARCHAR(255) elsewhere; see UnicodeCode.

Attributes:

Name Type Description
impl
cache_ok

impl class-attribute instance-attribute 

impl = Unicode(NAME_LENGTH)

cache_ok class-attribute instance-attribute 

cache_ok = True

UTCDateTime

Bases: TypeDecorator[datetime]

Store tz-aware datetimes as naive UTC (engine-portable). Read back naive (repo convention).

Methods:

Name Description
process_bind_param

Attributes:

Name Type Description
impl
cache_ok

impl class-attribute instance-attribute 

impl = DateTime

cache_ok class-attribute instance-attribute 

cache_ok = True

process_bind_param 

process_bind_param(
    value: datetime | None, dialect: Dialect
) -> datetime | None
Source code in src/commons/database/types.py 
93
94
95
96
97
98
def process_bind_param(self, value: datetime | None, dialect: Dialect) -> datetime | None:
    if value is None:
        return None
    if value.tzinfo is not None:
        value = value.astimezone(timezone.utc).replace(tzinfo=None)
    return value

UnitOfWork

Bases: ABC

Port for a single transactional scope (one with block).

Entering opens a transaction and returns the repository bundle plus a commit callable; leaving the block rolls back unless commit was awaited. Generic over T - the app-defined bundle of repositories the scope exposes.

Methods:

Name Description
__enter__
__exit__

__enter__ abstractmethod 

__enter__() -> tuple[T, CommitFunction]
Source code in src/commons/database/unit_of_work.py 
18
19
@abc.abstractmethod
def __enter__(self) -> tuple[T, CommitFunction]: ...

__exit__ abstractmethod 

__exit__(*args) -> None
Source code in src/commons/database/unit_of_work.py 
21
22
@abc.abstractmethod
def __exit__(self, *args) -> None: ...  # rolls back if not committed

UnitOfWorkFactory

Bases: ABC

Port that mints a fresh :class:UnitOfWork per call - safe to share process-wide.

for_user opens a unit of work attributed to the request user; for_system one attributed to the system user (background / bulk work). Each call yields its own session/scope, so concurrent requests never share transactional state or audit actor. The factory itself is not a context manager - always go through one of these methods so the object you enter is the object you exit.

Methods:

Name Description
for_user

Open a unit of work attributed to the request user (sets the audit actor).
for_system

Open a unit of work with history writing off (bulk seeding).
__enter__
__exit__

for_user abstractmethod 

for_user(user: User) -> UnitOfWork[T]

Open a unit of work attributed to the request user (sets the audit actor).

Source code in src/commons/database/unit_of_work.py 
35
36
37
38
@abc.abstractmethod
def for_user(self, user: User) -> "UnitOfWork[T]":
    """Open a unit of work attributed to the request user (sets the audit actor)."""
    ...

for_system abstractmethod 

for_system() -> UnitOfWork[T]

Open a unit of work with history writing off (bulk seeding).

Source code in src/commons/database/unit_of_work.py 
40
41
42
43
@abc.abstractmethod
def for_system(self) -> "UnitOfWork[T]":
    """Open a unit of work with history writing off (bulk seeding)."""
    ...

__enter__ abstractmethod 

__enter__() -> tuple[T, CommitFunction]
Source code in src/commons/database/unit_of_work.py 
45
46
@abc.abstractmethod
def __enter__(self) -> tuple[T, CommitFunction]: ...

__exit__ abstractmethod 

__exit__(*args) -> None
Source code in src/commons/database/unit_of_work.py 
48
49
@abc.abstractmethod
def __exit__(self, *args) -> None: ...  # rolls back if not committed

SqlUnitOfWork 

SqlUnitOfWork(
    session_factory: sessionmaker,
    repositories_factory: Callable[[Session, User], T],
    *,
    user: User
)

Bases: UnitOfWork[T]

One Session and one transaction for a single with block.

A fresh instance per with (minted by :class:SqlUnitOfWorkFactory), so its session and audit actor are never shared between concurrent requests - commit and __exit__ always act on this block's own session. __enter__ opens the session and returns the repository bundle plus an async commit; __exit__ rolls back (if commit was never awaited) and closes.

Methods:

Name Description
__enter__
__exit__
Source code in src/commons/database/unit_of_work_sql.py 
20
21
22
23
24
25
26
27
28
29
30
def __init__(
    self,
    session_factory: sessionmaker,
    repositories_factory: Callable[[Session, User], T],
    *,
    user: User,
) -> None:
    self._session_factory = session_factory
    self._repositories_factory = repositories_factory
    self._user = user
    self._session: Session | None = None

__enter__ 

__enter__() -> tuple[T, CommitFunction]
Source code in src/commons/database/unit_of_work_sql.py 
32
33
34
35
def __enter__(self) -> tuple[T, CommitFunction]:
    self._session = cast(Session, self._session_factory())

    return self._repositories_factory(self._session, self._user), self._commit

__exit__ 

__exit__(*args: object) -> None
Source code in src/commons/database/unit_of_work_sql.py 
37
38
39
40
41
def __exit__(self, *args: object) -> None:
    if self._session is not None:
        self._session.rollback()  # no-op once commit has flushed + committed
        self._session.close()
        self._session = None

SqlUnitOfWorkFactory 

SqlUnitOfWorkFactory(
    session_factory: sessionmaker,
    repositories_factory: Callable[[Session, User], T],
    current_user_provider: Callable[
        [], User
    ] = _system_user,
)

Bases: UnitOfWorkFactory[T]

Mints a fresh :class:SqlUnitOfWork per with block.

The factory is stateless and safe to share process-wide (it lives on the single composition root): for_user / for_system each return a new unit of work, so the object you enter is the object you exit. No per-request state lives on the shared instance, so concurrent requests cannot clobber each other's session or audit actor.

Use with uow.for_user(user) as (repos, commit) so every history row is attributed; for_system() turns history off (bulk seeding). Entering the factory itself as a context manager resolves the actor via current_user_provider - inject one that reads your web framework's request context (the default attributes work to the system user). Keeping actor resolution injected is deliberate: it keeps this library free of any web-framework dependency.

Methods:

Name Description
for_user

A unit of work attributed to user (sets the history actor).
for_system

A unit of work with history writing off (bulk seeding).
__enter__
__exit__
Source code in src/commons/database/unit_of_work_sql.py 
68
69
70
71
72
73
74
75
76
77
def __init__(
    self,
    session_factory: sessionmaker,
    repositories_factory: Callable[[Session, User], T],
    current_user_provider: Callable[[], User] = _system_user,
) -> None:
    self._session_factory = session_factory
    self._repositories_factory = repositories_factory
    self._current_user_provider = current_user_provider
    self._current: SqlUnitOfWork[T] | None = None

for_user 

for_user(user: User) -> SqlUnitOfWork[T]

A unit of work attributed to user (sets the history actor).

Source code in src/commons/database/unit_of_work_sql.py 
82
83
84
def for_user(self, user: User) -> SqlUnitOfWork[T]:
    """A unit of work attributed to ``user`` (sets the history actor)."""
    return self._unit_of_work(user=user)

for_system 

for_system() -> SqlUnitOfWork[T]

A unit of work with history writing off (bulk seeding).

Source code in src/commons/database/unit_of_work_sql.py 
86
87
88
def for_system(self) -> SqlUnitOfWork[T]:
    """A unit of work with history writing off (bulk seeding)."""
    return self._unit_of_work(user=SystemUser())

__enter__ 

__enter__() -> tuple[T, CommitFunction]
Source code in src/commons/database/unit_of_work_sql.py 
90
91
92
def __enter__(self) -> tuple[T, CommitFunction]:
    self._current = self._unit_of_work(user=self._current_user_provider())
    return self._current.__enter__()

__exit__ 

__exit__(*args: object) -> None
Source code in src/commons/database/unit_of_work_sql.py 
94
95
96
97
def __exit__(self, *args: object) -> None:
    if self._current is not None:
        self._current.__exit__(*args)
        self._current = None

new_id 

new_id() -> str

Return a new id: a time-ordered UUIDv7 as a 36-character string.

Ids created one after another in a process compare in creation order, so as primary keys they are appended to the index instead of landing anywhere in it. They reveal when they were created; don't use them as secrets.

Returns:

Type Description
str

The id, e.g. "019a3b2c-7d4e-7f10-8a2b-3c4d5e6f7a8b".
Source code in src/commons/id_operators.py 
 6
 7
 8
 9
10
11
12
13
14
15
def new_id() -> str:
    """Return a new id: a time-ordered UUIDv7 as a 36-character string.

    Ids created one after another in a process compare in creation order, so as primary keys they are appended to
    the index instead of landing anywhere in it. They reveal when they were created; don't use them as secrets.

    Returns:
        The id, e.g. ``"019a3b2c-7d4e-7f10-8a2b-3c4d5e6f7a8b"``.
    """
    return str(uuid7())