阿雷网站建设,wordpress固定连接设置文件,phpcms仿行业网站,免费的视频app哪个好用sqlalchemy FastAPI 前端实现数据库增删改查
仅个人学习笔记#xff0c;感谢点赞关注#xff01; 知识点
连接数据库sqlalchemy 创建表结构FastAPI get post put delete操作FastAPI 请求体 路径和修改参数 依赖项 代码
# -*- ecoding: utf-8 -*-
# Author: SuperLong
# Em…sqlalchemy FastAPI 前端实现数据库增删改查
仅个人学习笔记感谢点赞关注 知识点
连接数据库sqlalchemy 创建表结构FastAPI get post put delete操作FastAPI 请求体 路径和修改参数 依赖项 代码
# -*- ecoding: utf-8 -*-
# Author: SuperLong
# Email: miu_zxl163.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:long520localhost/test,echoTrue)
class Base(DeclarativeBase):pass
class StudentClass(Base):__tablename__ StudentClassid:Mapped[str]mapped_column(Integer,primary_keyTrue)name:Mapped[str]mapped_column(String(50),nullableFalse)gender:Mapped[str]mapped_column(String(5),nullableFalse)Base.metadata.create_all(engine)
Session sessionmaker(bindengine)
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:SessionDepends(get_db)):query select(StudentClass).order_by(StudentClass.id)return db.execute(query).scalars().all()app.post(/students,response_modelStudentOut)
async def create_students(student:StudentIn,db:SessionDepends(get_db)):query select(StudentClass).where(StudentClass.name student.name)result db.execute(query).scalars().all()if result:raise HTTPException(status_code400,detailf学生 {student.name} 已存在)new_student StudentClass(idstudent.id,namestudent.name,genderstudent.gender)db.add(new_student)db.commit()return new_studentapp.put(/students/{student_id},response_modelStudentOut)
async def update_students(*,student_id:intPath(...),student:StudentBase,db:SessionDepends(get_db)):query select(StudentClass).where(StudentClass.id student_id)result db.execute(query).scalar()if not result:raise HTTPException(status_code400, detailf学生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 resultapp.delete(/students/{student_id},response_modelStudentOut)
def delete_students(student_id:intPath(...),db:SessionDepends(get_db)):query select(StudentClass).where(student_id StudentClass.id)result db.execute(query).scalar()if not result:raise HTTPException(status_codestatus.HTTP_404_NOT_FOUND, detailf学生ID {student_id} 不存在)db.delete(result)db.commit()return resultif __name__ __main__:print(os.path.split(os.path.abspath(__file__))[1])uvicorn.run(port5025,appf{os.path.split(os.path.abspath(__file__))[1].split(.)[0]}:app,reloadTrue)# # todo 增
# students [
# StudentClass(id1, name张,gender男,phone_number13463135455),
# StudentClass(id2, name张龍,gender男,phone_number13463125455),
# StudentClass(id3, name张晓同,gender男,phone_number13463145455),
# StudentClass(id4, name张晓里,gender男,phone_number13463165455),
# ]
# 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()目前专注于NLP、大模型和前后端的技术学习和分享 感谢大家的关注与支持
