[FastAPI-38]ORM SQLAlchemy 新建和更新-526互联

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
from pydantic import BaseModel as SchemaBaseModel

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")


class UserBase(SchemaBaseModel):
    name: str


class UserIn(UserBase):
    password: str


class UserOut(UserBase):
    id: int

    class Config:
        orm_mode = True     # 设置后 response_model=UserOut时，才能直接返回 db_user


@app.post("/user", response_model=UserOut)
def create_user(user: UserIn, db: Session = Depends(get_db)):
    # db_user = User(name=user.name, password=user.password)
    db_user = User(**user.dict())       # 和上面的用法是等价的
    db.add(db_user)
    db.commit()
    db.refresh(db_user)     # refresh之后，db_user才有数据
    return db_user


@app.put("/user/{user_id}", response_model=UserOut)
def update_user_by_id(user_id: int, user: UserIn, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if not db_user:
        raise HTTPException(detail=f"Not found user with id: {user_id}", status_code=404)
    db_user.name = user.name
    db_user.password = user.password
    db.commit()
    db.refresh(db_user)
    return db_user