Quickstart

Simple example

Install aiosqlite for work with sqlite3:

pip install aiosqlite

Copy and paste this code in a file and run:

from datetime import datetime

import sqlalchemy as sa
from aiohttp import web
from sqlalchemy import orm

import aiohttp_sqlalchemy as ahsa


class Base(orm.DeclarativeBase): ...


class MyModel(Base):
    __tablename__ = "my_table"

    pk = sa.Column(sa.Integer, primary_key=True)
    timestamp = sa.Column(sa.DateTime(), default=datetime.now)


async def main(request):
    sa_session = aiohttp_sqlalchemy.get_session(request)

    async with sa_session.begin():
        sa_session.add(MyModel())
        result = await sa_session.execute(sa.select(MyModel))
        result = result.scalars()

    data = {instance.pk: instance.timestamp.isoformat() for instance in result}
    return web.json_response(data)


async def app_factory():
    app = web.Application()

    aiohttp_sqlalchemy.setup(
        app,
        [
            aiohttp_sqlalchemy.bind("sqlite+aiosqlite:///"),
        ],
    )
    await aiohttp_sqlalchemy.init_db(app, Base.metadata)

    app.add_routes([web.get("/", main)])
    return app


if __name__ == "__main__":
    web.run_app(app_factory(), port=8087)

SQLAlchemy and Asyncio

See Asynchronous I/O (asyncio) section in SQLAlchemy 1.4 documentation.

More control in configuration

import aiohttp_sqlalchemy as ahsa
from sqlalchemy import orm
from sqlalchemy.ext.asyncio import async_sessionmaker

url = 'sqlite+aiosqlite:///'
engine = create_async_engine(url, echo=True)
Session = orm.async_sessionmaker(main_engine, class_=AsyncSession, expire_on_commit=False)

ahsa.setup(app, [
    ahsa.bind(Session),
])

Class based views

from aiohttp import web
import aiohttp_sqlalchemy as ahsa


class MyClassBasedView(web.View, ahsa.SAMixin):
    async def get(self):
        db_session = self.get_sa_session()

        async with db_session.begin():
            # some your code


ahsa.setup(app, [
    ahsa.bind(MainSession),
])
app.add_routes([web.view('/', MyClassBasedView)])