当前位置：极客笔记 > FastAPI > FastAPI 问答 > FastAPI + SQLAlchemy + Pydantic：如何处理多对多关系

FastAPI + SQLAlchemy + Pydantic：如何处理多对多关系

FastAPI + SQLAlchemy + Pydantic：如何处理多对多关系

在本文中，我们将介绍如何使用FastAPI配合SQLAlchemy和Pydantic处理多对多关系。多对多关系是数据库设计中常见的一种关系，它表示两个实体之间的复杂关联。我们将使用FastAPI框架、SQLAlchemy ORM和Pydantic模型来展示如何创建和操作多对多关系。

阅读更多：FastAPI 教程

数据库设计

在开始之前，我们需要设计一个适合展示多对多关系的数据库模型。假设我们正在构建一个学生管理系统，其中包含学生和课程两个实体。一个学生可以选择多门课程，而一门课程也可以有多名学生选择。这就是一个典型的多对多关系。

下面是我们的数据库模型设计：

from sqlalchemy import Column, ForeignKey, Integer, String, Table
from sqlalchemy.orm import relationship

from database import Base

student_course_association = Table(
    "student_course_association",
    Base.metadata,
    Column("student_id", Integer, ForeignKey("students.id")),
    Column("course_id", Integer, ForeignKey("courses.id"))
)

class Student(Base):
    __tablename__ = "students"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    courses = relationship(
        "Course",
        secondary=student_course_association,
        back_populates="students"
    )

class Course(Base):
    __tablename__ = "courses"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    students = relationship(
        "Student",
        secondary=student_course_association,
        back_populates="courses"
    )

以上代码定义了两个表，其中Student和Course模型表示了学生和课程实体。多对多关系在两个模型之间通过student_course_association表来关联。我们使用relationship函数来定义模型之间的关系，并指定了back_populates参数来实现双向关联。

创建API路由

接下来，我们使用FastAPI来创建处理多对多关系的API路由。我们将实现以下功能：
– 创建学生和课程
– 获取所有学生和课程
– 为学生添加课程
– 获取学生的课程
– 为课程添加学生
– 获取课程的学生

下面是我们的API路由实现：

from typing import List

from fastapi import FastAPI, HTTPException
from sqlalchemy.orm import Session

import models
import schemas
from database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/students/", response_model=schemas.Student)
def create_student(student: schemas.StudentCreate, db: Session = Depends(get_db)):
    db_student = models.Student(name=student.name)
    db.add(db_student)
    db.commit()
    db.refresh(db_student)
    return db_student

@app.get("/students/", response_model=List[schemas.Student])
def get_students(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    students = db.query(models.Student).offset(skip).limit(limit).all()
    return students

@app.post("/students/{student_id}/courses/{course_id}/")
def add_student_course(student_id: int, course_id: int, db: Session = Depends(get_db)):
    db_student = db.query(models.Student).filter(models.Student.id == student_id).first()
    db_course = db.query(models.Course).filter(models.Course.id == course_id).first()
    if not db_student or not db_course:
        raise HTTPException(status_code=404, detail="Student or course not found")
    db_student.courses.append(db_course)
    db.commit()
    return {"message": "Course added to student"}

@app.get("/students/{student_id}/courses/", response_model=List[schemas.Course])
def get_student_courses(student_id: int, db: Session = Depends(get_db)):
    db_student = db.query(models.Student).filter(models.Student.id == student_id).first()
    if not db_student:
        raise HTTPException(status_code=404, detail="Student not found")
    return db_student.courses

@app.post("/courses/{course_id}/students/{student_id}/")
def add_course_student(course_id: int, student_id: int, db: Session = Depends(get_db)):
    db_course = db.query(models.Course).filter(models.Course.id == course_id).first()
    db_student = db.query(models.Student).filter(models.Student.id == student_id).first()
    if not db_course or not db_student:
        raise HTTPException(status_code=404, detail="Course or student not found")
    db_course.students.append(db_student)
    db.commit()
    return {"message": "Student added to course"}

@app.get("/courses/{course_id}/students/", response_model=List[schemas.Student])
def get_course_students(course_id: int, db: Session = Depends(get_db)):
    db_course = db.query(models.Course).filter(models.Course.id == course_id).first()
    if not db_course:
        raise HTTPException(status_code=404, detail="Course not found")
    return db_course.students

以上代码定义了一系列API路由，用于处理学生和课程的创建、获取和关联。我们使用response_model参数来指定返回的数据模型。

创建Pydantic模型

为了在API路由中使用数据验证和序列化，我们使用Pydantic模型来定义输入和输出数据的结构。下面是我们的Pydantic模型定义：

from typing import List

from pydantic import BaseModel

class CourseBase(BaseModel):
    name: str

class CourseCreate(CourseBase):
    pass

class Course(CourseBase):
    id: int
    students: List["Student"] = []

    class Config:
        orm_mode = True

class StudentBase(BaseModel):
    name: str

class StudentCreate(StudentBase):
    pass

class Student(StudentBase):
    id: int
    courses: List[Course] = []

    class Config:
        orm_mode = True

以上代码定义了Course和Student两个Pydantic模型，分别用于输入和输出数据的验证和序列化。Course模型包含了一个students字段，表示与课程关联的学生列表。Student模型则包含了一个courses字段，表示与学生关联的课程列表。

测试API

现在我们可以使用任何HTTP客户端测试我们的API路由了。你可以使用curl、Postman或者httpie等工具。

创建学生和课程

使用以下命令创建一个学生和一门课程：

$ curl -X POST -d '{"name":"John"}' -H "Content-Type: application/json" http://localhost:8000/students/
{"id": 1, "name": "John", "courses": []}

$ curl -X POST -d '{"name":"Math"}' -H "Content-Type: application/json" http://localhost:8000/courses/
{"id": 1, "name": "Math", "students": []}

获取所有学生和课程

使用以下命令获取所有的学生和课程：

$ curl http://localhost:8000/students/
[
    {
        "id": 1,
        "name": "John",
        "courses": []
    }
]

$ curl http://localhost:8000/courses/
[
    {
        "id": 1,
        "name": "Math",
        "students": []
    }
]

为学生添加课程

使用以下命令为学生添加一门课程：

$ curl -X POST http://localhost:8000/students/1/courses/1/
{"message": "Course added to student"}

获取学生的课程

使用以下命令获取学生的课程：

$ curl http://localhost:8000/students/1/courses/
[
    {
        "id": 1,
        "name": "Math",
        "students": []
    }
]

为课程添加学生

使用以下命令为课程添加一个学生：

$ curl -X POST http://localhost:8000/courses/1/students/1/
{"message": "Student added to course"}

获取课程的学生

使用以下命令获取课程的学生：

$ curl http://localhost:8000/courses/1/students/
[
    {
        "id": 1,
        "name": "John",
        "courses": []
    }
]

通过以上测试，我们可以看到我们成功地创建了学生和课程，并且能够通过API路由来处理多对多关系的操作。

总结

本文介绍了如何使用FastAPI、SQLAlchemy和Pydantic处理多对多关系。我们首先设计了一个适合展示多对多关系的数据库模型，然后使用FastAPI创建了API路由来处理学生和课程的创建、获取和关联操作。最后，我们使用Pydantic模型来验证和序列化输入和输出数据。通过这些步骤，我们能够轻松处理多对多关系，使我们的应用程序更加灵活和可扩展。

Camera课程

Camera API2 教程

API2 教程
Camera 性能教程

Perf 教程
Camera Native Framework 教程

Native FW 教程
Camera QCOM ISP 教程

ISP 教程
Camera QCOM Sensor 教程

Sensor 教程

Python教程

Python 教程

Python 教程
NumPy 教程

NumPy 教程
Django 教程

Django 教程

Java教程

Java 教程

Java 教程

Web教程

JavaScript 教程

JavaScript 教程
jQuery 教程

jQuery 教程
CSS 教程

CSS 教程
PHP 教程

PHP 教程
Laravel 教程

Laravel 教程
TypeScript 教程

TypeScript 教程
WordPress 教程

WordPress 教程

数据库教程

SQL 教程

SQL 教程
SQLite 教程

SQLite 教程
MySQL 教程

MySQL 教程
PL/SQL 教程

PL/SQL 教程
MongoDB 教程

MongoDB 教程

图形图像教程

OpenCV 教程

OpenCV 教程

办公软件教程

Excel 教程

Excel 教程

Linux教程

Linux命令全集

Linux命令全集
Linux内核API

Linux内核API
LVGL 教程

LVGL 教程

计算机教程

操作系统 教程

操作系统 教程
计算机网络 教程

计算机网络 教程
C语言 教程

C语言 教程
C++ 教程

C++ 教程
Swift 教程

Swift 教程

大数据教程

Hadoop 教程

Hadoop 教程
Spark 教程

Spark 教程
Scala 教程

Scala 教程

开发工具教程

Git 教程

Git 教程
Jenkins 教程

Jenkins 教程
ChatGPT 教程

ChatGPT 教程
IntelliJ 教程

IntelliJ 教程