FastAPI - 数据库操作5

先安装mysql驱动程序

pipenv install pymysql

安装数据库ORM库SQLAlchemy

pipenv install SQLAlchemy

修改文件main.py文件内容

设置数据库连接

# -*- coding:utf-8 –*-
from fastapi import FastAPIfrom sqlalchemy import create_engineHOST = '192.168.123.228'
PORT = 3306
USERNAME = 'root'
PASSWORD = '123456'
DATABASE = 'fenxiangyin'db_url = f"mysql+pymysql://{USERNAME}:{PASSWORD}@{HOST}:{PORT}/{DATABASE}"
engine = create_engine(db_url)

初始化会话工厂

from sqlalchemy.orm import sessionmaker
engine = create_engine()
SessionFactory = sessionmaker(autocommit=False, autoflush=False, bind=engine)
session = SessionFactory()

定义Base类：映射到数据表的ORM类

from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

构建模型层

from sqlalchemy import Boolean, Column, ForeignKey, Integer, Stringclass User(Base):__tablename__:str = "user"id = Column(Integer, primary_key=True)username = Column('username',String, unique=True, index=True)password = Column('password',String)

执行增删改查操作

app = FastAPI()#新增
@app.get("/user/add")
async def user_add():user1 = User(username="vvv", password="123")session.add(user1)session.commit()return user1.id#修改
@app.get("/user/update")
async def user_update():user1 = session.query(User).filter(User.id == 1).first()user1.password = "456"session.commit()return True#删除
@app.get("/user/delete")
async def user_find():user1 = session.query(User).filter(User.id == 1).first()session.delete(user1)session.commit()return True

完整代码

# -*- coding:utf-8 –*-
from fastapi import FastAPIfrom sqlalchemy import create_engineHOST = '192.168.123.228'
PORT = 3306
USERNAME = 'root'
PASSWORD = '123456'
DATABASE = 'fenxiangyin'db_url = f"mysql+pymysql://{USERNAME}:{PASSWORD}@{HOST}:{PORT}/{DATABASE}"from sqlalchemy.orm import sessionmakerengine = create_engine(db_url)
SessionFactory = sessionmaker(autocommit=False, autoflush=False, bind=engine)
session = SessionFactory()from sqlalchemy.ext.declarative import declarative_baseBase = declarative_base()from sqlalchemy import Boolean, Column, ForeignKey, Integer, Stringclass User(Base):__tablename__:str = "user"id = Column(Integer, primary_key=True)username = Column('username',String, unique=True, index=True)password = Column('password',String)app = FastAPI()#新增
@app.get("/user/add")
async def user_add():user1 = User(username="vvv", password="123")session.add(user1)session.commit()return user1.id#修改
@app.get("/user/update")
async def user_update():user1 = session.query(User).filter(User.id == 1).first()user1.password = "456"session.commit()return True#删除
@app.get("/user/delete")
async def user_find():user1 = session.query(User).filter(User.id == 1).first()session.delete(user1)session.commit()return True