# ORM
SQLAlchemyをSanicと共に使用する方法?
すべてのORMツールはSanicで動作しますが、非同期ORMツールはSanicのパフォーマンスに影響します。 これをサポートするormパッケージがいくつかあります。
現在、非同期性をサポートするORMはたくさんあります。一般的なライブラリには、次の2つがあります。
のSanicアプリケーションへの統合は非常に簡単です。
# SQLAlchemy
SQLAlchemy 1.4関数 (opens new window)がasyncio
のネイティブサポートを追加したので、SanicはついにSQLAlchemyでうまく動作するようになりました。SQLAlchemyプロジェクトでは、この機能はまだベータと見なされていることに注意してください。
# 依存関係
まず、必要な依存関係をインストールする必要があります。以前は、インストールされる依存関係はsqlalchemy'と
pymysql'でしたが、現在はsqlalchemy'と
aiomysql'が必要です。
pip install -U sqlalchemy
pip install -U aiomysql
# ORMモデルの定義
ORMモデルの作成は同じままです。
# ./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")
# Sanicアプリケーションと非同期エンジンの作成
ここではデータベースとしてmysqlを使用しますが、PostgreSQL/SQLiteを選択することもできます。ドライバをaiomysql
から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)
# ミドルウェアの登録
リクエスト・ミドルウェアは、使用可能なAsyncSession
オブジェクトを作成し、それをrequest.ctx
および_base_model_session_ctx
に設定します。
スレッドセーフな変数_base_model_session_ctx
を使用すると、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()
# ルートの登録
sqlalchemyの公式ドキュメントによると、session.query
は2.0年にレガシーになり、ORMオブジェクトをクエリする2.0の方法は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())
# リクエストを送信
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
# 依存関係
tortoise-ormの依存関係は非常に単純で、tortoise-ormをインストールするだけです。
pip install -U tortoise-orm
# ORMモデルの定義
Djangoに精通している方であれば、この部分は非常に馴染みがあるはずです。
# ./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}"
# Sanicアプリケーションと非同期エンジンの作成
Tortoise-ormは、ユーザーにとって便利な登録インタフェースのセットを提供し、これを使用してデータベース接続を簡単に作成できます。
# ./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
)
# ルートの登録
# ./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)
# リクエストを送信
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"}