GitLocker: The Coding Marketplace

Description:

aiohttpsqlalchemy 1.0.post0

SQLAlchemy 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 as ahsa

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):
sa_session = ahsa.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()

ahsa.setup(app, [
ahsa.bind('sqlite+aiosqlite:///'),
])
await ahsa.init_db(app, metadata)

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

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