How do I use SQLAlchemy with Sanic ?

All ORM tools can work with Sanic, but non-async ORM tool have a impact on Sanic performance. There are some orm packages who support

At present, there are many ORMs that support asynchronicity. Two of the more common libraries are:

Integration in to your Sanic application is fairly simple:

# SQLAlchemy

Because SQLAlchemy 1.4 (opens new window) has added native support for asyncio, Sanic can finally work well with SQLAlchemy. Be aware that this functionality is still considered beta by the SQLAlchemy project.

# Dependencies

First, we need to install the required dependencies. In the past, the dependencies installed were sqlalchemy and pymysql, but nowsqlalchemy' andaiomysql` are needed.

pip install -U sqlalchemy
pip install -U aiomysql
# ./models.py
from sqlalchemy import INTEGER, Column, ForeignKey, String
from sqlalchemy.orm import declarative_base, relationship
Base = declarative_base()
class BaseModel(Base):
    __abstract__ = True
    id = Column(INTEGER(), primary_key=True)
class Person(BaseModel):
    __tablename__ = "person"
    name = Column(String())
    cars = relationship("Car")
    def to_dict(self):
        return {"name": self.name, "cars": [{"brand": car.brand} for car in self.cars]}
class Car(BaseModel):
    __tablename__ = "car"
    brand = Column(String())
    user_id = Column(ForeignKey("person.id"))
    user = relationship("Person", back_populates="cars")

# Define ORM Model

ORM model creation remains the same.

# Create Sanic App and Async Engine

Here we use mysql as the database, and you can also choose PostgreSQL/SQLite. Pay attention to changing the driver from aiomysql to asyncpg/aiosqlite.

# ./server.py
from sanic import Sanic
from sqlalchemy.ext.asyncio import create_async_engine
app = Sanic("my_app")
bind = create_async_engine("mysql+aiomysql://root:root@localhost/test", echo=True)
# ./server.py
from contextvars import ContextVar
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import sessionmaker
_base_model_session_ctx = ContextVar("session")
async def inject_session(request):
    request.ctx.session = sessionmaker(bind, AsyncSession, expire_on_commit=False)()
    request.ctx.session_ctx_token = _base_model_session_ctx.set(request.ctx.session)
async def close_session(request, response):
    if hasattr(request.ctx, "session_ctx_token"):
        await request.ctx.session.close()

# Register Middlewares

The request middleware creates an usable AsyncSession object and set it to request.ctx and _base_model_session_ctx.

Thread-safe variable _base_model_session_ctx helps you to use the session object instead of fetching it from request.ctx.

# Register Routes

According to sqlalchemy official docs, session.query will be legacy in 2.0, and the 2.0 way to query an ORM object is using select.

# ./server.py
from sqlalchemy import select
from sqlalchemy.orm import selectinload
from sanic.response import json
from models import Car, Person
async def create_user(request):
    session = request.ctx.session
    async with session.begin():
        car = Car(brand="Tesla")
        person = Person(name="foo", cars=[car])
    return json(person.to_dict())
async def get_user(request, pk):
    session = request.ctx.session
    async with session.begin():
        stmt = select(Person).where(Person.id == pk).options(selectinload(Person.cars))
        result = await session.execute(stmt)
        person = result.scalar()
    if not person:
        return json({})
    return json(person.to_dict())

# Send Requests

curl --location --request POST ''
curl --location --request GET ''

# Tortoise-ORM

# Dependencies

tortoise-orm's dependency is very simple, you just need install tortoise-orm.

pip install -U tortoise-orm
# ./models.py
from tortoise import Model, fields
class Users(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(50)
    def __str__(self):
        return f"I am {self.name}"

# Define ORM Model

If you are familiar with Django, you should find this part very familiar.

# Create Sanic App and Async Engine

Tortoise-orm provides a set of registration interface, which is convenient for users, and you can use it to create database connection easily.

# ./main.py
from models import Users
from tortoise.contrib.sanic import register_tortoise
app = Sanic(__name__)
    app, db_url="mysql://root:root@localhost/test", modules={"models": ["models"]}, generate_schemas=True

# ./main.py
from models import Users
from sanic import Sanic, response
async def list_all(request):
    users = await Users.all()
    return response.json({"users": [str(user) for user in users]})
async def get_user(request, pk):
    user = await Users.query(pk=pk)
    return response.json({"user": str(user)})
if __name__ == "__main__":

# Register Routes

# Send Requests

curl --location --request POST ''
{"users":["I am foo", "I am bar"]}
curl --location --request GET ''
{"user": "I am foo"}
MIT Licensed
Copyright © 2018-present Sanic Community Organization

~ Made with ❤️ and ☕️ ~