文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

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

2023-09-06 16:56

关注

2 比 1 更容易理解,可以先看2(单文件级别)
在这里插入图片描述

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

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

1.0 创建mysql数据库

在这里插入图片描述

在这里插入图片描述

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

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: str    description: Union[str, None] = None# 1、创建初始 Pydantic模型/模式class ItemCreate(ItemBase):    pass# 2、创建用于读取/返回的Pydantic模型/模式class Item(ItemBase):    id: int    owner_id: int    class Config:        orm_mode = True# 1、创建初始 Pydantic模型/模式class UserBase(BaseModel):    email: str# 1、创建初始 Pydantic模型/模式class UserCreate(UserBase):    password: str# 2、创建用于读取/返回的Pydantic模型/模式class User(UserBase):    id: int    is_active: bool    items: 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 db    finally:        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 defdef差别的技术细节。

1.7 执行项目

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

在这里插入图片描述

(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 测试连接

# 导入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 = userid        self.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: int    username: str    def __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_name            session.commit()            session.close()            return {"code": "0000", "message": "修改成功"}        else:            return {"code": "0001", "message": "参数错误"}    except ArithmeticError:        return {"code": "0002", "message": "数据库错误"}# 方式2:class AlterUser(BaseModel):    userid: int    username: 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.username            session.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 = userid        self.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: int    username: str    def __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_name            session.commit()            session.close()            return {"code": "0000", "message": "修改成功"}        else:            return {"code": "0001", "message": "参数错误"}    except ArithmeticError:        return {"code": "0002", "message": "数据库错误"}# 方式2:class AlterUser(BaseModel):    userid: int    username: 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.username            session.commit()            session.close()            return {"code": "0000", "message": "修改成功"}        else:            return {"code": "0001", "message": "参数错误"}    except ArithmeticError:        return {"code": "0002", "message": "数据库错误"}

运行测试

uvicorn main:app --reload

在这里插入图片描述

来源地址:https://blog.csdn.net/weixin_46703850/article/details/128732274

阅读原文内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-数据库
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯