Tortoise ORM or Sqlalchemy

● Tortoise ORM：异步，API和Django ORM 大多类似集成Pydantic；多用于asgi starlette / sanic / FastAPI...
● Sqlalchemy：支持异步，Flask / Bottle/FastAPI ，集成Pydantic方案 可搜索SQLModel

SQLAlchemy + FastAPI

"""
Tortoise ORM & SQLAlchemy
"""

# SQlAlchemy
from sqlalchemy import Column, Integer, String, DateTime, func, select
from sqlalchemy.orm import declarative_base

Base = declarative_base()


class SqlAlchemyTable(Base):
    __tablename__ = "a"

    id = Column(Integer, primary_key=True)
    data = Column(String)
    create_date = Column(DateTime, server_default=func.now())


from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

engine = create_async_engine("sqlite+aiosqlite:///basic.sqlite3")
async_session = sessionmaker(
    engine, expire_on_commit=True, class_=AsyncSession
)


def get_db():
    session = async_session()
    try:
        yield session
    finally:
        session.close()


from fastapi import FastAPI, Depends
from pydantic import BaseModel

class Response(BaseModel):
    id: int
    data: str

app = FastAPI(title="SQLAlchemy")


@app.get("/", response_model=Response)
async def index():
    # 视图里面实例db对象
    async with async_session() as session:
        result = await session.execute(select(SqlAlchemyTable).order_by(SqlAlchemyTable.id))
        return result.scalars().first().__dict__ # 第一条


@app.get("/index")
async def index1(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(SqlAlchemyTable).order_by(SqlAlchemyTable.id))
    return result.scalars().all()


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("__main__:app", reload=True)

image.png

Tortoise ORM + FastAPI

from tortoise import models, fields
from tortoise.contrib.fastapi import register_tortoise
from tortoise.contrib.pydantic import pydantic_model_creator

class TortoiseOrm(models.Model):
    data = fields.CharField(max_length=30)
    create_date = fields.DatetimeField(auto_now_add=True, null=True)

    class Meta:
        table = 'a'

Response = pydantic_model_creator(TortoiseOrm, name="OutPut", exclude=("create_date",))


from fastapi import FastAPI
app = FastAPI(title="SQLAlchemy & TortoiseORM")

register_tortoise(
    app,
    db_url="sqlite://basic.sqlite3",
    modules={"models": ["__main__"]},
    add_exception_handlers=True,
)

@app.get("/", response_model=Response)
async def index():
    return await TortoiseOrm.all().order_by("id").first()


@app.get("/index")
async def index1():
    return await TortoiseOrm.all().order_by("id")
    
if __name__ == '__main__':
    import uvicorn
    
    uvicorn.run("__main__:app", reload=True)

image.png

image.png