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[source]
core.orm.session_manager.CONNECTION_LIFETIME[source]
core.orm.session_manager.query_schemas(connection: Connection | Engine, namespace: str | None = None) Iterator[str][source]

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)[source]

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[source]
_invalid_prefixes[source]
_reserved_schemas[source]
_thread_bound[source]
register_engine(engine: sqlalchemy.engine.Engine) None[source]

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[source]

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

activate() None[source]

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[source]

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

classmethod set_active(session_manager: SessionManager | None) None[source]
classmethod get_active() SessionManager | None[source]
set_locale(default_locale: str | None, current_locale: str | None) None[source]

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][source]

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[source]

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

is_valid_schema(schema: str | None) bool[source]

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[source]

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[source]

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[source]

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

bind_session(session: _S) _S[source]

Bind the session to the current schema.

session() sqlalchemy.orm.session.Session[source]

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][source]

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

is_schema_found_on_database(schema: str) bool[source]

Returns True if the given schema exists on the database.

create_required_extensions() None[source]

Creates the required extensions once per lifetime of the manager.

ensure_schema_exists(schema: str) None[source]

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