輕鬆流暢地使用 SQLAclhemy
pip insall fluent-alchemy
- 宣告 Models 並繼承
ActiveRecord
# models.py
from sqlalchemy.orm import DeclarativeBase
from fluent_alchemy import ActiveRecord
class Base(DeclarativeBase, ActiveRecord):
pass
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(BigInteger(), primary_key=True)
email: Mapped[str] = mapped_column(String(), unique=True)
password: Mapped[str] = mapped_column(String())
name: Mapped[str] = mapped_column(String(50))
state: Mapped[bool] = mapped_column(Boolean())
- 建立 SQLAclhemy Engine, 並指派給
ActiveRecord
內的 session handler
from sqlalchemy import create_engine
from models import Base
engine = create_engine("sqlite://:memory:", echo=True)
Base.set_engine(engine)
- 開始使用 model 操作 database !
from models import User
users = User.all()
- 使用完畢後,釋放 Session 資源
from models import Base
Base.remove_scoped_session()
利用 QueryBuilder
來處理 SQLAlchemy 的 select()
query statement
-
Create
from models import User user = User.create( name="Justin", email="[email protected]", password="NioWe9Wn#+" ) # or user = User( name="Justin", email="[email protected]", password="NioWe9Wn#+" ) user.save()
-
Read
-
Find by id
user = User.find(1)
-
只回傳特定欄位
user = User.select(User.id, User.name, User.email).first()
-
透過
where
增加查詢條件user = User.where(User.email == "[email protected]").first()
users = User.where(User.state.is_(True)).get()
-
-
Update
user = User.find(1) user.passwod = "6xjVAY$p&D" user.save()
-
Delete
user = User.find(1) user.delete()
-
Pagenate
# setting page number and rows count per page pagination = User.paginate(page=1, per_page=15) """ { "total": 100, "per_page": 15, "current_page": 1, "last_page": 7, "data": [ ... ], # ussers } """
讓指定的 Model class 繼承 TimestampMixin
,讓該 Model 補上 created_at
, updated_at
欄位。
from fluent_alchemy import ActiveRecord, TimestampMixin
class Base(DeclarativeBase, ActiveRecord):
pass
class User(Base, TimestampMixin):
__tablename__ = "users"
...
###
user = User.find(1)
print(user.created_at)
print(user.updated_at)
讓指定的 Model class 繼承 SoftDeleteMixin
,就可以讓該 Model 擁有 Soft delete 的能力。
from sqlalchemy.orm import DeclarativeBase
from fluent_alchemy import ActiveRecord, SoftDeleteMixin
class Base(DeclarativeBase, ActiveRecord):
pass
class User(Base, SoftDeleteMixin):
__tablename__ = "users"
id: Mapped[int] = mapped_column(BigInteger(), primary_key=True)
email: Mapped[str] = mapped_column(String(), unique=True)
password: Mapped[str] = mapped_column(String())
name: Mapped[str] = mapped_column(String(50))
state: Mapped[bool] = mapped_column(Boolean())
SoftDeleteMixin
會自動補上 deleted_at
欄位,依此欄位來處理 soft delete 的資料。
deleted_at: Mapped[Optional[datetime]] = mapped_column(TIMESTAMP(), nullable=True)
設定完成後,之後對此 Model 進行 Query 時,會在 statement 內的 WHERE 條件自動加上 deleted_at IS NULL
。
users_deleted = User.where(...).get(with_trashed=True)
user = User.find(1)
user.delete(force=True)
from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends
from app.models import BaseModel, User
def close_scoped_session():
"""
Remove the scoped session at the enf of every request.
"""
yield
BaseModel.remove_scoped_session()
@asynccontextmanager
async def lifespan(app: FastAPI):
"""
Set engine to the ScopedSessionHandler when FastAPI app started.
"""
BaseModel.set_engine(engine)
yield
app = FastAPI(
title="MyApp",
dependencies=[Depends(close_scoped_session)],
lifespan=lifespan
)
@app.get("/users")
def index():
return User.all()