Async Session#

The sqlservice.async_session.AsyncSession class is the main interface for asynchronously querying a database. It extends sqlalchemy.ext.asyncio.AsyncSession to provide additional methods for common ORM queries.

Executing Select Statements#

The following methods will execute a select-statement and return results. If the select-statement is based on ORM-models (e.g. User.select()), then the results will automatically be converted into ORM-model instances.

from datetime import datetime, timedelta

import sqlalchemy as sa

session = db.session()

users = await session.all(User.select())
users = await session.all(
    sa.text("SELECT * FROM users WHERE timestamp > :timestamp"),
    params={"timestamp": datetime.now() - timedelta(days=1)}
)

Saving Models#

The sqlservice.async_session.AsyncSession.save() and sqlservice.async_session.AsyncSession.save_all() method can be used to save model instances. These method differ from AsyncSession.add and AsyncSession.add_all in that they will automatically upsert records based on the model’s primary key(s). This allows models that were not loaded from the database to be automatically inserted or updated independent of the database backend.

user_1 = User(id=4, name='Max')
user_2 = User(id=8, name='Jack')
user_3 = User(id=1, name='Wes'')

saved_user_1 = await session.save(user_1)
assert save_user_1 is user_1

saved_users_2_3 = await session.save_all([user_2, user_3])
assert saved_users_2_3[0] is user_2
assert saved_users_2_3[1] is user_3