aiohttp-sqlalchemy 0.11.0

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;

• support for different types of request handlers.

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 aiohttp import web
import aiohttp_sqlalchemy
from aiohttp_sqlalchemy import sa_bind, sa_session
from datetime import datetime
import sqlalchemy as sa
from sqlalchemy import orm
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine


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


class MyModel(Base):
    __tablename__ = 'my_table'
    id = sa.Column(sa.Integer, primary_key=True)
    timestamp = sa.Column(sa.DateTime(), default=datetime.now)


async def main(request):
    async with sa_session(request).bind.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    async with sa_session(request).begin():
        sa_session(request).add_all([MyModel()])
        result = await sa_session(request).execute(sa.select(MyModel))
        data = {r.id: r.timestamp.isoformat() for r in result.scalars()}
        return web.json_response(data)


app = web.Application()

engine = create_async_engine('sqlite+aiosqlite:///')
Session = orm.sessionmaker(engine, AsyncSession)
aiohttp_sqlalchemy.setup(app, [sa_bind(Session)])

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

if __name__ == '__main__':
    web.run_app(app)