FastAPI并没有对具体的ORM做出限制,你可以使用你喜欢的任何ORM。
常用的ORM有:
SQLAlchemy
Peewee
Tortoise-ORM
Pony ORM
Ormar
GINO
其中,SQLAlchemy 是比较常用的ORM,它不仅支持关系型数据库,还支持NoSQL数据库。FastAPI也有很好的支持。在FastAPI官方文档中也有使用SQLAlchemy的示例。如果你之前已经熟悉了SQLAlchemy,那么可以考虑在FastAPI中使用它。
如果你想使用异步ORM,Tortoise-ORM和GINO是不错的选择。Tortoise-ORM是一种基于asyncio和databases的ORM,使用类似Django的语法。GINO是基于SQLAlchemy core的异步ORM,可以在FastAPI中非常方便地使用。
在选择ORM时,你可以根据你的项目需求、熟悉程度和团队喜好等因素进行选择。
用 fastapi 结合 pony orm 写一套学生信息管理系统
下面是使用FastAPI结合Pony ORM的示例代码,实现学生信息的增删改查功能:
创建一个main.py文件,编写以下代码:
from fastapi import FastAPI, HTTPException
from pony.orm import Database, Required, Json
from pydantic import BaseModel
# 初始化Pony ORM
db = Database()
# 定义学生数据模型
class Student(db.Entity):
name = Required(str)
age = Required(int)
address = Required(Json)
# 设置数据库连接
db.bind(provider='sqlite', filename='students.sqlite', create_db=True)
db.generate_mapping(create_tables=True)
# 初始化FastAPI
app = FastAPI()
# 定义输入数据模型
class StudentIn(BaseModel):
name: str
age: int
address: dict
# 定义输出数据模型
class StudentOut(BaseModel):
id: int
name: str
age: int
address: dict
# 定义路由,获取学生列表
@app.get("/students")
def get_students():
students = Student.select()
return [StudentOut(id=s.id, name=s.name, age=s.age, address=s.address) for s in students]
# 定义路由,获取单个学生
@app.get("/students/{id}")
def get_student(id: int):
student = Student.get(id=id)
if not student:
raise HTTPException(status_code=404, detail="Student not found")
return StudentOut(id=student.id, name=student.name, age=student.age, address=student.address)
# 定义路由,新增学生
@app.post("/students")
def create_student(student: StudentIn):
s = Student(name=student.name, age=student.age, address=student.address)
db.commit()
return StudentOut(id=s.id, name=s.name, age=s.age, address=s.address)
# 定义路由,更新学生
@app.put("/students/{id}")
def update_student(id: int, student: StudentIn):
s = Student.get(id=id)
if not s:
raise HTTPException(status_code=404, detail="Student not found")
s.name = student.name
s.age = student.age
s.address = student.address
db.commit()
return StudentOut(id=s.id, name=s.name, age=s.age, address=s.address)
# 定义路由,删除学生
@app.delete("/students/{id}")
def delete_student(id: int):
s = Student.get(id=id)
if not s:
raise HTTPException(status_code=404, detail="Student not found")
s.delete()
db.commit()
return {"message": "Student deleted successfully"}
# 运行FastAPI应用
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
这个示例代码中,我们使用Pony ORM来定义学生数据模型,并创建了一个SQLite数据库来存储数据。我们使用FastAPI来定义路由和输入/输出数据模型,实现了学生信息的增删改查功能。
启动应用后,可以通过http://localhost:8000/docs查看API文档
用 fastapi 结合 sqlalchemy 写一套学生信息管理系统
from typing import List
from fastapi import FastAPI, HTTPException
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 创建 SQLAlchemy 引擎和会话工厂
SQLALCHEMY_DATABASE_URL = "postgresql://user:password@localhost/database"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# 创建 SQLAlchemy 声明基类
Base = declarative_base()
# 定义 Student 模型
class Student(Base):
__tablename__ = "students"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
age = Column(Integer)
email = Column(String, unique=True, index=True)
# 创建 FastAPI 应用
app = FastAPI()
# 获取数据库会话
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# 定义创建学生的 API
@app.post("/students/", response_model=Student)
def create_student(student: Student, db: Session = Depends(get_db)):
db.add(student)
db.commit()
db.refresh(student)
return student
# 定义获取所有学生的 API
@app.get("/students/", response_model=List[Student])
def get_students(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
students = db.query(Student).offset(skip).limit(limit).all()
return students
# 定义获取单个学生的 API
@app.get("/students/{student_id}", response_model=Student)
def get_student(student_id: int, db: Session = Depends(get_db)):
student = db.query(Student).filter(Student.id == student_id).first()
if not student:
raise HTTPException(status_code=404, detail="Student not found")
return student
# 定义更新学生的 API
@app.put("/students/{student_id}", response_model=Student)
def update_student(student_id: int, student: StudentUpdate, db: Session = Depends(get_db)):
db_student = db.query(Student).filter(Student.id == student_id).first()
if not db_student:
raise HTTPException(status_code=404, detail="Student not found")
for var, value in vars(student).items():
if value is not None:
setattr(db_student, var, value)
db.commit()
db.refresh(db_student)
return db_student
# 定义删除学生的 API
@app.delete("/students/{student_id}")
def delete_student(student_id: int, db: Session = Depends(get_db)):
db_student = db.query(Student).filter(Student.id == student_id).first()
if not db_student:
raise HTTPException(status_code=404, detail="Student not found")
db.delete(db_student)
db.commit()
return {"message": "Student deleted"}
上面的代码中,我们使用 create_engine 函数创建一个 SQLAlchemy 引擎,并使用 sessionmaker 函数创建一个会话工厂。然后我们定义了一个 Student 模型,其中包含了学生的姓名、年龄和电子邮件地址。
下面是使用 FastAPI 和 Tortoise-ORM 写一套学生信息管理系统的示例代码:
from fastapi import FastAPI, HTTPException
from tortoise.contrib.fastapi import register_tortoise
from tortoise.fields import IntField, CharField, DateField
from tortoise.models import Model
# Define a student model using Tortoise-ORM
class Student(Model):
id = IntField(pk=True)
name = CharField(max_length=50)
birth_date = DateField()
class Meta:
table = "students"
# Create a FastAPI instance
app = FastAPI()
# Add endpoint for creating a new student
@app.post("/students")
async def create_student(name: str, birth_date: date):
student = await Student.create(name=name, birth_date=birth_date)
return {"id": student.id, "name": student.name, "birth_date": student.birth_date}
# Add endpoint for getting a student by id
@app.get("/students/{student_id}")
async def get_student(student_id: int):
student = await Student.filter(id=student_id).first().values()
if not student:
raise HTTPException(status_code=404, detail="Student not found")
return student
# Add endpoint for getting all students
@app.get("/students")
async def get_all_students():
students = await Student.all().values()
return students
# Register the Tortoise-ORM models with FastAPI
register_tortoise(
app,
db_url="sqlite://db.sqlite3",
modules={"models": ["main"]},
generate_schemas=True,
add_exception_handlers=True,
)
大家可以简单的感受一下几种ORM库的区别