fastapi Tortoise-ORM、SQLAlchemy、peewee使用orm查询的例子

以下是使用 FastAPI 中 Tortoise-ORM、SQLAlchemy 和 peewee 进行 ORM 查询的示例。

## Tortoise-ORM

from fastapi import FastAPI
from tortoise import fields
from tortoise.contrib.fastapi import register_tortoise, HTTPNotFoundError
from tortoise.models import Model
from tortoise import Tortoise

class User(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(50)
    email = fields.CharField(50)

    class Meta:
        table = "users"

app = FastAPI()

@app.on_event("startup")
async def startup():
    await Tortoise.init(
        db_url="sqlite://db.sqlite3",
        modules={"models": ["main"]}
    )
    await Tortoise.generate_schemas()

@app.on_event("shutdown")
async def shutdown():
    await Tortoise.close_connections()

@app.get("/users")
async def get_users():
    users = await User.all()
    return users

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = await User.get_or_none(id=user_id)
    if user is None:
        raise HTTPNotFoundError
    return user

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

## SQLAlchemy

from fastapi import FastAPI
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50))
    email = Column(String(50))

engine = create_engine("sqlite:///db.sqlite3")
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

app = FastAPI()

@app.get("/users")
async def get_users():
    db = SessionLocal()
    users = db.query(User).all()
    db.close()
    return users

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    db = SessionLocal()
    user = db.query(User).filter(User.id == user_id).first()
    db.close()
    if user is None:
        raise HTTPNotFoundError
    return user

## peewee

from fastapi import FastAPI
from peewee import SqliteDatabase, Model, CharField, IntegerField
from playhouse.shortcuts import model_to_dict

db = SqliteDatabase("db.sqlite3")

class User(Model):
    id = IntegerField(primary_key=True)
    name = CharField()
    email = CharField()

    class Meta:
        database = db
        table_name = "users"

app = FastAPI()

@app.on_event("startup")
def startup():
    db.connect()
    db.create_tables([User])

@app.on_event("shutdown")
def shutdown():
    db.close()

@app.get("/users")
async def get_users():
    users = [model_to_dict(user) for user in User.select()]
    return users

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = User.get_or_none(User.id == user_id)
    if user is None:
        raise HTTPNotFoundError
    return model_to_dict(user)

注意：以上示例中的代码仅用于演示 ORM 查询的基本用法，并且没有进行错误处理。在实际应用中，你应该根据需要添加适当的错误处理和安全性检查。