core.orm.session_manager

Attributes

_T
CONNECTION_LIFETIME

Classes

SessionManager

Holds sessions and creates schemas before binding sessions to schemas.

Functions

query_schemas(→ Iterator[str])

Yields all schemas or the ones with the given namespace.

Module Contents

core.orm.session_manager._T

core.orm.session_manager.CONNECTION_LIFETIME = 3600

core.orm.session_manager.query_schemas(connection: Connection | Engine, namespace: str | None = None) → Iterator[str]

Yields all schemas or the ones with the given namespace.

class core.orm.session_manager.SessionManager(dsn: str, base: type[Any], engine_config: dict[str, Any] | None = None, session_config: dict[str, Any] | None = None, pool_config: dict[str, Any] | None = None)

Holds sessions and creates schemas before binding sessions to schemas.

Threadsafe in theory, but not tested or well thought out. No global state though, so two different session managers will manage different sessions/schemas.

_valid_schema_name

_invalid_prefixes

_reserved_schemas

_thread_bound

dsn

bases

created_schemas: set[str]

current_schema: str | None = None

on_insert

on_update

on_delete

required_extensions

created_extensions: set[str]

engine

session_factory

register_engine(engine: sqlalchemy.engine.Engine) → None

Takes the given engine and registers it with the schema switching mechanism. Maybe used to register external engines with the session manager.

If used like this, make sure to call bind_session() before using the session provided by the external engine.

register_session(session: sqlalchemy.orm.session.Session) → None

Takes the given session and registers it with zope.sqlalchemy and various orm events.

activate() → None

Sets the currently active session_manager - this is basically a global variable we require because we hook into the orm/query events where we don’t know yet which session is going to be used and therefore cannot be sure about the used session manager either

For example, the Document model below needs access to the current session to get the current locale, but since it is instantiated without any session information, this can’t be known without a global variable:

document = Document(localized_title=”Das Dokument”)

We can only work around that with a global variable:

session_manager.activate() document = Document(localized_title=”Das Dokument”)

As a result session managers need to be activated, or they can’t be used (at least for translated columns). To do so in tests, use:

session_manager.activate()

To ease the use of testing the last session_manager is however also automatically activated when the schema is set, which covers most use-cases outside of testing

deactivate() → None

Sets the currently active session manager to None, if it is equal to self.

classmethod set_active(session_manager: SessionManager | None) → None

classmethod get_active() → SessionManager | None

set_locale(default_locale: str | None, current_locale: str | None) → None

Sets the default locale and the current locale so it may be shared with the translated ORM columns.

Note that the locales may be NONE. It is up to the consumer of these settings to assert their existence.

_scopefunc() → tuple[threading.Thread, str | None]

Returns the scope of the scoped_session used to create new sessions. Relies on self.current_schema being set before the session is created.

This function is as internal as they come and it exists only because we otherwise would have to create different session factories for each schema.

dispose() → None

Closes the connection to the server and cleans up. This only needed for testing.

is_valid_schema(schema: str | None) → bool

Returns True if the given schema looks valid enough to be created on the database with no danger of SQL injection or other unwanted sideeffects.

set_current_schema(schema: str) → None

Sets the current schema in use. The current schema is then used to bind the session to a schema. Note that this can’t be done in a functional way. We need the current schema to generate a new scope.

I would very much prefer to bind this to the session but this is not at all straight-forward with SQLAlchemy.

I tried a solution like this one, but it’s not good enough, because it still relies on some kind of global stage, even if it’s set locally.

Ideally a value could be bound to the session and an event would trigger every time the connection is used with that session. We could then set the schema on the connection every time that happens.

For now, the global option is okay though, because in practice we only set the schema once per request and we don’t do threading anyway.

create_schema(schema: str, validate: bool = True) → None

Creates the given schema. If said schema exists, expect this method to throw an error. If you use set_current_schema(), this is invoked automatically if needed. So you usually shouldn’t have to care about this function.

create_schema_if_not_exists(schema: str, validate: bool = True) → None

Creates the given schema if it doesn’t exist yet.

bind_session(session: _S) → _S

Bind the session to the current schema.

session() → sqlalchemy.orm.session.Session

Returns a new session or an existing session. Sessions with different schemas are kept independent, though they might reuse each others connections.

This means that a session retrieved thusly:

mgr = SessionManager('postgres://...')
mgr.set_current_schema('foo')
session = mgr.session()

Will not see objects attached to this session:

mgr.set_current_schema('bar')
session = mgr.session()

list_schemas(namespace: str | None = None) → list[str]

Returns a tuple containing all schemas defined in the current database.

is_schema_found_on_database(schema: str) → bool

Returns True if the given schema exists on the database.

create_required_extensions() → None

Creates the required extensions once per lifetime of the manager.

ensure_schema_exists(schema: str) → None

Makes sure the schema exists on the database. If it doesn’t, it is created.