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

Posted 2023-03-31 一个平凡de人

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了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_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 连接mysql数据库需要导入pymysql模块
import pymysql
pymysql.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, String
from 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: 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 Session

from . import models, schemas


def 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 List

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

# 1、创建数据库表
models.Base.metadata.create_all(bind=engine)

app = FastAPI()


# 2、创建依赖项
# Dependency
def 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 def于def差别的技术细节。

1.7 执行项目

uvicorn sql_app.main:app --reload
INFO:     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 FastAPI
app = FastAPI()

# 1、连接mysql数据库需要导入pymysql模块
import pymysql
pymysql.install_as_MySQLdb()

# 2、配置数据库
from sqlalchemy import create_engine
from 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_base
Base = declarative_base()


# 5、创建数据库模型（定义表结构:表名称，字段名称以及字段类型）
from sqlalchemy import Column, String, Integer

class 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()API接口开发其实特简单，Python FastApi Web 框架教程来了



 文章目录
 技术交流、源码分享
一、FastApi？？
1.FastAPI可以做什么
2.为什么要学习 FastAPI
   
二、准备工作
1.编译器工具
2.python安装教程
3.pycharm安装教程
4.安装虚拟环境
1、创建项目工程
2、安装环境
3、安装fastapi
   
   
三、教程
1.开启服务和接口访问
2.json数据
3.api文档在线生成
4.发送请求
5.获取URL参数
6.获取请求头参数
7.表单获取数据
8.自定义返回JSON信息
8.自定义返回HTML
9.自定义返回文件
10.自定义返回HTML页面
11.代办事项小案例
12.绑定数据库
13.数据库访问
14.数据库写入
  
 

 
技术交流、源码分享 
目前群友已超过3000人，添加时最好的备注方式为：来源+兴趣方向，技术交流、核心代码分享都搞定。 
 
 方式①、添加微信号：dkl88191，备注：来自CSDN+FastApi
 方式②、微信搜索公众号：Python学习与数据挖掘，后台回复：FastApi 
 
一、FastApi？？ 
1.FastAPI可以做什么 
它由 Sebastian Ramirez 开发 
开发网站
开发 web API
做一个测试平台
做一个持续集成工具
生成文档 
2.为什么要学习 FastAPI 
FastAPI的使用方法和设计参考了几十个 python 明星项目 博采众长 吸收各家精华
FastAPI和Flask一样简约
FastAPI非常新 并且基于python3.6的类型声明系统
自动生成文档
类型声明和数据检验
内置身份认证
性能强
原生支持异步
强大的依赖注入系统
GraphQL和WebSocket支持 
二、准备工作 
1.编译器工具 
Python、Pycharm 
2.python安装教程 
https://www.runoob.com/python/python-install.html
 
3.pycharm安装教程 
runoob.com/w3cnote/pycharm-windows-install.html
 
4.安装虚拟环境 
1、创建项目工程 
 
2、安装环境 
 
3、安装fastapi 
 
三、教程 
1.开启服务和接口访问 
main.py 
import uvicorn
from fastapi import FastAPI

app=FastAPI()

if __name__ == '__main__':
    uvicorn.run(app)
 
 
服务器运行 
 
 
添加接口 
 
main 
import uvicorn
from fastapi import FastAPI

app=FastAPI()

# 添加首页
@app.get("/")
def index():
    return "This is Home Page."

if __name__ == '__main__':
    uvicorn.run(app)
 
重新运行项目
  
 
2.json数据 
 
重新项目 
 
json数据格式解析 
https://www.json.cn/
 
 
3.api文档在线生成 

 文档解析
 
 文档备注信息
  
 
4.发送请求 
POST请求
  
使用ApiPost接口测试工具来访问接口 
 
 
定义多种请求格式 
 
5.获取URL参数 

  
main.py 
import uvicorn
from fastapi import FastAPI

app=FastAPI()

@app.get("/user/id")
def user(id):
    return "id":id

if __name__ == '__main__':
    uvicorn.run(app)

 
6.获取请求头参数 
 
 
main.py 
import uvicorn
from fastapi import FastAPI,Header

app=FastAPI()

@app.get("/user")
def user(id, token=Header(None)):
    return "id":id,"token":token

if __name__ == '__main__':
    uvicorn.run(app)
 
7.表单获取数据 
安装包 
python-multipart
 

  
main.py 
import uvicorn
from fastapi import FastAPI,Header,Form

app=FastAPI()

@app.get("/user")
def user(id, token=Header(None)):
    return "id":id,"token":token

@app.post("/login")
def login(username=Form(None),password=Form(None)):
    return "data":"username":username,"password":password

if __name__ == '__main__':
    uvicorn.run(app)
 
8.自定义返回JSON信息 
 
main.py 
import uvicorn
from fastapi import FastAPI
from fastapi.responses import JSONResponse

app=FastAPI()

@app.get("/user")
def user():
    return JSONResponse(content="msg":"get user",
                        status_code=202,
                        headers="a":"b")

if __name__ == '__main__':
    uvicorn.run(app)

 

  
8.自定义返回HTML 

  
main.py 
import uvicorn
from fastapi import FastAPI
from fastapi.responses import JSONResponse,HTMLResponse

app=FastAPI()

@app.get("/user")
def user():
    return JSONResponse(content="msg":"get user",
                        status_code=202,
                        headers="a":"b")

@app.get("/")
def user():
    html_content="""
    <html>
        <body><p style="color:red">Hello World</p></body>
        </html>
    """
    return HTMLResponse(content=html_content)

if __name__ == '__main__':
    uvicorn.run(app)
 
9.自定义返回文件 

  
main.py 
import uvicorn
from fastapi import FastAPI
from fastapi.responses import JSONResponse,HTMLResponse
from starlette.responses import FileResponse

app=FastAPI()

@app.get("/user")
def user():
    return JSONResponse(content="msg":"get user",
                        status_code=202,
                        headers="a":"b")

@app.get("/")
def user():
    html_content="""
    <html>
        <body><p style="color:red">Hello World</p></body>
        </html>
    """
    return HTMLResponse(content=html_content)

@app.get("/avatar")
def user():
    avatar='./static/violet.jpg'
    return FileResponse(avatar)

if __name__ == '__main__':
    uvicorn.run(app)
 
10.自定义返回HTML页面 

 
  
main.py 
import uvicorn
from fastapi import FastAPI,Request
from fastapi.templating import Jinja2Templates

app=FastAPI()
template=Jinja2Templates("pages")

@app.get("/")
def user(req:Request):
    return template.TemplateResponse("index.html",context="request":req)

if __name__ == '__main__':
    uvicorn.run(app)
 
返回结果 

 
 
  
11.代办事项小案例 
main.py 
 
index.html 
 
运行项目 
 
main.py 
 
index.html 
 
运行项目 
 
12.绑定数据库 
安装 tortoise-orm 
 
安装 aiomysql 
 
main.py
  
13.数据库访问 
 
models.py 
from tortoise import Model,fields

class Todo(Model):
    id=fields.IntField(pk=True)
    content=fields.CharField(max_length=500)
    create_at=fields.DatetimeField(auto_now_add=True)
    updated_at=fields.DatetimeField(auto_now=True)
 
 
main.py 
import uvicorn
from fastapi import FastAPI, Request, Form
from fastapi.responses import RedirectResponse
from starlette.templating import Jinja2Templates
from tortoise.contrib.fastapi import register_tortoise

from dao.models import Todo

app=FastAPI()
template=Jinja2Templates("pages")

# 数据库绑定
register_tortoise(app,db_url="mysql://root:123456@localhost:3306/fastapi",
                  modules="models":['dao.models'],
                  add_exception_handlers=True,
                  generate_schemas=True)

todos = ["写日记", "看电影", "玩游戏"]

# 添加首页
@app.get("/")
async def user(req:Request):
    todos=await Todo.all()
    print(todos)
    return template.TemplateResponse("index.html",context="request":req,"todos":todos)

@app.post("/todo")
def todo(todo=Form(None)):
    todos.insert(0,todo)
    return RedirectResponse("/",status_code=302)

if __name__ == '__main__':
    uvicorn.run(app)
 
运行项目
  
14.数据库写入 
main.py 
 
运行项目 
 
数据库里就多了个字段 
 
此时就是显示有问题 
 
index.html 

 
以上是关于Python FastAPI 框架 操作Mysql数据库 增删改查的主要内容，如果未能解决你的问题，请参考以下文章 
 API接口开发其实特简单，Python FastApi Web 框架教程来了
 API接口开发其实特简单，Python FastApi Web 框架教程来了
 python fastapi 返回echart
 python框架fastapi, AttributeError: module 'asyncio' has no attribute 'iscoroutinefunction&
 FastAPI - 一款新型的 Python Web 框架(对比 Flask)
 FastAPI 1：安装FastAPI

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 增删改查代码

增

查

删

API接口开发其实特简单，Python FastApi Web 框架教程来了

文章目录

技术交流、源码分享

一、FastApi？？

1.FastAPI可以做什么

2.为什么要学习 FastAPI

二、准备工作

1.编译器工具

2.python安装教程

3.pycharm安装教程

4.安装虚拟环境

1、创建项目工程

2、安装环境

3、安装fastapi

三、教程

1.开启服务和接口访问

2.json数据

3.api文档在线生成

4.发送请求

5.获取URL参数

6.获取请求头参数

7.表单获取数据

8.自定义返回JSON信息

8.自定义返回HTML

9.自定义返回文件

10.自定义返回HTML页面

11.代办事项小案例

12.绑定数据库

13.数据库访问

14.数据库写入

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

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

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