aiohttp-sqlalchemy 0.17.3

SQLAlchemy 1.4 / 2.0 support for aiohttp.

SQLAlchemy 1.4 / 2.0 support for AIOHTTP.

The library provides the next features:

• initializing asynchronous sessions through a middlewares;

• initializing asynchronous sessions through a decorators;

• simple access to one asynchronous session by default key;

• preventing attributes from being expired after commit by default;

• support different types of request handlers;

• support nested applications.

Documentation

https://aiohttp-sqlalchemy.readthedocs.io

Installation

pip install aiohttp-sqlalchemy

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
from aiohttp_sqlalchemy import sa_session

metadata = sa.MetaData()
Base = orm.declarative_base(metadata=metadata)


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):
    db_session = sa_session(request)

    async with db_session.begin():
        db_session.add(MyModel())
        result = await db_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()

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

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

    return app


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