JZTXT

[FastAPI-37]ORM sqlalchemy 查询和删除

发布时间 2023-03-28 19:29:41作者: LeoShi2020
import typing

from fastapi import FastAPI, HTTPException, Depends
import pymysql
from sqlalchemy import create_engine, Column, String, Integer
from sqlalchemy.orm import sessionmaker, Session, DeclarativeBase
from urllib.parse import quote_plus as urlquote

pymysql.install_as_MySQLdb()


# 指定连接的MySQL数据库
PASSWORD = 'root@demodemo.com'
DATABASE_URL = f"mysql://root:{urlquote(PASSWORD)}@10.105.212.1:3306/db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)


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


# 创建对象的基类:
class Base(DeclarativeBase):
    pass


# 定义User对象:
class User(Base):
    __tablename__ = "users"  # 指定数据库中表的名字

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    name = Column(String(255))
    password = Column(String(255))

    def __str__(self):
        return f"id: {self.id}, name: {self.name}, password: {self.password}"


app = FastAPI(title="FastAPI + SqlAlchemy")


@app.get("/users")
def get_users(page: int = 1, size: int = 3, db: Session = Depends(get_db)):
    users: typing.List[User] = db.query(User).all()[(page - 1) * size:page * size]
    return [{"id": u.id, "name": u.name} for u in users]


@app.get("/user/{user_id}")
def get_user_by_id(user_id: int, db: Session = Depends(get_db)):
    user: User = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(detail=f"Not found user with id: {user_id}", status_code=404)
    return {"id": user.id, "name": user.name}


@app.delete("/user/{user_id}")
def delete_user_by_id(user_id: int, db: Session = Depends(get_db)):
    db.query(User).filter(User.id == user_id).delete()
    return {"code": 1, "msg": "success"}