网站架构师招聘seo方法图片

sqlalchemy FastAPI 前端实现数据库增删改查

仅个人学习笔记，感谢点赞关注！

知识点

• 连接数据库
• sqlalchemy 创建表结构
• FastAPI get post put delete操作
• FastAPI 请求体 路径和修改参数 依赖项

代码

# -*- ecoding: utf-8 -*-
# @Author: SuperLong
# @Email: miu_zxl@163.com
# @Time: 2024/9/9 17:04
import os
import uvicorn
from fastapi import FastAPI, Depends, HTTPException, Path,status
from pydantic import BaseModel
from typing import List,Optional,Set
from sqlalchemy import create_engine, Column, Integer, String, and_, select, update
from sqlalchemy.orm import sessionmaker, Mapped, DeclarativeBase, mapped_columnengine = create_engine('mysql://root:long520@localhost/test',echo=True)
class Base(DeclarativeBase):pass
class StudentClass(Base):__tablename__ = "StudentClass"id:Mapped[str]=mapped_column(Integer,primary_key=True)name:Mapped[str]=mapped_column(String(50),nullable=False)gender:Mapped[str]=mapped_column(String(5),nullable=False)Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()class StudentBase(BaseModel):id:intname:strgender:strclass StudentIn(StudentBase):passclass StudentOut(StudentBase):passdef get_db():db = Session()try:yield dbfinally:db.close()app = FastAPI()@app.get('/students')
async def get_students(db:Session=Depends(get_db)):query = select(StudentClass).order_by(StudentClass.id)return db.execute(query).scalars().all()@app.post('/students',response_model=StudentOut)
async def create_students(student:StudentIn,db:Session=Depends(get_db)):query = select(StudentClass).where(StudentClass.name == student.name)result = db.execute(query).scalars().all()if result:raise HTTPException(status_code=400,detail=f"学生 {student.name} 已存在")new_student = StudentClass(id=student.id,name=student.name,gender=student.gender)db.add(new_student)db.commit()return new_student@app.put('/students/{student_id}',response_model=StudentOut)
async def update_students(*,student_id:int=Path(...),student:StudentBase,db:Session=Depends(get_db)):query = select(StudentClass).where(StudentClass.id == student_id)result = db.execute(query).scalar()if not result:raise HTTPException(status_code=400, detail=f"学生ID {student_id} 不存在")def update_mm(students:dict,changes:dict):for keys,values in changes.items():setattr(students,keys,values)update_mm(result,student.model_dump())db.commit()return result@app.delete('/students/{student_id}',response_model=StudentOut)
def delete_students(student_id:int=Path(...),db:Session=Depends(get_db)):query = select(StudentClass).where(student_id == StudentClass.id)result = db.execute(query).scalar()if not result:raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"学生ID {student_id} 不存在")db.delete(result)db.commit()return resultif __name__ == '__main__':print(os.path.split(os.path.abspath(__file__))[1])uvicorn.run(port=5025,app=f"{os.path.split(os.path.abspath(__file__))[1].split('.')[0]}:app",reload=True)# # todo 增
# students = [
#             StudentClass(id=1, name="张",gender="男",phone_number="13463135455"),
#             StudentClass(id=2, name="张龍",gender="男",phone_number="13463125455"),
#             StudentClass(id=3, name="张晓同",gender="男",phone_number="13463145455"),
#             StudentClass(id=4, name="张晓里",gender="男",phone_number="13463165455"),
# ]
# session.add_all(students)
# session.commit()
# # todo 查
# result = session.query(StudentClass).filter(StudentClass.gender == "男").all()
# for ii in result:
#     print("name:",ii.id)
#     print("brithday:",ii.name)# todo 改
# result = session.query(StudentClass).filter(
#     and_(
#     StudentClass.gender == "男",
#     StudentClass.name == "李楠"
#     )
# ).update(
#     {StudentClass.phone_number:"123456789"}
# )
#
# session.commit()
# todo 删
# result = session.query(StudentClass).filter(
#     and_(
#         StudentClass.gender == "男",
#         StudentClass.name == "李佳"
#     )
# ).delete()
# session.commit()