Skip to content

Working with the session#

The Session is the mean to communicate with the database. There are two main ways to use it:

Use the scoped session db.s#

The “scoped_session” is really a proxy to a session automatically scoped to the current thread.

This allows having a global session, so the session can be shared without the need to pass it explicitly.

A scoped session is the recommended way to work in a web application, however, you must remember to call db.s.remove() the session at the end of the request.

Use your framework’s “on request end” hook, to do that. For example, in Flask:

@app.teardown_request
def remove_db_scoped_session(error=None):
    db.s.remove()

The db.s.remove() method close the current session and dispose it. A new session will be created when db.s is called again.

Outside a web request cycle, like in a background job, you must call manually call db.s.remove() at the end.

Instantiate db.Session#

You can use a context manager:

with db.Session() as dbs:
    # work with the session here

When the session is created in this way, a database transaction is automatically initiated when required, and the dbs.flush(), dbs.commit(), and dbs.rollback() methods can be used as needed.

The session is automatically closed and returned to the session pool when the context manager block ends.

You can also create it without a context manager and close it manually:

dbs = db.Session():
# work with the session here
dbs.close()

Instantiate db.Session is the recommended way to work when the session is not shared like in a command-line script.

API#

SQLAlchemy default Session class has the method .get(Model, pk) to query and return a record by its primary key.

This class extends the sqlalchemy.orm.Session class with some useful active-record-like methods and a pagination helper.

class sqla_wrapper.Session(bind=None, autoflush=True, future=False, expire_on_commit=True, autocommit=False, twophase=False, binds=None, enable_baked_queries=True, info=None, query_cls=None)
all(self, Model, **attrs)

Returns all the object found with these attributes.

The filtering is done with a simple .filter_by() so is limited to “equality” comparisons against the columns of the model. Also, there is no way to sort the results. If you need sorting or more complex filtering, you are better served using a db.select().

Examples:

users = db.s.all(User)
users = db.s.all(User, deleted=False)
users = db.s.all(User, account_id=123, deleted=False)
create(self, Model, **attrs)

Creates a new object and adds it to the session.

This is a shortcut for:

obj = Model(**attrs)
db.s.add(obj)
db.s.flush()

Note that this does a db.s.flush(), so you must later call db.s.commit() to persist the new object.

Example:

new_user = db.s.create(User, email='foo@example.com')
db.s.commit()
first(self, Model, **attrs)

Returns the first object found with these attributes or None if there isn’t one.

The filtering is done with a simple .filter_by() so is limited to “equality” comparisons against the columns of the model. Also, there is no way to sort the results. If you need sorting or more complex filtering, you are better served using a db.select().

Examples:

user = db.s.first(User)
user = db.s.first(User, deleted=False)
first_or_create(self, Model, **attrs)

Tries to find an object and if none exists, it tries to create a new one first. Use this method when you expect the object to already exists but want to create it in case it doesn’t.

This does a db.s.flush(), so you must later call db.s.commit() to persist the new object (in case one has been created).

Examples:

user1 = db.s.first_or_create(User, email='foo@example.com')
user2 = db.s.first_or_create(User, email='foo@example.com')
user1 is user2
create_or_first(self, Model, **attrs)

Tries to create a new object, and if it fails because already exists, return the first it founds. For this to work one or more of the attributes must be unique so it does fail, otherwise you will be creating a new different object.

Use this method when you expect that the object does not exists but want to avoid an exception in case it does.

This does a db.s.flush(), so you must later call db.s.commit() to persist the new object (in case one has been created).

Examples:

user1 = db.s.create_or_first(User, email='foo@example.com')
user2 = db.s.create_or_first(User, email='foo@example.com')
user1 is user2
paginate(self, query, *, total, page=1, per_page=20, padding=0)

Returns a Paginator of the query results. Note that you must calculate the total number of unpaginated results first.

Arguments:

  • query: A select() statement for all items.
  • total: Total number of items. You must pre-calculate this value.
  • page: Number of the current page (first page is 1) It can be a number, a string with a number, or the strings “first” or “last”.
  • per_page: Max number of items to display on each page.
  • padding: Number of elements of the previous and next page to show. For example, if per_page is 10 and padding is 2, every page will show 14 items, the first two from the previous page and the last two for the next one. This extra items will be repeated again on their own pages.

Example:

query = select(User) \
    .where(User.deleted.is_(None)) \
    .order_by(User.created_at)

total = db.s.scalar(
    select(func.count(User.id))
    .where(User.deleted.is_(None))
)

pag = db.s.paginate(query, total=total, page=1, per_page=20)

As always, I recommend reading the official SQLAlchemy tutorial to learn more how to work with the session.

Back to top