whisker.sqlalchemy

whisker/sqlalchemy.py


Copyright © 2011-2020 Rudolf Cardinal (rudolf@pobox.com).

This file is part of the Whisker Python client library.

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.


SQLAlchemy helper functions for Whisker tasks.

whisker.sqlalchemy.database_is_mysql(dbsettings: Dict[str, str]) → bool[source]

Checks the URL in dbsettings['url']: is it a MySQL database?

whisker.sqlalchemy.database_is_postgresql(dbsettings: Dict[str, str]) → bool[source]

Checks the URL in dbsettings['url']: is it a PostgreSQL database?

whisker.sqlalchemy.database_is_sqlite(dbsettings: Dict[str, str]) → bool[source]

Checks the URL in dbsettings['url']: is it an SQLite database?

whisker.sqlalchemy.get_database_engine(settings: Dict[str, Any], unbreak_sqlite_transactions: bool = True, pool_pre_ping: bool = True) → sqlalchemy.engine.base.Engine[source]

Get an SQLAlchemy database Engine from a simple definition.

Parameters:
  • settings

    a dictionary with the following keys:

    • url: a string
    • echo: a boolean
    • connect_args: a dictionary

    All are passed to SQLAlchemy’s create_engine() function.

  • unbreak_sqlite_transactions – hook in events to unbreak SQLite transaction support? (Detailed in sqlalchemy/dialects/sqlite/pysqlite.py; see “Serializable isolation / Savepoints / Transactional DDL”.)
  • pool_pre_ping – boolean; requires SQLAlchemy 1.2
Returns:

an SQLAlchemy Engine

whisker.sqlalchemy.get_database_engine_session_thread_scope(settings: Dict[str, Any], readonly: bool = False, autoflush: bool = True) → Tuple[sqlalchemy.engine.base.Engine, sqlalchemy.orm.session.Session][source]

Gets a thread-scoped SQLAlchemy Engine and Session.

Parameters:
  • settings – passed to get_database_engine()
  • readonly – make the session read-only?
  • autoflush – passed to sessionmaker()
Returns:

(engine, session)

Return type:

tuple

whisker.sqlalchemy.get_database_session_thread_scope(*args, **kwargs) → sqlalchemy.orm.session.Session[source]

Gets a thread-scoped SQLAlchemy Session.

Parameters:
Returns:

the session

whisker.sqlalchemy.get_database_session_thread_unaware(settings: Dict[str, Any]) → sqlalchemy.orm.session.Session[source]

Returns an SQLAlchemy database session.

Warning

DEPRECATED: this function is not thread-aware.

Parameters:settings – passed to get_database_engine()
Returns:an SQLAlchemy Session
whisker.sqlalchemy.noflush_readonly(*args, **kwargs) → None[source]

Does nothing, and is thereby used to block a database session flush.

whisker.sqlalchemy.session_scope_thread_unaware(settings: Dict[str, Any]) → Generator[[sqlalchemy.orm.session.Session, None], None][source]

Context manager to provide an SQLAlchemy database session (which executes a COMMIT on success or a ROLLBACK on failure).

Warning

DEPRECATED: this function is not thread-aware.

Parameters:settings – passed to get_database_session_thread_unaware()
Yields:an SQLAlchemy Session
whisker.sqlalchemy.session_thread_scope(settings: Dict[str, Any], readonly: bool = False) → Generator[[sqlalchemy.orm.session.Session, None], None][source]

Context manager to provide a thread-safe SQLAlchemy database session (which executes a COMMIT on success or a ROLLBACK on failure).

Parameters:
Yields:

an SQLAlchemy Session