SQLAlchemy(非同期)のCRUD

1.はじめに

最近、PythonのWebフレームワークであるFastAPIを使っています。
SQLAlchemyのORMapperを使って、PostgreSQLのデータベースを操作するプログラムを書きましたので、備忘録も兼ねて本記事に実装方法を残します。

今回は、以下のようなテーブルを題材にします。

Testテーブル

カラム名	型	備考
id	integer	PK,Autoincrement
name	character(10)
created_at	date	Default: 現在時刻

Model

models/test.py

from sqlalchemy import Column, Integer, String, DateTime
from datetime import datetime
from database.database import Base

class Test(Base):
    __tablename__ = "test"
    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    name = Column(String(10), index=True)
    created_at = Column(DateTime, default=datetime.now)

2.CRUD(Controller)

controllers/test.py

from typing import Any, List, Tuple
from sqlalchemy import select, insert, update, delete
from sqlalchemy.engine import Result
from sqlalchemy.ext.asyncio import AsyncSession
from models.test import Test as test_model
import datetime

# データを登録(C)
async def create(db: AsyncSession, name: str) -> Any:
    result = await db.execute(insert(test_model).values(name=name, created_at=datetime.datetime.now()))
    await db.commit()
    return result

# データを全件検索(R)
async def read(db: AsyncSession) -> List[Tuple[int, str, datetime.datetime]]:
    result: Result = await db.execute(select(
        test_model.id,
        test_model.name,
        test_model.created_at
    )
    return result.all()

# データを更新(U)
async def update(db: AsyncSession, id: int, name: str) -> Any:
    result = await db.execute(update(test_model).values(name=name, created_at=datetime.datetime.now()).where(user_model.id == id))
    await db.commit()
    return result

# データを削除(D)
async def delete(db: AsyncSession, id: int) -> Any:
    result = await db.execute(delete(test_model).where(test_model.id == id))
    await db.commit()
    return result

3.終わりに

今回は、CR(全件検索)UDの書き方について記事にしました。
Rについては、await db.execute(...).filter(...)と書くことで、特定の条件のレコードのみ取得することができます。
本記事はルーティングについて触れていませんので、(気が向いたら)別記事としてあげたいと思います。