# ORM
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 Python's async/await keywords. Some possible choices include：
Integration in to your Sanic application is fairly simple:
# Mayim
Mayim ships with an extension for Sanic Extensions (opens new window), which makes it super simple to get started with Sanic. It is certainly possible to run Mayim with Sanic without the extension, but it is recommended because it handles all of the lifecycle events (opens new window) and dependency injections (opens new window).
# Dependencies
First, we need to install the required dependencies. See Mayim docs (opens new window) for the installation needed for your DB driver.
pip install sanic-ext
pip install mayim[postgres]
# Define ORM Model
Mayim allows you to use whatever you want for models. Whether it is dataclasses (opens new window), pydantic (opens new window), attrs (opens new window), or even just plain dict objects. Since it works very nicely out of the box with Pydantic (opens new window), that is what we will use here.
# ./models.py
from pydantic import BaseModel
class City(BaseModel):
id: int
name: str
district: str
population: int
class Country(BaseModel):
code: str
name: str
continent: str
region: str
capital: City
# Define SQL
If you are unfamiliar, Mayim is different from other ORMs in that it is one-way, SQL-first. This means you define your own queries either inline, or in a separate .sql file, which is what we will do here.
-- ./queries/select_all_countries.sql
SELECT country.code,
country.name,
country.continent,
country.region,
(
SELECT row_to_json(q)
FROM (
SELECT city.id,
city.name,
city.district,
city.population
) q
) capital
FROM country
JOIN city ON country.capital = city.id
ORDER BY country.name ASC
LIMIT $limit OFFSET $offset;
# Create Sanic App and Async Engine
We need to create the app instance and attach the SanicMayimExtension with any executors.
# ./server.py
from sanic import Sanic, Request, json
from sanic_ext import Extend
from mayim.executor import PostgresExecutor
from mayim.extensions import SanicMayimExtension
from models import Country
class CountryExecutor(PostgresExecutor):
async def select_all_countries(
self, limit: int = 4, offset: int = 0
) -> list[Country]:
...
app = Sanic("Test")
Extend.register(
SanicMayimExtension(
executors=[CountryExecutor],
dsn="postgres://...",
)
)
# Register Routes
Because we are using Mayim's extension for Sanic, we have the automatic CountryExecutor injection into the route handler. It makes for an easy, type-annotated development experience.
@app.get("/")
async def handler(request: Request, executor: CountryExecutor):
countries = await executor.select_all_countries()
return json({"countries": [country.dict() for country in co
# Send Requests
curl 'http://127.0.0.1:8000'
{"countries":[{"code":"AFG","name":"Afghanistan","continent":"Asia","region":"Southern and Central Asia","capital":{"id":1,"name":"Kabul","district":"Kabol","population":1780000}},{"code":"ALB","name":"Albania","continent":"Europe","region":"Southern Europe","capital":{"id":34,"name":"Tirana","district":"Tirana","population":270000}},{"code":"DZA","name":"Algeria","continent":"Africa","region":"Northern Africa","capital":{"id":35,"name":"Alger","district":"Alger","population":2168000}},{"code":"ASM","name":"American Samoa","continent":"Oceania","region":"Polynesia","capital":{"id":54,"name":"Fagatogo","district":"Tutuila","population":2323}}]}
# 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 now sqlalchemy and aiomysql are needed.
pip install -U sqlalchemy
pip install -U aiomysql
# Define ORM Model
ORM model creation remains the same.
# ./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")
# 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)
# 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.
# ./server.py
from contextvars import ContextVar
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import sessionmaker
_sessionmaker = sessionmaker(bind, AsyncSession, expire_on_commit=False)
_base_model_session_ctx = ContextVar("session")
@app.middleware("request")
async def inject_session(request):
request.ctx.session = _sessionmaker()
request.ctx.session_ctx_token = _base_model_session_ctx.set(request.ctx.session)
@app.middleware("response")
async def close_session(request, response):
if hasattr(request.ctx, "session_ctx_token"):
_base_model_session_ctx.reset(request.ctx.session_ctx_token)
await request.ctx.session.close()
# 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
@app.post("/user")
async def create_user(request):
session = request.ctx.session
async with session.begin():
car = Car(brand="Tesla")
person = Person(name="foo", cars=[car])
session.add_all([person])
return json(person.to_dict())
@app.get("/user/<pk:int>")
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 'http://127.0.0.1:8000/user'
{"name":"foo","cars":[{"brand":"Tesla"}]}
curl --location --request GET 'http://127.0.0.1:8000/user/1'
{"name":"foo","cars":[{"brand":"Tesla"}]}
# Tortoise-ORM
# Dependencies
tortoise-orm's dependency is very simple, you just need install tortoise-orm.
pip install -U tortoise-orm
# Define ORM Model
If you are familiar with Django, you should find this part very familiar.
# ./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}"
# 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__)
register_tortoise(
app, db_url="mysql://root:root@localhost/test", modules={"models": ["models"]}, generate_schemas=True
)
# Register Routes
# ./main.py
from models import Users
from sanic import Sanic, response
@app.route("/user")
async def list_all(request):
users = await Users.all()
return response.json({"users": [str(user) for user in users]})
@app.route("/user/<pk:int>")
async def get_user(request, pk):
user = await Users.query(pk=pk)
return response.json({"user": str(user)})
if __name__ == "__main__":
app.run(port=5000)
# Send Requests
curl --location --request POST 'http://127.0.0.1:8000/user'
{"users":["I am foo", "I am bar"]}
curl --location --request GET 'http://127.0.0.1:8000/user/1'
{"user": "I am foo"}