Python FastAPI 框架操作Mysql数据库增删改查

2 比 1 更容易理解，可以先看2（单文件级别）

1、FastAPI 框架操作Mysql数据库(项目多文件级别)

FastAPI 可以使用任何您想要的关系型数据库。
在这里，让我们看一个使用着SQLAlchemy的示例。
您可以很容易地将SQLAlchemy支持任何数据库，像：

PostgreSQL
MySQL
SQLite
Oracle
Microsoft SQL Server，等等其它数据库
在此示例中，我们将使用SQLite，因为它使用单个文件并且在Python中具有集成支持。因此，您可以复制此示例并按原样来运行它。

稍后，对于您的产品级别的应用程序，您可能会要使用像PostgreSQL这样的数据库服务器。

1.0 创建mysql数据库

创建test数据库，数据库创建users表和items表
users表

items表

1.1 测试项目文件结构

对于这些示例，假设您有一个名为的目录my_super_project，其中包含一个名为的子目录sql_app，其结构如下：

.└── sql_app├── __init__.py├── crud.py├── database.py├── main.py├── models.py└── schemas.py

该文件__init__.py只是一个空文件，但它告诉 Python 其中sql_app的所有模块（Python 文件）都是一个包。

1.2 数据库配置 database.py

# 1、导入 SQLAlchemy 部件from sqlalchemy import create_enginefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import sessionmaker# 连接mysql数据库需要导入pymysql模块import pymysqlpymysql.install_as_MySQLdb()# 2、为 SQLAlchemy 定义数据库 URL地址# 配置数据库地址：数据库类型+数据库驱动名称://用户名:密码@机器地址:端口号/数据库名SQLALCHEMY_DATABASE_URL = "mysql://test:123456@127.0.0.1:3306/test"# 3、创建 SQLAlchemy 引擎engine = create_engine(SQLALCHEMY_DATABASE_URL, encoding='utf-8')# 4、创建数据库会话SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)# 5、创建一个Base类declarative_base# 稍后我们将用这个类继承，来创建每个数据库模型或类（ORM 模型）Base = declarative_base()

1.3 创建数据库模型 models.py

用Base类来创建 SQLAlchemy 模型
我们将使用我们之前创建的Base类来创建 SQLAlchemy 模型。

SQLAlchemy 使用的“模型”这个术语来指代与数据库交互的这些类和实例。
而 Pydantic 也使用“模型”这个术语来指代不同的东西，即数据验证、转换以及文档类和实例。

from sqlalchemy import Boolean, Column, ForeignKey, Integer, Stringfrom sqlalchemy.orm import relationship# 1、从database.py导入Base类from .database import Base# User继承Base类class User(Base):# 表名__tablename__ = "users"# 2、创建模型属性/列，使用Column来表示 SQLAlchemy 中的默认值。id = Column(Integer, primary_key=True, index=True)email = Column(String, unique=True, index=True)hashed_password = Column(String)is_active = Column(Boolean, default=True)# 3、创建关系# 当访问 user 中的属性items时，如 中my_user.items，它将有一个ItemSQLAlchemy 模型列表（来自items表），这些模型具有指向users表中此记录的外键# 当您访问my_user.items时，SQLAlchemy 实际上会从items表中的获取一批记录并在此处填充进去。# 同样，当访问 Item中的属性owner时，它将包含表中的UserSQLAlchemy 模型users。使用owner_id属性/列及其外键来了解要从users表中获取哪条记录。items = relationship("Item", back_populates="owner")# Item继承Base类class Item(Base):__tablename__ = "items"id = Column(Integer, primary_key=True, index=True)title = Column(String, index=True)description = Column(String, index=True)owner_id = Column(Integer, ForeignKey("users.id"))owner = relationship("User", back_populates="items")

创建关系

items = relationship("Item", back_populates="owner")

当访问 user 中的属性items时，如中my_user.items，它将有一个ItemSQLAlchemy 模型列表（来自items表），这些模型具有指向users表中此记录的外键
当您访问my_user.items时，SQLAlchemy 实际上会从items表中的获取一批记录并在此处填充进去。
同样，当访问 Item中的属性owner时，它将包含表中的-UserSQLAlchemy 模型users。使用owner_id属性/列及其外键来了解要从users表中获取哪条记录。

1.4 创建 Pydantic 模型 schemas.py

现在让我们查看一下文件sql_app/schemas.py。

为了避免 SQLAlchemy模型和 Pydantic模型之间的混淆，我们将有models.py（SQLAlchemy 模型的文件）和schemas.py（ Pydantic 模型的文件）。
这些 Pydantic 模型或多或少地定义了一个“schema”（一个有效的数据形状）。
因此，这将帮助我们在使用两者时避免混淆。
创建初始 Pydantic模型/模式¶
创建一个ItemBase和UserBasePydantic模型（或者我们说“schema”）以及在创建或读取数据时具有共同的属性。

ItemCreate为创建一个UserCreate继承自它们的所有属性（因此它们将具有相同的属性），以及创建所需的任何其他数据（属性）。

因此在创建时也应当有一个password属性。

但是为了安全起见，password不会出现在其他同类 Pydantic模型中，例如用户请求时不应该从 API 返回响应中包含它。

from typing import List, Union# 1、创建初始 Pydantic模型/模式from pydantic import BaseModel# 1、创建初始 Pydantic模型/模式class ItemBase(BaseModel):title: strdescription: Union[str, None] = None# 1、创建初始 Pydantic模型/模式class ItemCreate(ItemBase):pass# 2、创建用于读取/返回的Pydantic模型/模式class Item(ItemBase):id: intowner_id: intclass Config:orm_mode = True# 1、创建初始 Pydantic模型/模式class UserBase(BaseModel):email: str# 1、创建初始 Pydantic模型/模式class UserCreate(UserBase):password: str# 2、创建用于读取/返回的Pydantic模型/模式class User(UserBase):id: intis_active: boolitems: List[Item] = []class Config:orm_mode = True

请注意，读取用户（从 API 返回）时将使用不包括password的User Pydantic模型。

SQLAlchemy 风格和 Pydantic 风格

请注意，SQLAlchemy模型使用 =来定义属性，并将类型作为参数传递给Column，例如：

name = Column(String)

虽然 Pydantic模型使用: 声明类型，但新的类型注释语法/类型提示是：

name: str

请牢记这一点，这样您在使用:还是=时就不会感到困惑。

1.5 CRUD工具 crud.py

从 sqlalchemy.orm中导入Session，这将允许您声明db参数的类型，并在您的函数中进行更好的类型检查和完成。

导入之前的models（SQLAlchemy 模型）和schemas（Pydantic模型/模式）。

from sqlalchemy.orm import Sessionfrom . import models, schemasdef get_user(db: Session, user_id: int):return db.query(models.User).filter(models.User.id == user_id).first()# 通过 ID 和电子邮件查询单个用户def get_user_by_email(db: Session, email: str):return db.query(models.User).filter(models.User.email == email).first()# 查询多个用户def get_users(db: Session, skip: int = 0, limit: int = 100):return db.query(models.User).offset(skip).limit(limit).all()def create_user(db: Session, user: schemas.UserCreate):fake_hashed_password = user.password + "notreallyhashed"# 使用您的数据创建一个 SQLAlchemy 模型实例。db_user = models.User(email=user.email, hashed_password=fake_hashed_password)# 使用add来将该实例对象添加到您的数据库。db.add(db_user)# 使用commit来对数据库的事务提交（以便保存它们）。db.commit()# 使用refresh来刷新您的数据库实例（以便它包含来自数据库的任何新数据，例如生成的 ID）。db.refresh(db_user)return db_user# 查询多个项目def get_items(db: Session, skip: int = 0, limit: int = 100):return db.query(models.Item).offset(skip).limit(limit).all()def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):db_item = models.Item(**item.dict(), owner_id=user_id)db.add(db_item)db.commit()db.refresh(db_item)return db_item

1.6 主FastAPI应用程序 main.py

from typing import Listfrom fastapi import Depends, FastAPI, HTTPExceptionfrom sqlalchemy.orm import Sessionfrom . import crud, models, schemasfrom .database import SessionLocal, engine# 1、创建数据库表models.Base.metadata.create_all(bind=engine)app = FastAPI()# 2、创建依赖项# Dependencydef get_db():# 我们需要每个请求有一个独立的数据库会话/连接（SessionLocal），# 在所有请求中使用相同的会话，然后在请求完成后关闭它。db = SessionLocal()# 我们的依赖项将创建一个新的 SQLAlchemy SessionLocal，# 它将在单个请求中使用，然后在请求完成后关闭它。try:yield dbfinally:db.close()# 4、创建您的FastAPI 路径操作# 3、db: Session = Depends(get_db) ：当在路径操作函数中使用依赖项时，我们使用Session，直接从 SQLAlchemy 导入的类型声明它。@app.post("/users/", response_model=schemas.User)def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):db_user = crud.get_user_by_email(db, email=user.email)if db_user:raise HTTPException(status_code=400, detail="Email already registered")return crud.create_user(db=db, user=user)# 4、创建您的FastAPI 路径操作@app.get("/users/", response_model=List[schemas.User])def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):users = crud.get_users(db, skip=skip, limit=limit)return users# 4、创建您的FastAPI 路径操作@app.get("/users/{user_id}", response_model=schemas.User)def read_user(user_id: int, db: Session = Depends(get_db)):db_user = crud.get_user(db, user_id=user_id)if db_user is None:raise HTTPException(status_code=404, detail="User not found")return db_user# 4、创建您的FastAPI 路径操作@app.post("/users/{user_id}/items/", response_model=schemas.Item)def create_item_for_user(user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)):return crud.create_user_item(db=db, item=item, user_id=user_id)# 4、创建您的FastAPI 路径操作@app.get("/items/", response_model=List[schemas.Item])def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):items = crud.get_items(db, skip=skip, limit=limit)return items

1.6 关于 `def` 对比 `async def`

在这里，我们在路径操作函数和依赖项中都使用着 SQLAlchemy 模型，它将与外部数据库进行通信。

这会需要一些“等待时间”。

但是由于 SQLAlchemy 不具有await直接使用的兼容性，因此类似于：

user = await db.query(User).first()

…相反，我们可以使用：

user = db.query(User).first()

然后我们应该声明路径操作函数和不带的依赖关系async def，只需使用普通的def，如下：

@app.get("/users/{user_id}", response_model=schemas.User)def read_user(user_id: int, db: Session = Depends(get_db)):db_user = crud.get_user(db, user_id=user_id)...

如果您需要异步连接到关系数据库，请参阅Async SQL (Relational) Databases
如果您很好奇并且拥有深厚的技术知识，您可以在Async文档中查看有关如何处理 async def于def差别的技术细节。

1.7 执行项目

uvicorn sql_app.main:app --reloadINFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

2、FastAPI 对MySQL 数据库的操作（单文件级别）

(317条消息) FastAPI 对MySQL 数据库的操作(win64)_fastapi 连接mysql_梧钴的博客-CSDN博客

2.1 安装依赖

$ pip install sqlalchemy -i https://pypi.tuna.tsinghua.edu.cn/simple$ pip install pymysql -i https://pypi.tuna.tsinghua.edu.cn/simple

2.2 测试连接

main.py

# 导入FastAPI模块from fastapi import FastAPIapp = FastAPI()# 1、连接mysql数据库需要导入pymysql模块import pymysqlpymysql.install_as_MySQLdb()# 2、配置数据库from sqlalchemy import create_enginefrom sqlalchemy.orm import sessionmaker# 配置数据库地址：数据库类型+数据库驱动名称://用户名:密码@机器地址:端口号/数据库名SQLALCHEMY_DATABASE_URL = "mysql://test:123456@127.0.0.1:3306/test"engine = create_engine(SQLALCHEMY_DATABASE_URL, encoding='utf-8')# 3、把当前的引擎绑定给这个会话；# autocommit：是否自动提交 autoflush：是否自动刷新并加载数据库 bind：绑定数据库引擎Session = sessionmaker(autocommit=False, autoflush=False, bind=engine)# 实例化session = Session()# 4、创建Base实例# declarative_base类维持了一个从类到表的关系，通常一个应用使用一个Base实例，所有实体类都应该继承此类对象from sqlalchemy.ext.declarative import declarative_baseBase = declarative_base()# 5、创建数据库模型（定义表结构:表名称，字段名称以及字段类型）from sqlalchemy import Column, String, Integerclass User(Base):# 定义表名__tablename__ = 'user'# 定义字段# primary_key=True 设置为主键userid = Column(Integer, primary_key=True)username = Column(String(255))# 构造函数def __init__(self, userid, username):self.userid = useridself.username = username# 打印形式def __str__(self):return "id:%s, name:%s" % (str(self.userid), self.username)# 6、在数据库中生成表Base.metadata.create_all(bind=engine)

运行

$ uvicorn main:app --reload

2.3 增删改查代码

增

from pydantic import BaseModel# 定义数据模型class CreatUser(BaseModel):userid: intusername: strdef __str__(self):return "id：%s, name：%s" % (str(self.userid), self.username)# 添加单个数据@app.post("/user/addUser")async def InserUser(user: CreatUser):try:# 添加数据dataUser = User(userid=user.userid, username=user.username)session.add(dataUser)session.commit()session.close()except ArithmeticError:return {"code": "0002", "message": "数据库异常"}return {"code": "0000", "message": "添加成功"}from typing import List# 添加多个数据@app.post("/user/addUserList")async def addUserList(*, user: List[CreatUser]):try:# user是一个列表，每个内部元素均为CreatUser类型for u in user:# 自定义的数据模型可以用.访问属性dataUser = User(userid=u.userid, username=u.username)session.add(dataUser)session.commit()session.close()except ArithmeticError:return {"code": "0002", "message": "数据库异常"}return {"code": "0000", "message": "添加成功"}

查

# 按照user_id查询@app.get("/user/{user_id}")async def queryUserByUserId(user_id: int):# 创建Query查询，filter是where条件，调用one返回唯一行，调用all则是返回所有行try:# one与first的区别：# one：要求结果集中只有一个结果；如果数据库返回0或2个或更多结果，并且将引发异常，则为错误。# first：返回可能更大的结果集中的第一个，如果没有结果，则返回None。不会引发异常。# filter_by与filter的区别：# filter_by接收的参数形式是关键字参数，而filter接收的参数是更加灵活的SQL表达式结构# user1 = session.query(User).filter_by(userid=user_id).first()user1 = session.query(User).filter(User.userid == user_id).first()session.close()# 由于user1只有一个值，所以它直接是一个字典if user1:return {"code": "0000", "message": "请求成功", "data": user1}else:return {"code": "0001", "message": "查询无结果"}except ArithmeticError:return {"code": "0002", "message": "数据库异常"}## 查询所有@app.get("/user/selectall/")async def queryUserByUserId():# 创建Query查询，filter是where条件，调用one返回唯一行，调用all则是返回所有行try:user1 = session.query(User).all()session.close()# user1 是一个列表，内部元素为字典return {"code": "0000", "message": "请求成功", "data": user1}except ArithmeticError:return {"code": "0002", "message": "数据库异常"}

删

# 根据user_id删除单个@app.delete("/user/deleteUser/{user_id}")async def deleteUser(user_id: int):try:user1 = session.query(User).filter(User.userid == user_id).first()if user1:session.delete(user1)session.commit()session.close()return {"code": "0000", "message": "删除成功"}else:return {"code": "0001", "message": "参数错误"}except ArithmeticError:return {"code": "0002", "message": "数据库错误"}from typing import List# 删除多个@app.delete("/user/deleteUserList")async def deleteUser(user_ids: List[int]):try:for user_id in user_ids:user1 = session.query(User).filter(User.userid == user_id).first()if user1:session.delete(user1)session.commit()session.close()return {"code": "0000", "message": "删除成功"}except ArithmeticError:return {"code": "0002", "message": "数据库错误"}

改

# 根据user_id修改user_name@app.put("/user/updateUser/")# 定义查询参数user_id和user_nameasync def updateUser(user_id: int, user_name: str):try:user1 = session.query(User).filter(User.userid == user_id).first()print(user1)if user1:user1.username = user_namesession.commit()session.close()return {"code": "0000", "message": "修改成功"}else:return {"code": "0001", "message": "参数错误"}except ArithmeticError:return {"code": "0002", "message": "数据库错误"}# 方式2：class AlterUser(BaseModel):userid: intusername: str@app.put("/user/updateUser01/")async def deleteUser(user: AlterUser):try:user1 = session.query(User).filter(User.userid == user.userid).first()if user1:user1.username = user.usernamesession.commit()session.close()return {"code": "0000", "message": "修改成功"}else:return {"code": "0001", "message": "参数错误"}except ArithmeticError:return {"code": "0002", "message": "数据库错误"}

完整代码

# 导入FastAPI模块from fastapi import FastAPIapp = FastAPI()# 1、连接mysql数据库需要导入pymysql模块import pymysqlpymysql.install_as_MySQLdb()# 2、配置数据库from sqlalchemy import create_enginefrom sqlalchemy.orm import sessionmaker# 配置数据库地址：数据库类型+数据库驱动名称://用户名:密码@机器地址:端口号/数据库名SQLALCHEMY_DATABASE_URL = "mysql://test:123456@127.0.0.1:3306/test"engine = create_engine(SQLALCHEMY_DATABASE_URL, encoding='utf-8')# 3、把当前的引擎绑定给这个会话；# autocommit：是否自动提交 autoflush：是否自动刷新并加载数据库 bind：绑定数据库引擎Session = sessionmaker(autocommit=False, autoflush=False, bind=engine)# 实例化session = Session()# 4、创建Base实例# declarative_base类维持了一个从类到表的关系，通常一个应用使用一个Base实例，所有实体类都应该继承此类对象from sqlalchemy.ext.declarative import declarative_baseBase = declarative_base()# 5、创建数据库模型（定义表结构:表名称，字段名称以及字段类型）from sqlalchemy import Column, String, Integerclass User(Base):# 定义表名__tablename__ = 'user'# 定义字段# primary_key=True 设置为主键userid = Column(Integer, primary_key=True)username = Column(String(255))# 构造函数def __init__(self, userid, username):self.userid = useridself.username = username# 打印形式def __str__(self):return "id:%s, name:%s" % (str(self.userid), self.username)# 6、在数据库中生成表Base.metadata.create_all(bind=engine)# 【增】from pydantic import BaseModel# 定义数据模型class CreatUser(BaseModel):userid: intusername: strdef __str__(self):return "id：%s, name：%s" % (str(self.userid), self.username)# 添加单个@app.post("/user/addUser")async def InserUser(user: CreatUser):try:# 添加数据dataUser = User(userid=user.userid, username=user.username)session.add(dataUser)session.commit()session.close()except ArithmeticError:return {"code": "0002", "message": "数据库异常"}return {"code": "0000", "message": "添加成功"}from typing import List# 添加多个@app.post("/user/addUserList")async def addUserList(*, user: List[CreatUser]):try:# user是一个列表，每个内部元素均为CreatUser类型for u in user:# 自定义的数据模型可以用.访问属性dataUser = User(userid=u.userid, username=u.username)session.add(dataUser)session.commit()session.close()except ArithmeticError:return {"code": "0002", "message": "数据库异常"}return {"code": "0000", "message": "添加成功"}# 【查】# 按照user_id查询@app.get("/user/{user_id}")async def queryUserByUserId(user_id: int):# 创建Query查询，filter是where条件，调用one返回唯一行，调用all则是返回所有行try:# one与first的区别：# one：要求结果集中只有一个结果；如果数据库返回0或2个或更多结果，并且将引发异常，则为错误。# first：返回可能更大的结果集中的第一个，如果没有结果，则返回None。不会引发异常。# filter_by与filter的区别：# filter_by接收的参数形式是关键字参数，而filter接收的参数是更加灵活的SQL表达式结构# user1 = session.query(User).filter_by(userid=user_id).first()user1 = session.query(User).filter(User.userid == user_id).first()session.close()# 由于user1只有一个值，所以它直接是一个字典if user1:return {"code": "0000", "message": "请求成功", "data": user1}else:return {"code": "0001", "message": "查询无结果"}except ArithmeticError:return {"code": "0002", "message": "数据库异常"}## 查询所有@app.get("/user/selectall/")async def queryUserByUserId():# 创建Query查询，filter是where条件，调用one返回唯一行，调用all则是返回所有行try:user1 = session.query(User).all()session.close()# user1 是一个列表，内部元素为字典return {"code": "0000", "message": "请求成功", "data": user1}except ArithmeticError:return {"code": "0002", "message": "数据库异常"}#【删】# 根据user_id删除单个@app.delete("/user/deleteUser/{user_id}")async def deleteUser(user_id: int):try:user1 = session.query(User).filter(User.userid == user_id).first()if user1:session.delete(user1)session.commit()session.close()return {"code": "0000", "message": "删除成功"}else:return {"code": "0001", "message": "参数错误"}except ArithmeticError:return {"code": "0002", "message": "数据库错误"}from typing import List## 删除多个@app.delete("/user/deleteUserList")async def deleteUser(user_ids: List[int]):try:for user_id in user_ids:user1 = session.query(User).filter(User.userid == user_id).first()if user1:session.delete(user1)session.commit()session.close()return {"code": "0000", "message": "删除成功"}except ArithmeticError:return {"code": "0002", "message": "数据库错误"}# 【改】## 根据user_id修改user_name@app.put("/user/updateUser/")# 定义查询参数user_id和user_nameasync def updateUser(user_id: int, user_name: str):try:user1 = session.query(User).filter(User.userid == user_id).first()print(user1)if user1:user1.username = user_namesession.commit()session.close()return {"code": "0000", "message": "修改成功"}else:return {"code": "0001", "message": "参数错误"}except ArithmeticError:return {"code": "0002", "message": "数据库错误"}# 方式2：class AlterUser(BaseModel):userid: intusername: str@app.put("/user/updateUser01/")async def deleteUser(user: AlterUser):try:user1 = session.query(User).filter(User.userid == user.userid).first()if user1:user1.username = user.usernamesession.commit()session.close()return {"code": "0000", "message": "修改成功"}else:return {"code": "0001", "message": "参数错误"}except ArithmeticError:return {"code": "0002", "message": "数据库错误"}

运行测试

uvicorn main:app --reload

Python FastAPI 框架操作Mysql数据库增删改查

1、FastAPI 框架操作Mysql数据库(项目多文件级别)

1.0 创建mysql数据库

1.1 测试项目文件结构

1.2 数据库配置 database.py

1.3 创建数据库模型 models.py

1.4 创建 Pydantic 模型 schemas.py

SQLAlchemy 风格和 Pydantic 风格

1.5 CRUD工具 crud.py

1.6 主FastAPI应用程序 main.py

1.6 关于 `def` 对比 `async def`

1.7 执行项目

2、FastAPI 对MySQL 数据库的操作（单文件级别）

2.1 安装依赖

2.2 测试连接

2.3 增删改查代码

增

查

删

改

完整代码

运行测试

最新关注

热文推荐

mac pro m1：安装dump文件内存分析工具——MAT

asp中文数字验证码

maven 的安装与配置（Command ‘mvn‘ not found）修改配置文件后新终端依旧无法识别到 mvn 命令

报错:SON parse error: Cannot deserialize value of type `java.lang.String` from Array value (token `Jso

sklearn 逻辑回归Demo

Python代码报错：Import “requests“ could not be resolved from sourcePylance（reportMissingModuleSource）解决

Python FastAPI 框架 操作Mysql数据库 增删改查

1、FastAPI 框架 操作Mysql数据库(项目多文件级别)

1.0 创建mysql数据库

1.1 测试项目文件结构

1.2 数据库配置 database.py

1.3 创建数据库模型 models.py

1.4 创建 Pydantic 模型 schemas.py

SQLAlchemy 风格和 Pydantic 风格

1.5 CRUD工具 crud.py

1.6 主FastAPI应用程序 main.py

1.6 关于 def 对比 async def

1.7 执行项目

2、FastAPI 对MySQL 数据库的操作（单文件级别）

2.1 安装依赖

2.2 测试连接

2.3 增删改查代码

增

查

删

改

完整代码

运行测试

相关文章

最新关注

热文推荐

Python FastAPI 框架操作Mysql数据库增删改查

1、FastAPI 框架操作Mysql数据库(项目多文件级别)

1.6 关于 `def` 对比 `async def`