FastAPI

前言

因为我本身是学习Java的，如果你也是，那么你学习 FastAPI 会非常顺畅！很多概念几乎可以无缝迁移。

让我用熟悉的 Spring Boot 视角来帮你快速建立对应关系：

核心概念对应表（Spring Boot → FastAPI）

Spring Boot 概念	FastAPI 对应概念	相似度
@RestController	@app.get()/@app.post() 等路由装饰器	★★★★★
@RequestMapping	路径参数和装饰器（@app.api_route）	★★★★☆
@Autowired	Depends() 依赖注入	★★★★☆
@Component/@Service	普通函数 + Depends()	★★★★☆
@Repository	数据库操作函数/类	★★★★☆
@Configuration	启动时初始化代码/依赖项	★★★☆☆
@Bean	使用 lifespan 或单例依赖	★★★☆☆
Interceptor/Filter	依赖项或中间件	★★★★☆
@ControllerAdvice	异常处理器（@app.exception_handler）	★★★★☆
@Valid 和 DTO	Pydantic 模型验证	★★★★★
application.properties	环境变量或 .env 文件	★★★★☆
Spring Security	依赖项 + OAuth2 工具	★★★★☆

你会立即感到熟悉的几个方面

1. 注解式开发 → 装饰器开发

# FastAPI（Python装饰器风格）
@app.get("/users/{id}")  # ← 就像 @GetMapping
def get_user(id: int, db: Session = Depends(get_db)):  # ← 自动注入依赖
    return db.query(User).filter(User.id == id).first()

2. 依赖注入的使用

# 定义一个“服务类”（类似Spring的@Service）
class UserService:
    def get_user(self, user_id: int):
        # 业务逻辑...
        return {"id": user_id, "name": "John"}
    
# 创建实例（类似@Bean）
user_service = UserService()

# 注入使用（类似@Autowired）
@app.get("/users/{id}")
def read_user(id: int, service: UserService = Depends(lambda: user_service)):
    return service.get_user(id)

3. 数据验证（DTO → Pydantic）

# 类似Java的DTO类
from pydantic import BaseModel

class UserCreate(BaseModel):  # ← 就像你的Java DTO
    username: str
    email: str
    age: int
    
    # 内置验证（类似@NotBlank @Email）
    @validator('email')
    def validate_email(cls, v):
        if '@' not in v:
            raise ValueError('Invalid email')
        return v

@app.post("/users/")
def create_user(user: UserCreate):  # ← 自动验证和转换
    # user.username, user.email 已经是验证过的数据
    return {"message": "User created"}

需要稍作调整的思维模式

1. 更函数式，更少“仪式感”

   • 不需要定义接口
   • 不需要 getter/setter
   • 没有复杂的 XML 配置

2. 更直接的请求-响应

   • 参数直接从函数参数获取
   • 返回值直接就是响应体
   • 异步支持是天生的（async/await）

3. 更灵活的依赖注入

   • 可以是函数，可以是类，都可以注入
   • 可以基于路径参数、查询参数、header 等动态创建依赖

学习建议

1. 直接找对应关系：每当你想“在 Spring Boot 里我会怎么做”，就去查 FastAPI 的对应方案
2. 从 Pydantic 开始：这是 FastAPI 的“杀手锏”，类似 Lombok + Validation 的超强组合
3. 理解 Python 的特性：类型提示、装饰器、async/await - 这些是 FastAPI 强大的基础
4. 别怕写“简单”代码：Python 代码看起来比 Java 简单，但功能一样强大

快速入门路径

# Day 1: 基本对应
# Java: @GetMapping("/hello") → FastAPI:
@app.get("/hello")
def hello(name: str = "World"):  # 查询参数自动处理
    return {"message": f"Hello {name}"}

# Day 2: 依赖注入
# Java: @Autowired UserService → FastAPI:
def get_db():  # 类似 @Bean
    db = SessionLocal()
    try:
        yield db  # 类似 @Scope("request")
    finally:
        db.close()

@app.get("/items")
def get_items(db: Session = Depends(get_db)):  # 自动注入
    return db.query(Item).all()

# Day 3: 安全验证
# Java: Spring Security → FastAPI:
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def get_current_user(token: str = Depends(oauth2_scheme)):
    # 验证逻辑...
    return user

@app.get("/users/me")
def read_me(current_user = Depends(get_current_user)):
    return current_user

好消息是…

你的 Spring Boot 经验有 80% 可以直接迁移！

• REST API 设计
• 分层架构（Controller/Service/Repository）
• 依赖注入的价值
• 数据验证的重要性
• 配置管理
• 错误处理

FastAPI 只是用更 Pythonic、更简洁的方式实现了相同的理念。你会发现在 FastAPI 中实现同样的功能，代码量可能只有 Spring Boot 的 1/3 到 1/2，但功能和性能一样强大。

我们学习java学的是理念，FastAPI 只是换了种更轻量、更现代的语言和语法来实现它。

下面是我的笔记~

FastAPI 入门 - FastAPI 框架

Python 版本最低3.8 推荐3.10以上 我是用3.13

Python环境

为了适应各种Python版本，这里使用MiniConda 来作为Python环境，MiniConda 可以在创建虚拟环境的时候指定Python 版本号，并且没有Anaconda 那么多乱七八糟的第三方库

一：MiniConda

Miniconda - Anaconda

Download Success - post download | Anaconda

安装完后要将路径添加到环境变量当中

# 根路径
D:\miniconda
# Scripts 路径
D:\miniconda\Scripts
# bin路径
D:\miniconda\Library\bin

验证：

C:\Users\Lenovo>conda <==== 输入conda
usage: conda-script.py [-h] [-v] [--no-plugins] [-V] COMMAND ...

conda is a tool for managing and deploying applications, environments and packages.

options:
  -h, --help            Show this help message and exit.
  -v, --verbose         Can be used multiple times. Once for detailed output, twice for INFO logging, thrice for DEBUG
                        logging, four times for TRACE logging.
  --no-plugins          Disable all plugins that are not built into conda.
  -V, --version         Show the conda version number and exit.

commands:
  The following built-in and plugins subcommands are available.

  COMMAND
    activate            Activate a conda environment.
    clean               Remove unused packages and caches.
    commands            List all available conda subcommands (including those from plugins). Generally only used by
                        tab-completion.
    compare             Compare packages between conda environments.
    config              Modify configuration values in .condarc.
    content-trust       Signing and verification tools for Conda
    create              Create a new conda environment from a list of specified packages.
    deactivate          Deactivate the current active conda environment.
    doctor              Display a health report for your environment.
    env                 Create and manage conda environments.
    export              Export a given environment
    info                Display information about current conda install.
    init                Initialize conda for shell interaction.
    install             Install a list of packages into a specified conda environment.
    list                List installed packages in a conda environment.
    menuinst            A subcommand for installing and removing shortcuts via menuinst.
    notices             Retrieve latest channel notifications.
    package             Create low-level conda packages. (EXPERIMENTAL)
    remove (uninstall)  Remove a list of packages from a specified conda environment.
    rename              Rename an existing environment.
    repoquery           Advanced search for repodata.
    run                 Run an executable in a conda environment.
    search              Search for packages and display associated information using the MatchSpec format.
    token               Set repository access token and configure default_channels
    tos                 A subcommand for viewing, accepting, rejecting, and otherwise interacting with a channel's
                        Terms of Service (ToS). This plugin periodically checks for updated Terms of Service for the
                        active/selected channels. Channels with a Terms of Service will need to be accepted or
                        rejected prior to use. Conda will only allow package installation from channels without a
                        Terms of Service or with an accepted Terms of Service. Attempting to use a channel with a
                        rejected Terms of Service will result in an error.
    update (upgrade)    Update conda packages to the latest compatible version.

C:\Users\Lenovo>

二：虚拟环境

默认情况下，我们直接通过pip install 的方式将包安装在系统级别的Python环境中，这样存在一个问题，就是如果有个老项目用FastAPI是老版本做的，然后你现在想用最新的FastAPI版本来开发新项目，这时候就会碰到一个问题，如何在电脑中同事拥有两套环境？这个时候就是使用我们的虚拟环境来解决这个问题。

首先，我们创建一个基于Python 3.13 版本的虚拟环境：

# 使用3.13
conda create -n fastapi-env python=3.13

通过下列命令可以管理虚拟环境：

# 查看所有的虚拟环境
conda env list
# 删除虚拟环境
conda env remove -n [虚拟环境名称]

# 进入虚拟环境
# 首次运行要执行：conda init
conda activate [虚拟环境名称]

# 退出虚拟环境
conda deactivate

三：设置pip源

默认情况下，使用pip命令会从python官网上下载，由于Python官网服务器在国外，下载速度有点慢，我们可以改为国内滴，比如清华,搜索引擎搜索一下也有很多其他的

pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple

结果：

C:\Users\Lenovo>conda activate fastapi-env

(fastapi-env) C:\Users\Lenovo>pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple
Writing to C:\Users\Lenovo\AppData\Roaming\pip\pip.ini

开发工具

PyCharm

这个不多说啦

我是教育邮箱申请的pro

或者你可以去某宝购买。

Mysql

javaer应该都会的

Navicat

也可以使用其他的数据库可视化软件

FastAPI 介绍

一:FastAPI 简介

FastAPI 是一个高性能的 Python Web 框架，旨在为开发人员提供快速、简洁且强大的 API 开发体验。它于 2017 年由 Sebastian Ramirez 发布，旨在解决当时市场上现有框架在处理 HTTP 请求和响应时的复杂性和维护困难问题。

Sebastian Ramirez 在开发 FastAPI 时，面临着构建 API 的挑战，他发现现有的框架在处理 HTTP 请求和响应时过于复杂，难以维护。因此，他决定开发一个更简洁、更易于使用的框架，以帮助其他开发人员更专注于业务逻辑，而不是处理 HTTP 请求和响应。

FastAPI 支持异步编程，这意味着它可以更有效地处理并发请求，提高应用程序的性能。此外，FastAPI 还支持多种数据库和数据存储后端，如 PostgreSQL、MySQL、SQLite、MongoDB 等，以及 ORM 框架，如 SQLAlchemy。这使得开发者可以轻松地将 FastAPI 与各种数据库集成，构建强大的数据驱动应用程序。

许多知名的公司和项目已经开始使用 FastAPI，包括 Google、Netflix、Amazon、Dropbox 等。这些公司在使用 FastAPI 时，都对其高性能和简洁性给予了高度评价。

FastAPI 的基准测试成绩也证明了其高性能。根据官方的基准测试结果，FastAPI 在处理并发请求时，其性能优于其他流行的 Web 框架，如 Flask 和 Django。这使得 FastAPI 成为构建高性能 Web 应用程序和 API 的理想选择。

Django、Flask 和 FastAPI 三大框架的特点：

• 功能：Django > Flask > FastAPI
• 性能：FastAPI > Flask > Django

二:FastAPI的优点

• 快速：与NodeJS和Go性能相当，是最快的Python web框架之一。
• 高效编码：提升功能开发速度200%至300%。
• 更少bug：减少约40%开发者人为错误。
• 智能：编辑器自动补全以减少调试时间。
• 简单：易使用和学习，读文档时间更短。
• 简短：代码重复最小化，通过参数声明实现丰富功能，bug更少。
• 健壮：生产可用级别代码，含自动生成的交互式文档。
• 标准化：基于OpenAPI和JSON Schema。

三:安装

使用下面这个命令即可安装：

pip install "fastapi[standard]"

四:一个最简单的FastAPI程序

在Pycharm中选择FastAPI项目新建

当然了记得创建.gitignore 因为ide会自带一些文件

from fastapi import FastAPI

app = FastAPI()

# 服务器在定义URL的时候，用了什么method，那么客户端在请求这个URL的时候就要使用相同的method
# GET:从服务器上获取资源
# POST：数据到服务器
# DELETE: 要删除服务器上的数据
# PUT：要求改服务器上的数据

# async
@app.get("/")
async def root():
    # 异步函数，协程
    # await 访问数据库() 等IO操作

    return {"message": "Hello World"}

# /hello/yjy
# /hello/yangjiayu
@app.get("/hello/{name}")
async def say_hello(name: str):
    return {"message": f"Hello {name}"}

Uvicorn

uvicorn是一个高性能的异步web服务器。我们也可以使用uvicorn来直接运行FastAPI项目。首先通过以下命令安装uvicorn:

pip install "uvicorn[standard]"

然后可以通过下面命令运行项目：

uvicorn main:app --reload

测试API

另外我们可以在项目根路径下test_main.http 中测试我们写的API是否有错误

具体项目代码（后续所有代码都放在这里）：yjyrichard/fastapi-learning: fastapi学习代码

Pydantic介绍

在其他web框架中，比如Django或Flask，我们通过定义表单类来校验数据。而在FastAPI中，我们使用Pydantic来实现这一功能。

Pydantic的特点：

• 由类型提示支持：少学习、代码少、与IDE工具集成。
• 速度快：Rust核心验证，Python最快数据验证库之一。
• JSON模式：与其他工具集成。
• 生态系统丰富：PyPI约8000个包，含FastAPI等流行库。
• Battle测试：月下载超7000万次，被多家大型科技公司使用。

安装

由于FastAPI依赖Pydantic，因此在安装FastAPI时会把pydantic也一起安装了。

基本使用

Pydanyic 是通过定义模型，以及在模型中指定字段来对值进行校验的，示例代码如下：

# 1. 导入所需的模块
from datetime import date
from typing import List
# 除了BaseModel，我们还需要导入ValidationError来处理验证错误
from pydantic import BaseModel, ValidationError

# 2. 定义数据模型 (与之前相同)
class User(BaseModel):
    """
    定义一个名为User的模型，它继承自pydantic.BaseModel。
    这个类定义了我们期望的数据结构、类型和约束。
    """
    id: int  # 必填字段，必须是整数
    name: str = 'John Doe'  # 可选字段，默认值为'John Doe'
    date_joined: date | None  # 可选字段，可以是日期或None
    departments: List[str] | None  # 可选字段，可以是字符串列表或None

# --- 场景一：使用有效数据，验证成功 ---
print("--- 场景一：使用有效数据 ---")
# 准备符合模型要求的外部数据
valid_data = {
    'id': 123,
    'date_joined': '2023-06-01', # Pydantic会自动将字符串转换为date对象
    'departments': ['技术部', '产品部']
}

try:
    # 尝试创建User实例
    user = User(**valid_data)
    print("✅ 数据验证成功！")
    print(f"用户ID: {user.id}")
    print(f"用户姓名: {user.name}")
    print(f"加入日期: {user.date_joined} (类型: {type(user.date_joined)})")
except ValidationError as e:
    # 如果数据有效，这个块不会被执行
    print("❌ 数据验证失败！")
    print(e)

print("\n" + "="*50 + "\n")

# --- 场景二：使用无效数据，捕获ValidationError ---
print("--- 场景二：使用无效数据并捕获错误 ---")
# 准备不符合模型要求的外部数据
invalid_data = {
    'id': 'abc', # 错误：id应该是int，但这里是str
    'name': 456,   # 错误：name应该是str，但这里是int
    'date_joined': '2023-13-01', # 错误：这不是一个有效的日期格式
    'departments': '不是列表' # 错误：departments应该是List[str]，但这里是str
}

try:
    # 尝试用无效数据创建User实例
    # 这一行会抛出 ValidationError
    user_invalid = User(**invalid_data)
    print("✅ 数据验证成功！") # 这行代码不会被执行

except ValidationError as e:
    # 捕获到ValidationError异常
    print("❌ 数据验证失败！捕获到 ValidationError。")
    
    # 打印一个人类可读的错误摘要
    print("\n--- 错误摘要 ---")
    print(e)

    # 打印详细的错误列表（推荐用于程序化处理）
    # e.errors() 返回一个包含所有错误详细信息的字典列表
    print("\n--- 详细错误列表 ---")
    error_list = e.errors()
    for i, error in enumerate(error_list, 1):
        print(f"错误 {i}:")
        # 'loc' 字段指明了错误发生的位置（字段名）
        print(f"  - 位置: {' -> '.join(map(str, error['loc']))}")
        # 'msg' 字段提供了人类可读的错误信息
        print(f"  - 信息: {error['msg']}")
        # 'type' 字段是机器可读的错误类型标识符
        print(f"  - 类型: {error['type']}")
        print("-" * 20)

请求数据

请求数据包括有路由参数，Body参数，以下分别进行讲解：

这个跟java没有什么区别 直接来看例子：

# 导入FastAPI及相关类型和工具
from fastapi import FastAPI, Body, Query, Path, Form, File, UploadFile
from pydantic import BaseModel, Field
from typing import Annotated

# 创建FastAPI应用实例
app = FastAPI()

# --- 1. 路径参数 ---
# 路径参数是URL路径的一部分，用于标识特定资源。
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    # FastAPI会自动将URL中的item_id转换为int类型
    # 如果传入的不是整数，会返回一个清晰的验证错误
    return {"item_id": item_id}

# --- 2. 查询参数 ---
# 查询参数是URL中 `?` 后面的键值对，用于过滤或修改请求。
@app.get("/users/")
async def read_users(q: Annotated[str | None, Query(max_length=50)] = None):
    # q是可选的查询参数，如果没有提供，默认为None
    # 使用Query可以添加额外的验证，如最大长度
    if q:
        return {"query": q}
    return {"message": "No query parameter provided"}

# --- 3. 请求体 - Pydantic 模型 ---
# 请求体通常用于POST、PUT等请求，通过JSON格式传递复杂的数据结构。
# 首先定义一个Pydantic模型用于数据验证和序列化
class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None

@app.post("/items/")
async def create_item(item: Item):
    # 声明item参数的类型为Item模型
    # FastAPI会读取请求体，验证数据，并将其转换为Item实例
    return item

# --- 4. 混合使用路径、查询和请求体参数 ---
# 一个操作中可以同时包含多种类型的参数。
@app.put("/items/{item_id}")
async def update_item(
    item_id: Annotated[int, Path(title="The ID of the item to update", ge=1)],
    q: str | None = None,
    item: Annotated[Item, Body(embed=True)]
):
    # item_id: 路径参数，使用Path添加了元数据和验证（必须大于等于1）
    # q: 查询参数
    # item: 请求体参数，Body(embed=True)表示JSON需要有一个键"item"来包裹数据
    results = {"item_id": item_id, "item": item}
    if q:
        results["q"] = q
    return results

# --- 5. 表单数据 ---
# 当Content-Type为 `application/x-www-form-urlencoded` 时使用，常用于HTML表单提交。
@app.post("/login/")
async def login(username: str = Form(), password: str = Form()):
    # 使用Form来声明表单字段
    return {"username": username}

# --- 6. 文件上传 ---
# 当Content-Type为 `multipart/form-data` 时使用，用于上传文件。
@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    # 使用UploadFile来接收文件，它比直接使用bytes更节省内存
    # file.filename: 获取文件名
    # file.content_type: 获取文件类型
    # await file.read(): 异步读取文件内容
    return {
        "filename": file.filename,
        "content_type": file.content_type
    }

# --- 7. 文件和表单字段混合 ---
# 可以在同一个请求中同时接收文件和表单字段。
@app.post("/files/")
async def create_file(
    file: bytes = File(), description: str = Form(...)
):
    # 使用bytes = File()会将文件内容全部读入内存
    # 使用Form接收其他表单字段
    # 适用于小文件
    return {
        "file_size": len(file),
        "description": description
    }

依赖注入

依赖注入可以让我们的视图函数在执行之前先执行一段逻辑代码，这段逻辑代码可以返回新的值给视图函数。在以下场景中可以使用依赖注入：

• 共享业务逻辑（复用相同的代码逻辑）
• 共享数据库连接
• 实现安全、验证、角色权限
• 等等。

总之，就是将一些重复性的代码单独写成依赖，然后在需要的视图函数中注入这个依赖。

因为我之前是学习Java的 这样解释我反而有点看不太懂

跟spring的核心理念是相同的意思是：

“控制反转(IoC)”和“依赖倒置(DIP)”：将对象的创建和管理权交给框架，使用者只声明“我需要什么”，框架负责“提供什么”。

springboot这样写嘛：

@Service
public class UserService {
    public User getUser(Long id) {
        // ... 业务逻辑
    }
}

@RestController
public class UserController {
    @Autowired  // 核心：将业务Bean注入控制器
    private UserService userService;

    @GetMapping("/user/{id}")
    public User getUser(@PathVariable Long id) {
        return userService.getUser(id);
    }
}

fastapi这样写：

# 1. 声明一个依赖项（如权限检查）
def get_current_user(token: str = Depends(oauth2_scheme)):
    # ... 验证token，返回用户信息
    return user

# 2. 在视图函数中注入这个依赖
@app.get("/users/me")
def read_users_me(current_user: User = Depends(get_current_user)):
    # 框架会自动先执行 get_current_user，将其结果注入给 current_user
    return current_user

总之：就是别总是 “new ”啦 找框架要吧。

将一些重复性的代码单独写成以来，任何在需要的视图函数中，注入这个依赖。示例代码如下：

# 依赖注入
from fastapi import Depends
from typing import Dict

async def page_common(page: int=0,size: int=10):
    return {"page": page,"size": size}

@app.get("/user/list")
async def user_list(page_params: Dict=Depends(page_common)):
    page = page_params["page"]
    size = page_params["size"]
    return {"page": page,"size": size}

APIRouter

在FastAPI项目中，我们不可能把所有视图都放到main.py 文件中，这时候久需要将视图进行分类，任何同类的放到一个单独的子路由下。在FastAPI 中可以使用 APIRouter 类进行实现。【好比 java里面的 [业务名称]Controller 这样看起来更加清晰嘛】

新建一个包：routers

里面新建两个py文件分别为：article.py 和 user.py

user.py

from fastapi import APIRouter

router = APIRouter(prefix="/user", tags=["user"])

# /user/list
@router.get("/list")
async def user_list():
    return {"users" : ["zs","ls"]}

# /user/123
@router.get('/{user_id}')
async def user_detail(user_id):
    return {"user_id" : user_id}

article.py 也差不多 就不粘贴了

现在还没有跟main.py进行关联

from fastapi import FastAPI
from routers.user import router as user_router
from routers.article import router as article_router #避免覆盖
app = FastAPI()
app.include_router(user_router)
app.include_router(article_router)

这个配置方式让我感觉跟vue很像

想象成组装一台电脑：

• FastAPI/Vue 方式（模块化组装）：

  1. 买一个空机箱（app = FastAPI() / createApp()）
  2. 买独立显卡（user_router）
  3. 买独立内存条（article_router）
  4. 组装起来（app.include_router()）

  灵活：需要什么装什么

• Spring Boot 方式（一体化方案）：

  1. 买品牌整机（@SpringBootApplication）
  2. 开机即用（自动扫描所有组件）
  3. 配置都在说明书里（application.properties）

  省心：开箱即用，功能齐全

学习迁移建议

1. FastAPI 的 app ≈ Vue 的 app ≈ Spring Boot 的启动类
   • 都是应用的核心容器
   • 都负责全局配置
   • 都是模块注册的中心
2. FastAPI 的 APIRouter ≈ Vue 的组件 ≈ Spring Boot 的 @Controller
   • 都是功能模块
   • 都可以独立开发和测试
   • 都可以被主应用"挂载"
3. 最大的区别：
   • FastAPI/Vue: 显式注册（你要手动 include_router）
   • Spring Boot: 隐式注册（自动扫描 @Controller）

现代框架都在向 “核心应用 + 模块化插件” 的模式发展。这种模式让代码更清晰、更易维护，也让学习曲线更平滑——学会一个，其他的就很容易理解了。

app = FastAPI() 就是 FastAPI 世界的 “Spring Boot Application”，是整个应用的启动器和配置中心。**

数据库连接和模型

在Python框架中，我们通常会使用ORM（Object Relationship Mapping）框架来操作数据库。目前主流的ORM框架有：SQLALchemy、Peewee ORM、Pony ORM、GINO、Tortoise ORM、orman等。而SQLALchemy是功能最强大，且社区最活跃的ORM库，并且还支持异步，因此我选择用SQLALchemy作为操作数据库的ORM框架。

另外，FastAPI作者开发了一个ORM框架叫做SQLModel，但是这个框架目前来说还不完善，文档都还处于开发期间，它底层也是基于SQLALchemy Core来实现的。SQLAlchemy文档地址为：https://www.sqlalchemy.org/

安装

1.安装sqlalchemy

通过以下命令安装：

pip install "sqlalchemy[asyncio]"

将会安装异步版本的sqlalchemy

然后再执行以下命令安装aiomysql 驱动：

# 同步：mysqlclient
# 异步：aimysql
pip install aiomysql

2.安装cryptography

使用Python连接MySQL需要用cryptography对密码进行加密，所以还需要安装以下包：

pip install cryptography

创建连接

配置连接参数

使用SQLAlchemy 连接数据库，是通过设置一个固定格式的字符串来实现的。mysql+aiomysql 为例，那么其连接格式如下：

DB_URL = "mysql+aiomysql://用户名:密码@主机名:端口号/数据库名称?charset=utf8mb4"

示例如下：

DB_URL = "mysql+aiomysql://root:root@127.0.0.1:3306/bookdb?charset=utf8mb4"

创建Engine 对象

SQLAlchemy 中的 Engine 对象，它负责管理数据库连接的创建(并不直接操作数据库)，连接池的维护，SQL语句的翻译等。Engine对象在整个程序中只能有一个。创建Engine对象的代码如下：

from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(
    DB_URI,
    # 将输出所有执行SQL的日志（默认是关闭的） 生产环境关掉
    echo=True,
    # 连接池大小（默认是5个）
    pool_size=10,
    # 允许连接池最大的连接数（默认是10个）
    max_overflow=20,
    # 获得连接超时时间（默认是30s）
    pool_timeout=10,
    # 连接回收时间（默认是-1，代表永不回收）
    pool_recycle=3600,
    # 连接前是否预检查（默认为False）
    pool_pre_ping=True,
)

创建会话工厂

使用 sqlalchemy_orm.sessionmarker 类来创建会话工厂，这个会话工厂实际就是Session 或者它的子类，以后如果要操作数据库，那么就需要创建一个会话工厂的对象 (即：Session类的对象)，来完成相关操作。示例代码如下：

from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.asyncio import AsyncSession

AsyncSessionFactory = sessionmaker(
    # Engine或者其子类对象（这里是AsyncEngine）
    bind=engine,
    # Session类的代替（默认是Session类）
    class_=AsyncSession,
    # 是否在查找之前执行flush操作（默认是True）
    autoflush=True,
    # 是否在执行commit操作后Session就过期（默认是True）
    expire_on_commit=False
)

创建模型

定义Base 类

Base 类是所有ORM Model 类的父类，一个ORM Model 类对应数据库中的一张表。ORM Model 中的一个Column 类属性对应数据库表中的一个字段。Base类的生成可以使用 sqlalchemy.ext.declarative.declarative_base 函数来实现，也可以继承 sqlalchemy.MetaData 类，实现自己的子类，并在子类中编写约束规范。

示例代码如下：

from sqlalchemy.orm import DeclarativeBase
from sqlalchemy import MetaData

# 定义命名约定的Base类
class Base(DeclarativeBase):
    metadata = MetaData(naming_conventions={
        # ix: index, 索引。
        "ix": 'ix_x(column_0_label)s',
        # un: unique, 唯一约束
        "uq": "uq_x(table_name)s_x(column_0_name)s",
        # ck: Check, 检查约束
        "ck": "ck_x(table_name)s_x(constraint_name)s",
        # fk: Foreign Key, 外键约束
        "fk": "fk_x(table_name)s_x(column_0_name)s_x(referred_table_name)s",
        # pk: Primary Key, 主键约束
        "pk": "pk_x(table_name)s"
    })

创建ORM模型

这里我们以创建一个User模型为例来说明模型的创建：

from typing import Optional
from sqlalchemy import Integer, String, select
from sqlalchemy.orm import Mapped, mapped_column

class User(Base):
    __tablename__ = 'user'
    
    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    email: Mapped[str] = mapped_column(String(100), unique=True, index=True)
    username: Mapped[str] = mapped_column(String(100))
    password: Mapped[str] = mapped_column(String(200))

模型关系

这里我们使用外键来实现一对一，一对多和多对多，示例代码如下：