Skip to content

Instantly share code, notes, and snippets.

@dunossauro
Created May 12, 2021 12:52
Show Gist options
  • Save dunossauro/075cf06bc9dc7e16ddfa8717d6ee9c41 to your computer and use it in GitHub Desktop.
Save dunossauro/075cf06bc9dc7e16ddfa8717d6ee9c41 to your computer and use it in GitHub Desktop.
Flask 2.0 async support + SQLAlchemy 1.4 async ORM
from flask import Flask, request
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.future import select
from sqlalchemy.orm import declarative_base, sessionmaker
engine = create_async_engine('sqlite+aiosqlite:///./db.db')
async_session = sessionmaker(
engine, expire_on_commit=False, class_=AsyncSession
)
Base = declarative_base()
class A(Base):
__tablename__ = 'a'
id = Column(Integer, primary_key=True)
name = Column(String)
app = Flask(__name__)
@app.get('/person/<name>')
async def person_get(name):
async with async_session() as session:
query = await session.execute(select(A).where(A.name == name))
result = query.scalar()
return {"name": result.name, "id": result.id}
@app.post('/person/')
async def person_post():
req = request.get_json()
async with async_session() as session:
session.add(A(name=req['name']))
await session.commit()
return 'created', 201
app.run()
@MichaelDeMattos
Copy link

Could you please share your requirements.txt that you used for this example?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment