SQLAlchemy-SQLSchema

This library provides the capability to specify the active SQL-schema through the maintain_schema() context manager/decorator, which will set the active schema and maintain it until its exit. Multiple transactions with commits and/or rollbacks may take place inside the context manager without affecting the active SQL-schema.

The main use case for such functionality is when schemas need to be changed dynamically. This is often needed when using schemas to implement multi-tenancy, i.e. when identical tables are placed in multiple different schemas but an end-user has access to only one of them. This allows to maximize re-use of code and database operations while providing isolation.

SQL-schemas are supported by PostgreSQL, Oracle, and Microsoft SQL Server. PostgreSQL and Oracle support temporarily setting the schema for the active session. The MS SQL method (with ALTER USER <user> WITH DEFAULT SCHEMA <schema>) results in permanent changes for a user which is rather unsafe, and is not supported by this library.

Usage

Assuming you have hold of a Session, you can use maintain_schema() as a context manager:

from sqlalchemy_sqlschema import maintain_schema

with maintain_schema("my_schema", session):
    schema = session.execute("show search_path").scalar()
    assert schema == "my_schema"

    # the following query needs to find a `my_schema.my_model_table` table
    session.query(MyModel)

    # a rollback still maintains the schema
    session.rollback()
    assert session.execute("show search_path").scalar() == "my_schema"

Or as a decorator:

from sqlalchemy_sqlschema import maintain_schema

@maintain_schema("my_schema", session)
def query_data():
    assert session.execute("show search_path").scalar() == "my_schema"
    return session.query(MyModel).all()

Implementation

The SQL schema is set by using dialect-specific SQL clauses, of which only the PostgreSQL implementations are implemented. SQL Alchemy events are used to set the schema again right after a new transaction is started (which is needed since a rollback will reset the schema to the value it had before the transaction start).

API

sqlalchemy_sqlschema.maintain_schema(schema, session)

Context manager/decorator that will apply the SQL schema schema using the session. The schema will persist across different transactions, if these happen within the context manager’s body.

After the context manager exits, it will restore the SQL schema that was found to be active when it was entered.

The context manager can also be nested. Exiting the nested context manager will restore the SQL schema set by the outer context manager.

Example:

>>> assert session.execute('SHOW search_path').scalar() == 'public'
>>> with maintain_schema('new_schema', session):
>>>     assert session.execute('SHOW search_path').scalar() == 'new_schema'
>>>     # still maintained after a rollback
>>>     session.rollback()
>>>     assert session.execute('SHOW search_path').scalar() == 'new_schema'
>>> # back to original
>>> assert session.execute('SHOW search_path').scalar() == 'public'

Parameters:

• schema – str to be set as the SQL schema
• session – a Session which will be used to set the SQL schema

sqlalchemy_sqlschema.sql.get_schema()

An executable SQL Alchemy clause that can be used to get the active SQL schema.

See also set_schema().

Example:

>>> stmt = get_schema()
>>> stmt
'SHOW search_path'
>>> session.execute(stmt).scalar()
'public'

sqlalchemy_sqlschema.sql.set_schema(schema)

An executeble SQL Alchemy clause that can be sed to set the active SQL schema.

See also get_schema().

Example:

>>> stmt = set_schema('new_schema')
>>> stmt
'SET search_path TO new_schema'
>>> session.execute(stmt)
>>> assert session.execute(get_schema()).scalar() == 'new_schema'

Parameters:schema – str to be set as the new SQL schema

Web Application Example

A useful scenario is when a web application redirects different users to different SQL schemas.

First, we need a way to know the SQL schema per user. In this case, it is a column on the user table directly:

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    id = Column(Integer, primary_key=True)
    schema = Column(String)

Let’s setup our web application to set the right SQL schema. We are using Flask and Flask-Login to get access to the current_user:

from flask import Flask, jsonify
from flask_login import current_user

app = Flask(__name__)

@app.route("/api/data")
def data():
    with maintain_schema(current_user.schema, session)
        data = session.query(MyModel).all()
        return jsonify(data=data)

In the example above, the table of MyModel needs to exist in the selected schema otherwise the query will fail. Setting the schema also means that the user is “locked” in that schema and cannot see any other tables in different schemas.

The above can be achieved more succinctly with a decorator:

from decorator import decorator

@decorator
def set_user_schema(f, *args, **kwargs):
    """Call `maintain_schema` with the current_user's schema."""
    with maintain_schema(current_user.schema, session):
        return f(*args, **kwargs)

@app.route("/api/data")
@set_user_schema
def data():
    data = session.query(Data).all()
    return jsonify(data=data)

The same decorator could be applied in a View:

from flask import View

class SecretView(View):
    methods = ['GET']
    decorators = [set_user_schema]

    def dispatch_request(self):
        # ...