👏🏻 你好！欢迎访问IT教程网，0门教程，教程全部原创，计算机教程大全，全免费！

🔥 新增教程

《黑神话 悟空》游戏开发教程，共40节，完全免费，点击学习

《AI副业教程》，完全原创教程，点击学习

发表更新软件开发 / FastAPI教程6 分钟读完 (大约830个字)

25 中间件与依赖注入之中间件的概念与使用

在上一篇文章中，我们探讨了如何在 FastAPI 中进行错误处理，尤其是处理 HTTP 异常响应。本文将重点介绍中间件的概念以及如何在 FastAPI 中使用中间件。接下来，我们将为即将到来的依赖注入主题做好铺垫。

中间件的概念

在 Web 应用程序中，中间件是位于请求和响应处理之间的组件。它们通常用于：

  • 记录日志
  • 跨域资源共享（CORS）
  • 身份验证和授权
  • 请求修改
  • 响应修改
  • 处理请求失败的情况

在 FastAPI 中，中间件可以通过 FastAPI 的应用实例添加，它们会包裹在请求处理的过程之间。

创建中间件

要在 FastAPI 中创建中间件，我们需要使用 starlette.middleware 模块。下面是一个简单的中间件示例，它会在每个请求时记录请求的路径。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware

class SimpleLoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        # 在请求处理之前执行的操作
        print(f"请求路径: {request.url.path}")
        
        # 调用下一个中间件/处理程序
        response = await call_next(request)
        
        # 在请求处理之后执行的操作
        print(f"返回状态: {response.status_code}")
        
        return response

app = FastAPI()

# 添加中间件
app.add_middleware(SimpleLoggingMiddleware)

@app.get("/")
async def read_root():
    return {"message": "Hello, World!"}

在这个例子中，我们创建了一个 SimpleLoggingMiddleware 类，它继承了 BaseHTTPMiddleware。这个中间件会在每个请求被处理时，记录请求的路径与响应的状态码。这对于调试和跟踪请求流程非常有帮助。

中间件的顺序

请注意，中间件的执行顺序是依据它们添加的顺序。最先添加的中间件会在请求处理链的最外层，最后添加的中间件会在请求链的最内层。以下是一个示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
class FirstMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        print("第一中间件开始")
        response = await call_next(request)
        print("第一中间件结束")
        return response

class SecondMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        print("第二中间件开始")
        response = await call_next(request)
        print("第二中间件结束")
        return response

app.add_middleware(FirstMiddleware)
app.add_middleware(SecondMiddleware)

在这个示例中，输出将会是：

1
2
3
4
第一中间件开始
第二中间件开始
第二中间件结束
第一中间件结束

这说明第一个中间件包裹了整个处理过程，而第二个中间件是在它的内部执行。

快速总结

  • 中间件是请求和响应之间的处理层，可以用于多种目的，例如日志记录、身份验证和请求修改等。
  • 在 FastAPI 中，可以通过 add_middleware 方法添加中间件。
  • 中间件的执行顺序是基于添加的顺序。

在接下来的文章中，我们将深入探讨依赖注入的基本用法。这是 FastAPI 的一个强大特性，可以帮助我们更好地管理应用程序的依赖关系。

希望本篇关于中间件的概念与使用的讨论能帮助你更好地理解 FastAPI 的工作机制！

分享转发

发表更新软件开发 / FastAPI教程7 分钟读完 (大约1087个字)

26 中间件与依赖注入之依赖注入的基本用法

在上一篇中，我们讨论了 FastAPI 中间件的概念与使用方法。这篇文章将重点介绍 FastAPI 中的依赖注入（Dependency Injection，DI），并讲解其基本用法。依赖注入是 FastAPI 的一个强大特性，能够帮助我们管理复杂的应用程序逻辑。

一、依赖注入的基本概念

在 FastAPI 中，依赖注入是一种在处理请求时将特定功能或数据注入到路由处理函数中的方法。这可以让你在多个路由中复用逻辑，提高代码的可维护性和清晰度。通过将一些逻辑提取到依赖项中，我们可以让路由处理函数更专注于其核心功能。

依赖项可以是简单的函数或类，都能被 FastAPI 自动解析并注入。

二、定义依赖项

1. 创建简单的依赖项

以下是一个简单的依赖项示例。我们将创建一个函数，它将返回一个字符串。当我们在路由处理函数中使用这个依赖项时，它将被注入到相应的函数中。

1
2
3
4
5
6
7
8
9
10
11
from fastapi import FastAPI, Depends

app = FastAPI()

# 定义一个简单的依赖项
def get_query_param(q: str = None):
    return q

@app.get("/items/")
async def read_item(q: str = Depends(get_query_param)):
    return {"q": q}

在上面的代码中，get_query_param 函数是一个依赖项，其参数 q 通过 FastAPI 的请求查询参数自动填充。当访问 /items/?q=example 时，返回的 JSON 将是 {"q": "example"}。

2. 依赖项的参数

依赖项的参数可以有默认值，或者可以通过请求上下文自动填充。下面的示例定义了一个更复杂的依赖项，包括默认值和类型注解。

1
2
3
4
5
6
7
8
9
10
11
12
13
from fastapi import FastAPI, Depends, HTTPException

app = FastAPI()

# 定义一个依赖项，带有参数和异常处理
def verify_query_param(q: str):
    if q != "allowed":
        raise HTTPException(status_code=400, detail="Query param must be 'allowed'")
    return q

@app.get("/secure-items/")
async def read_secure_item(q: str = Depends(verify_query_param)):
    return {"q": q}

这里，如果用户请求的 q 参数不等于 "allowed"，会抛出一个 400 错误。在这一点上，我们就实现了一个简单的参数验证。

三、依赖项的复用

依赖项可以被多个路由使用，极大地提升了代码复用性。我们可以将常用的依赖项提取到单独的函数中，供多个路由使用。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from fastapi import FastAPI, Depends

app = FastAPI()

# 创建依赖项
def get_query_param(q: str = None):
    return q

@app.get("/items/")
async def read_item(q: str = Depends(get_query_param)):
    return {"q": q}

@app.get("/another-items/")
async def read_another_item(q: str = Depends(get_query_param)):
    return {"another_q": q}

在上述示例中，两个不同的路由都使用了同一个依赖项 get_query_param。无论是 /items/ 还是 /another-items/，都能复用这段代码。

四、使用类作为依赖项

除了函数，FastAPI 还支持将类用作依赖项。这通常用于需要状态或初始化的复杂依赖项。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from fastapi import FastAPI, Depends

app = FastAPI()

class QueryParam:
    def __init__(self, q: str = None):
        self.q = q

# 定义一个依赖项类
def get_query_param_class(q: str = None):
    return QueryParam(q)

@app.get("/class-items/")
async def read_item(query_param: QueryParam = Depends(get_query_param_class)):
    return {"q": query_param.q}

在这个例子中，我们定义了一个 QueryParam 类，并将其实例作为依赖项注入到路由处理函数中。这样，可以轻松扩展功能，保持代码的整洁性。

五、总结

依赖注入是 FastAPI 的一个重要特性，它允许我们轻松地重用逻辑、管理复杂性并增强代码的可读性。在本篇文章中，我们介绍了依赖项的基本用法，以及如何创建和使用依赖项。

在下一篇文章中，我们将继续探讨 全局依赖 与 路径依赖，如何在 FastAPI 中灵活运用这些特性来构建更为复杂的应用逻辑。希望本篇教程能让你对 FastAPI 的依赖注入有一个基本的了解，并能在实际项目中灵活应用。

分享转发

发表更新软件开发 / FastAPI教程5 分钟读完 (大约817个字)

27 中间件与依赖注入之全局依赖与路径依赖

在前面的章节中，我们讨论了 FastAPI中 依赖注入的基本用法。在这一篇中，我们将深入探讨全局依赖与路径依赖的概念以及如何在 FastAPI 中实现它们。这将对我们后续的数据库操作提供良好的基础。

依赖注入的概念

在 FastAPI 中，依赖注入的核心思想是将一个部分的功能（即依赖）传递给其他部分，而不是在每个部分中重复实现。这可以帮助我们保持代码的整洁和可维护性。全局依赖和路径依赖都是依赖注入的一种实现方式。

全局依赖

全局依赖是指在整个应用中都可以使用的依赖。无论哪个路径调用，都会自动使用这些依赖。

实现全局依赖

在 FastAPI 中，我们可以通过 app.dependency_overrides 来实现全局依赖。以下是一个示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from fastapi import FastAPI, Depends, HTTPException

app = FastAPI()

# 创建一个简单的依赖
def get_query_param(q: str = None):
    return q

# 将其设置为全局依赖
app.dependency_overrides[get_query_param] = lambda: "全局依赖注入的查询参数"

@app.get("/")
async def read_root(query_param: str = Depends(get_query_param)):
    return {"message": f"你传入的全局依赖参数是: {query_param}"}

在这个示例中，无论何时调用根路径 /，都会自动使用 get_query_param 的全局依赖。在这个例子中，返回的消息将始终为 "全局依赖注入的查询参数"。

路径依赖

路径依赖是指仅在特定路径中使用的依赖。可以根据不同的路径定义不同的依赖，允许更多的灵活性。

实现路径依赖

我们可以在 path operation 函数中使用依赖注入，轻松实现路径依赖。以下是一个示例：

1
2
3
4
5
6
7
8
9
10
11
from fastapi import FastAPI, Depends

app = FastAPI()

# 创建一个特定于路径的依赖
def get_user(user_id: int):
    return {"user_id": user_id}

@app.get("/users/{user_id}")
async def read_user(user: dict = Depends(get_user)):
    return {"message": f"你获取到的用户ID是: {user['user_id']}"}

在这个示例中，get_user 依赖仅在 GET /users/{user_id} 路径中使用。我们可以看到，这种方式允许我们为每个路径定义特定的依赖。

小结

在 FastAPI 中，全局依赖和路径依赖是非常有用的功能，帮助我们实现灵活且可重用的代码结构。全局依赖适合在整个应用中使用的通用功能，而路径依赖则适合于特定路径且有可能不同的任务或逻辑。通过合理地利用这两种依赖注入的方式，我们能够编写出清晰且易于维护的 API。

在下一节中，我们将探索如何使用 SQLAlchemy 来连接数据库，这将使我们的应用变得更加强大，能够持久化数据。希望通过这些内容，你能够更加熟悉 FastAPI 中的依赖注入概念，为后续的数据库操作打下坚实的基础。

分享转发

发表更新软件开发 / FastAPI教程5 分钟读完 (大约752个字)

28 使用数据库之SQLAlchemy连接数据库

在上一篇中，我们探讨了 FastAPI 的中间件与依赖注入，理解了如何使用全局依赖与路径依赖来管理请求中的共享资源。本篇将重点讲解如何在 FastAPI 应用中使用 SQLAlchemy 连接到数据库，以便进行数据持久化。

安装依赖

首先，您需要确保安装了 FastAPI 和 SQLAlchemy。可以使用 pip 安装所需的库：

1
pip install fastapi[all] sqlalchemy databases
  • fastapi[all]: 包含 FastAPI 的所有功能。
  • sqlalchemy: Python 的 SQL 工具包和对象关系映射器。
  • databases: 一个异步数据库库，适用于 SQLAlchemy。

创建数据库模型

接下来，我们需要定义我们的数据库模型。假设我们正在创建一个简单的用户管理系统，用户具有 id、name 和 email 三个属性。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./test.db"  # 使用 SQLite 数据库

# 创建数据库引擎
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})

# 创建会话
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 创建基本模型类
Base = declarative_base()

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    email = Column(String, unique=True, index=True)

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

在上面的代码中，我们定义了一个 User 模型，并使用 SQLAlchemy 的 declarative_base 创建基础模型类。然后，我们通过调用 Base.metadata.create_all(bind=engine) 创建了数据库中对应的表。

数据库连接的依赖注入

在 FastAPI 中，使用依赖注入来管理数据库会话是一种简洁而有效的方法。下面的代码展示了如何为 FastAPI 应用配置数据库会话的依赖项：

1
2
3
4
5
6
7
8
9
10
11
from fastapi import FastAPI, Depends, HTTPException

app = FastAPI()

# 依赖项：获取数据库会话
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

在这里，get_db 函数定义了一个生成器，它创建了一个数据库会话并在请求结束时自动关闭会话。

结论

到目前为止，我们已经定义了数据库模型并设置了 SQLAlchemy 连接。通过依赖注入，我们也准备好了数据库会话的管理。接下来的章节，我们将深入探讨如何在此基础上实现基本的 CRUD 操作，这是数据管理系统中的重要环节。

如果您在中间件与依赖注入的上下文中考虑数据库连接，那么这个简单的依赖注入实现能够让数据库连接在请求的生存期内保持有效，同时确保在请求完成后能正确关闭连接。

接下来，我们将讨论如何执行基本的 CRUD 操作，进一步完善我们的用户管理系统。请继续关注本系列教程！

分享转发

发表更新软件开发 / FastAPI教程7 分钟读完 (大约979个字)

29 使用数据库之基本CRUD操作

在上一篇中，我们探讨了如何使用 SQLAlchemy 连接数据库。在本节中，我们将继续扩展我们的 FastAPI 应用程序，实施基本的 CRUD（创建、读取、更新、删除）操作。这些操作可以帮助我们有效地管理数据库中的数据。在接下来的内容中，我们将通过案例演示如何实现这些操作。

准备工作

在开始之前，请确保你已经安装了所需的依赖包。你可以使用以下命令安装 FastAPI 和 SQLAlchemy ：

1
pip install fastapi[all] sqlalchemy databases

我们还需要配置数据库连接。以下示例将使用 SQLite 数据库。

1
2
3
4
5
6
7
8
9
10
# database.py
from sqlalchemy import create_engine, MetaData
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

创建模型

接下来，我们需要定义用于 CRUD 操作的数据库模型。例如，假设我们正在管理一个 用户 数据库表。

1
2
3
4
5
6
7
8
9
10
11
12
13
# models.py
from sqlalchemy import Column, Integer, String
from database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    email = Column(String, unique=True, index=True)

    def __repr__(self):
        return f"<User(name={self.name}, email={self.email})>"

现在我们已经定义了 User 模型，接下来将创建 CRUD 操作。

CRUD 操作实现

CRUD 操作通常会集中在一个服务模块中。我们可以创建一个名为 crud.py 的文件来实现这些功能。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
# crud.py
from sqlalchemy.orm import Session
from models import User

# 创建用户
def create_user(db: Session, name: str, email: str):
    db_user = User(name=name, email=email)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

# 获取用户
def get_user(db: Session, user_id: int):
    return db.query(User).filter(User.id == user_id).first()

# 获取所有用户
def get_users(db: Session, skip: int = 0, limit: int = 10):
    return db.query(User).offset(skip).limit(limit).all()

# 更新用户
def update_user(db: Session, user_id: int, name: str, email: str):
    user = db.query(User).filter(User.id == user_id).first()
    if user:
        user.name = name
        user.email = email
        db.commit()
        db.refresh(user)
    return user

# 删除用户
def delete_user(db: Session, user_id: int):
    user = db.query(User).filter(User.id == user_id).first()
    if user:
        db.delete(user)
        db.commit()
    return user

主应用程序

现在，让我们将这些操作与 FastAPI 结合。在 main.py 文件中，我们将定义相应的 API 端点。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
# main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from database import SessionLocal, engine, Base
import crud
from models import User

# 创建所有表
Base.metadata.create_all(bind=engine)

app = FastAPI()

# 依赖项：获取数据库会话
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/users/", response_model=User)
def create_user(name: str, email: str, db: Session = Depends(get_db)):
    return crud.create_user(db=db, name=name, email=email)

@app.get("/users/{user_id}", response_model=User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db=db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

@app.get("/users/", response_model=list[User])
def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    return crud.get_users(db=db, skip=skip, limit=limit)

@app.put("/users/{user_id}", response_model=User)
def update_user(user_id: int, name: str, email: str, db: Session = Depends(get_db)):
    user = crud.update_user(db=db, user_id=user_id, name=name, email=email)
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.delete("/users/{user_id}", response_model=User)
def delete_user(user_id: int, db: Session = Depends(get_db)):
    user = crud.delete_user(db=db, user_id=user_id)
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

总结

在本篇文章中，我们实现了基本的 CRUD 操作来管理用户数据。我们定义了数据库模型、CRUD 操作以及如何使用 FastAPI 在 API 端点中调用这些操作。这样，你就能够通过 HTTP 请求进行数据的创建、读取、更新和删除。

下一篇将讨论如何实现 异步数据库操作，以提高应用的性能。请继续关注。

分享转发

发表更新软件开发 / FastAPI教程6 分钟读完 (大约969个字)

30 使用数据库的异步数据库操作

在上一篇教程中，我们探讨了使用数据库的基本CRUD操作。这一篇，我们将深入了解如何在FastAPI中实现异步数据库操作。使用异步机制的优势在于，它能够在IO密集型任务（如数据库访问）中提高性能，使应用更具响应性。

环境准备

在开始之前，请确保已经安装了以下库：

1
pip install fastapi databases asyncpg uvicorn
  • databases 是一个支持异步操作的数据库库。
  • asyncpg 是一个异步PostgreSQL数据库适配器。
  • uvicorn 是一个用于运行FastAPI的ASGI服务器。

创建数据库模型

我们将使用SQLAlchemy定义数据模型，然后通过asyncpg连接到PostgreSQL。以下是一个简单的用户模型示例：

1
2
3
4
5
6
7
8
9
10
11
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    email = Column(String, unique=True, index=True)

配置异步数据库连接

接下来，我们需要配置异步数据库连接。使用databases库，我们可以轻松地定义连接字符串，并且支持异步操作。

1
2
3
4
from databases import Database

DATABASE_URL = "postgresql+asyncpg://username:password@localhost/dbname"
database = Database(DATABASE_URL)

请根据您的数据库配置替换 username、password 和 dbname。

创建和破坏数据库连接

在FastAPI应用程序中，我们需要在启动时连接到数据库，并在关闭时断开连接。可以使用 FastAPI 的事件处理程序实现：

1
2
3
4
5
6
7
8
9
10
11
from fastapi import FastAPI

app = FastAPI()

@app.on_event("startup")
async def startup():
    await database.connect()

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

实现异步CRUD操作

现在我们已经连接了数据库，让我们来实现异步的CRUD操作。我们将创建一些异步函数来处理用户的插入、查询、更新和删除。

创建用户

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from pydantic import BaseModel
from fastapi import HTTPException

class UserIn(BaseModel):
    name: str
    email: str

class UserOut(BaseModel):
    id: int
    name: str
    email: str

@app.post("/users/", response_model=UserOut)
async def create_user(user: UserIn):
    query = User.__table__.insert().values(name=user.name, email=user.email)
    last_record_id = await database.execute(query)
    return {**user.dict(), "id": last_record_id}

查询所有用户

1
2
3
4
@app.get("/users/", response_model=list[UserOut])
async def read_users():
    query = User.__table__.select()
    return await database.fetch_all(query)

查询单个用户

1
2
3
4
5
6
7
@app.get("/users/{user_id}", response_model=UserOut)
async def read_user(user_id: int):
    query = User.__table__.select().where(User.id == user_id)
    user = await database.fetch_one(query)
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

更新用户

1
2
3
4
5
@app.put("/users/{user_id}", response_model=UserOut)
async def update_user(user_id: int, user: UserIn):
    query = User.__table__.update().where(User.id == user_id).values(name=user.name, email=user.email)
    await database.execute(query)
    return {**user.dict(), "id": user_id}

删除用户

1
2
3
4
5
@app.delete("/users/{user_id}", response_model=dict)
async def delete_user(user_id: int):
    query = User.__table__.delete().where(User.id == user_id)
    await database.execute(query)
    return {"message": "User deleted successfully"}

测试异步数据库操作

为了确保我们实现的异步数据库操作有效，建议使用FastAPI的测试客户端进行测试。在下一篇教程中，我们将讨论如何编写单元测试，以验证我们创建的API是否按预期工作。

小结

在本篇教程中，我们讲解了如何在FastAPI中实现异步数据库操作。通过使用databases库，我们能够高效地管理数据库连接并执行异步的CRUD操作。接下来，我们将进入测试阶段，确保我们的API在生产环境中表现良好，这将是我们系列教程的一部分。

后续我们将探讨如何编写单元测试，以确保我们的代码的稳健性。希望这篇教程对您理解FastAPI的异步数据库操作有所帮助！

分享转发

发表更新软件开发 / FastAPI教程7 分钟读完 (大约1020个字)

31 编写测试之编写单元测试

在上一篇文章中，我们讨论了如何使用异步操作与数据库进行交互。在本篇中，我们将专注于如何为我们的FastAPI应用编写单元测试。单元测试是验证我们应用程序单个组件是否按预期工作的重要步骤。在编写单元测试时，我们通常使用unittest模块或第三方测试框架pytest。在这篇文章中，我们将重点介绍如何使用pytest编写简单的单元测试。

1. FastAPI应用概述

在开始之前，我们先回顾一下我们在上一篇文章中创建的FastAPI应用。假设我们有一个简单的用户API，能进行CRUD操作。我们的应用包含一个用户模型和相应的路由：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List

app = FastAPI()

class User(BaseModel):
    id: int
    name: str
    email: str

# 模拟数据库
fake_users_db = []

@app.post("/users/", response_model=User)
async def create_user(user: User):
    fake_users_db.append(user)
    return user

@app.get("/users/", response_model=List[User])
async def read_users():
    return fake_users_db

这个简单的应用可以创建和读取用户。接下来，我们将为这些功能编写相应的单元测试。

2. 单元测试的基本结构

单元测试通常分为三个阶段：

  1. 准备阶段：设置测试所需的数据和环境。
  2. 执行阶段：执行被测试的模块或函数。
  3. 断言阶段：验证结果是否符合预期。

2.1. 安装 pytest

如果你还没有安装pytest，可以通过以下命令进行安装：

1
pip install pytest httpx

2.2. 编写测试用例

我们将在新的文件中创建测试用例，例如：test_main.py。下面是为我们的用户API编写测试的示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import pytest
from fastapi.testclient import TestClient
from main import app, User

client = TestClient(app)

def test_create_user():
    user_data = {"id": 1, "name": "Alice", "email": "alice@example.com"}
    response = client.post("/users/", json=user_data)
    
    assert response.status_code == 200
    assert response.json() == user_data

def test_read_users():
    response = client.get("/users/")
    
    assert response.status_code == 200
    assert response.json() == [{"id": 1, "name": "Alice", "email": "alice@example.com"}]

在上面的例子中：

  • 我们使用 TestClient 来模拟 HTTP 请求。
  • test_create_user 测试了创建用户的功能。
  • test_read_users 测试了获取用户列表的功能。

3. 运行测试

要运行测试，只需在终端中导航到包含 test_main.py 文件的目录，然后运行：

1
pytest test_main.py

如果一切正常，你应该会看到类似以下的输出：

1
2
3
4
5
6
================================= test session starts =================================
collected 2 items

test_main.py ..                                                                  [100%]

================================== 2 passed in 0.03s ==================================

4. 进一步扩展

在真实的应用开发中，通常还需要：

  • 处理异常：测试在特定条件下是否合理抛出异常。
  • 边界情况：测试利用边界条件来加固你的测试（比如空输入等）。
  • 数据库测试：在需要连接数据库的情况下，考虑如何模拟数据库或使用测试数据库。

4.1. 异常处理示例

下面的示例展示了如何测试错误处理：

1
2
3
4
5
def test_create_user_without_email():
    user_data = {"id": 2, "name": "Bob"}
    response = client.post("/users/", json=user_data)
    
    assert response.status_code == 422  # Unprocessable Entity

4.2. 边界情况示例

你也可以测试使用不合理的输入：

1
2
3
4
5
def test_create_user_with_invalid_data():
    user_data = {"id": "not-a-number", "name": "Charlie", "email": "charlie@example.com"}
    response = client.post("/users/", json=user_data)

    assert response.status_code == 422  # Unprocessable Entity

5. 小结

在这一章中，我们学习了如何为我们的FastAPI应用编写单元测试，包括创建、读取用户的API。单元测试不只是确保代码的质量，更是确保我们在进行变更时不会引入新的错误。

在下一篇文章中，我们将进一步探讨如何使用 pytest 进行更深入的测试，涉及到更复杂的用例以及如何通过pytest的功能来简化测试过程。希望大家能继续关注！

分享转发

发表更新软件开发 / FastAPI教程6 分钟读完 (大约927个字)

32 使用pytest进行FastAPI测试

在上一篇文章中，我们探讨了如何编写单元测试，以确保我们的FastAPI应用能够按预期的方式运行。在本篇文章中，我们将进一步深入，研究如何使用pytest进行测试。pytest是一个强大的测试框架，能够方便地编写简单和复杂的测试。

安装pytest

在开始之前，我们需要确保安装了pytest和httpx两个依赖。httpx是一个用于发送HTTP请求的库，我们将用它来测试FastAPI应用。

你可以通过以下命令安装这两个库：

1
pip install pytest httpx

创建测试文件

在FastAPI项目的根目录下，通常会创建一个tests文件夹来存放测试文件。下面是一个测试文件的目录结构示例：

1
2
3
4
5
/your-fastapi-project
├── app
│   └── main.py
└── tests
    └── test_main.py

在test_main.py中，我们将编写我们的第一个测试。

编写测试

在FastAPI中，我们可以使用TestClient来自fastapi.testclient进行集成测试。以下是一个简单的示例，展示如何使用pytest和FastAPI测试框架。

假设我们的main.py文件中有一个简单的API端点：

1
2
3
4
5
6
7
8
9
10
11
12
# app/main.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def read_root():
    return {"Hello": "World"}

@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

我们将为这个API编写测试。

在test_main.py中，代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# tests/test_main.py
import pytest
from httpx import AsyncClient
from fastapi import FastAPI
from app.main import app  # 导入FastAPI应用实例

@pytest.mark.asyncio
async def test_read_root():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.get("/")
    assert response.status_code == 200
    assert response.json() == {"Hello": "World"}

@pytest.mark.asyncio
async def test_read_item():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.get("/items/5?q=somequery")
    assert response.status_code == 200
    assert response.json() == {"item_id": 5, "q": "somequery"}

解释测试代码

  1. 使用pytest.mark.asyncio: 由于我们的FastAPI端点是异步的，我们需要使用@pytest.mark.asyncio修饰器来标记测试为异步测试。

  2. 使用AsyncClient: 我们用AsyncClient来发起HTTP请求，这个客户端会在事件循环中运行，非常适合测试异步应用。

  3. 使用assert语句: 我们使用assert来验证返回的状态码和响应内容是否符合预期。

运行测试

要运行测试，只需在项目的根目录执行以下命令：

1
pytest

这将会自动发现所有以test_开头的函数并运行它们。

测试覆盖率

在项目中，查看测试覆盖率可以帮助我们了解哪些代码已经得到测试。可以通过安装pytest-cov来实现。

安装命令：

1
pip install pytest-cov

然后，你可以使用以下命令在运行测试时查看测试覆盖率：

1
pytest --cov=app tests/

这条命令将显示在app文件夹下代码的测试覆盖率。

总结

在本篇文章中，我们学习了如何使用pytest和httpx来测试FastAPI应用。我们编写了基本的集成测试，以验证API端点的行为。测试是保证代码稳定性和可维护性的重要手段，希望你能在你的FastAPI项目中积极使用测试。

在下一篇文章中，我们将深入探讨如何使用数据库进行测试，确保我们的数据交互部分也是稳定和可靠的。希望你继续关注接下来的内容。

分享转发

发表更新软件开发 / FastAPI教程7 分钟读完 (大约1110个字)

33 使用测试数据库

在上一篇中，我们介绍了如何使用 pytest 进行 FastAPI 应用的测试。在进行测试时，尤其是涉及到数据库的功能时，我们需要一个快速且可控的方式来测试数据库交互。为此，我们将创建一个 测试数据库。

什么是测试数据库？

测试数据库是专门用于测试目的的数据库实例，它通常与生产环境和开发环境数据库分开。通过使用测试数据库，我们可以：

  • 避免对生产数据造成影响。
  • 使测试结果独立于开发和生产环境。
  • 更容易地创建和销毁测试数据。

在本教程中，我们将使用 SQLite 作为我们的测试数据库，因为它简单且易于设置。接下来我们将展示如何将其集成到我们的 FastAPI 测试中。

设置测试数据库

首先，我们需要安装必要的依赖包。确保你的 requirements.txt 中包含以下行：

1
2
3
4
5
fastapi
pytest
httpx
sqlalchemy
databases

然后，我们可以创建一个 conftest.py 文件，来配置测试数据库。

conftest.py 示例

在项目根目录下，创建一个名为 conftest.py 的文件：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
import pytest
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from app.models import Base
from app.main import app  # 确保导入你的FastAPI应用实例
import databases

# 创建一个测试数据库连接
DATABASE_URL = "sqlite+aiosqlite:///./test.db"  # 使用SQLite作为测试数据库
engine = create_async_engine(DATABASE_URL, echo=True)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

@pytest.fixture(scope="session")
async def setup_database():
    # 创建数据库表
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    yield

    # 在测试结束后，销毁数据库
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

@pytest.fixture
async def db_session(setup_database):
    # 创建一个新的数据库会话
    async_session: AsyncSession = TestingSessionLocal()
    yield async_session
    await async_session.close()

这里，我们使用了 pytest 的 fixture 来设置和清理测试数据库。setup_database fixture 会在测试会话开始时创建所有数据库表，并在测试完成后删除它们。

使用测试数据库进行测试

接下来，我们需要编写测试用例，以确保我们的 FastAPI 应用与数据库交互正常。在此例中，我们将测试一个简单的 CRUD 操作。

假设的模型和API

假设我们有一个 User 模型和相应的 FastAPI 路由。首先，我们定义模型：

1
2
3
4
5
6
7
8
9
10
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)

然后，我们在 FastAPI 中定义 API 路由：

1
2
3
4
5
6
7
8
9
10
11
12
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from app.database import get_db  # 假设有一个获取DB的依赖函数

app = FastAPI()

@app.post("/users/", response_model=User)
async def create_user(user: User, db: Session = Depends(get_db)):
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

编写测试用例

现在，我们来编写一个简单的测试，来验证我们的 create_user 路由。

在你的测试文件中（比如 test_api.py），添加以下内容：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import pytest
from httpx import AsyncClient
from app.main import app
from app.models import User

@pytest.mark.asyncio
async def test_create_user(db_session):
    async with AsyncClient(app=app, base_url="http://test") as client:
        user_data = {"name": "Test User"}

        response = await client.post("/users/", json=user_data)

        assert response.status_code == 200
        assert response.json()["name"] == user_data["name"]

        # 验证数据库中是否插入了该用户
        result = await db_session.execute(select(User).where(User.name == "Test User"))
        user = result.scalars().first()
        assert user is not None
        assert user.name == "Test User"

在这个测试中，我们首先使用 AsyncClient 发送一个 POST 请求来创建用户。然后，我们验证响应的 status_code 和返回的数据。最后，我们通过与数据库会话的交互，检查用户是否实际被插入到测试数据库中。

总结

通过使用一个独立的测试数据库，我们可以确保我们的 FastAPI 应用在运行测试时不会影响生产数据。这种方法也提高了测试的可靠性和可维护性。接下来，在下一篇教程中，我们将讨论如何 选择 FastAPI 应用的部署方式。保持关注！

分享转发

发表更新软件开发 / FastAPI教程7 分钟读完 (大约1036个字)

34 FastAPI应用部署之选择部署方式

在上一篇教程中，我们探讨了如何编写测试以及如何使用测试数据库。这一节中，我们将着重于 FastAPI 应用的部署方式选择。正确的部署方式不仅可以提高应用的性能，还能增强其安全性和可扩展性。我们将讨论几种流行的部署方式，以及在不同场景下的适用性。

1. 部署方式概述

在选择适合的部署方式之前，我们需要考虑以下几个方面：

  • 应用规模：您的应用是小型项目还是大型企业级应用？
  • 流量需求：您预期的流量有多大？是否需要高可用性？
  • 开发和运维经验：您或您的团队对不同技术栈的熟悉程度如何？

1.1 常见部署方式

以下是几种常见的 FastAPI 部署方式：

  1. 使用 ASGI 服务器

    • 如 Uvicorn 或 Daphne，直接将应用部署到服务器上。
    • 适合小型应用或开发阶段。

  2. 使用应用程序服务器

    • 如 Gunicorn 配合 Uvicorn。
    • 适合中型到大型的生产应用，提供更好的性能和并发控制。

  3. 容器化部署

    • 使用 Docker 将应用打包，便于在各类环境中部署。
    • 适合需要多个环境一致性的团队开发和运维。

  4. 云平台部署

    • 使用 AWS、Google Cloud、Azure 等云服务提供商的服务。
    • 适合需要可扩展性和管理便利的应用。

  5. 无服务器架构

    • 如使用 AWS Lambda 配合 API Gateway。
    • 适合流量波动较大的应用，可按需扩展。

2. 使用 ASGI 服务器

对于小型应用或开发测试阶段，您可以简单地使用 Uvicorn 来部署 FastAPI 应用。下面是一个基本的示例：

1
2
# 使用 Uvicorn 启动 FastAPI 应用
uvicorn main:app --host 0.0.0.0 --port 8000

在这个命令中，main 是包含 FastAPI 实例的 Python 文件名，app 是 FastAPI 实例。通过上面的命令，您可以快速启动应用，并在本地访问 http://127.0.0.1:8000。

3. 使用 Gunicorn 和 Uvicorn

当您的应用开始成长，流量增加时，推荐使用 Gunicorn 作为进程管理器，与 Uvicorn 结合进行部署。以下是部署示例：

1
2
3
4
5
# 安装 Gunicorn 和 Uvicorn
pip install gunicorn uvicorn

# 使用 Gunicorn 启动应用（4个工作进程）
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

这里的 -w 4 表示启动 4 个工作进程，-k uvicorn.workers.UvicornWorker 指定了使用的工作类，允许 Uvicorn 作为工作进程来处理请求。

4. 容器化部署

容器化部署是现代应用程序部署的趋势。Docker 可以帮助您更轻松地管理应用的环境和依赖。以下是容器化 FastAPI 应用的步骤：

  1. 编写 Dockerfile
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# 使用适合的基础镜像
FROM python:3.9-slim

# 设置工作目录
WORKDIR /app

# 复制依赖文件
COPY requirements.txt .

# 安装依赖
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用代码
COPY . .

# 暴露端口
EXPOSE 8000

# 启动应用
CMD ["gunicorn", "main:app", "-w", "4", "-k", "uvicorn.workers.UvicornWorker", "--bind", "0.0.0.0:8000"]
  1. 构建和运行 Docker 容器
1
2
3
4
5
# 构建 Docker 镜像
docker build -t fastapi-app .

# 运行容器
docker run -d -p 8000:8000 fastapi-app

这个示例展示了如何创建一个 Docker 镜像并以容器方式运行 FastAPI 应用。通过这种方式，您可以确保在开发、测试和生产环境中运行的应用是相同的。

5. 总结

在选择 FastAPI 应用的部署方式时，您需要考虑多个因素，包括应用的规模、流量需求和团队的技能水平。无论是简单的 ASGI 服务器部署，还是容器化和云平台的复杂部署，每种方式都有其适用场景。

在下一篇教程中，我们将更详细地探讨如何使用 Docker 部署我们的 FastAPI 应用。希望本篇对您选择适合的部署方式有所帮助！

分享转发

发表更新软件开发 / FastAPI教程6 分钟读完 (大约885个字)

35 部署FastAPI应用之使用Docker部署

在上一篇文章中，我们讨论了如何选择适合的部署方式来托管我们的FastAPI应用。在这篇文章中，我们将深入探讨如何使用Docker来部署FastAPI应用。Docker是一种流行的容器化工具，可以让你的应用在任何环境中以一致的方式运行。

了解Docker

Docker使开发者能够将应用打包到一个轻量级的容器中，这个容器包含了应用运行所需的所有依赖项和环境配置。这样，在不同的机器上部署应用时，可以确保应用的行为一致。

为FastAPI创建Dockerfile

首先，我们需要在FastAPI应用的根目录下创建一个Dockerfile。Dockerfile是一个文本文件，其中包含了构建Docker镜像的所有指令。以下是一个示例Dockerfile：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 使用Python官方镜像
FROM python:3.9-slim

# 设置工作目录
WORKDIR /app

# 复制依赖文件到工作目录
COPY requirements.txt .

# 安装所需的依赖
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用代码到工作目录
COPY . .

# 启动FastAPI应用
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

在以上示例中，我们首先选择了一个基础镜像python:3.9-slim，并将工作目录设置为/app。接着，我们复制了requirements.txt文件（里面列出了所有的Python依赖）到容器中，并安装这些依赖。最后，我们将FastAPI应用的代码复制到容器，并使用uvicorn启动应用。

创建requirements.txt

确保你有一个requirements.txt文件，它列出了FastAPI和其他所需库的版本。例如：

1
2
fastapi==0.95.0
uvicorn==0.18.2

构建Docker镜像

在终端中，导航到应用的根目录，然后运行以下命令来构建Docker镜像：

1
docker build -t myfastapiapp .

这里的myfastapiapp是你为镜像指定的名称。

运行Docker容器

构建完成后，我们可以运行Docker容器。使用以下命令启动容器并映射端口，以便能够在浏览器中访问FastAPI应用：

1
docker run -d --name fastapi-container -p 8000:8000 myfastapiapp

在这个命令中，-d参数表示在后台运行容器，--name指定了容器的名称，-p参数映射了主机的8000端口到容器的8000端口。

测试FastAPI应用

现在，容器正在运行，你可以通过浏览器或API客户端访问你的FastAPI应用：

1
http://localhost:8000/docs

在这里，你可以看到FastAPI自动生成的API文档。

管理Docker容器

使用以下命令可以查看正在运行的容器：

1
docker ps

如果需要停止容器，可以使用：

1
docker stop fastapi-container

如果要删除容器，可以使用：

1
docker rm fastapi-container

结语

通过Docker部署FastAPI应用使得我们的应用能够在任何支持Docker的平台上轻松运行，并且提供了一种简单的一致的方式来管理依赖。在接下来的文章中，我们将讨论如何在云平台上进一步部署我们的FastAPI应用。这样，我们将能够充分利用云服务的优势，进行更大规模的应用托管和负载均衡。

希望这篇文章对你使用Docker部署FastAPI应用有所帮助。如有问题，欢迎在评论区提问！

分享转发

发表更新软件开发 / FastAPI教程6 分钟读完 (大约951个字)

36 在云平台上部署FastAPI应用

在上一篇中，我们探讨了如何使用Docker来部署FastAPI应用。这种容器化的方法非常灵活，但如果你希望简化部署流程，或者希望利用云服务提供的可扩展性与高可用性，云平台部署将是一个极佳的选择。在本篇中，我们将详细讨论如何在云平台上部署FastAPI应用。

选择云平台

常见的云平台有很多，包括：

  • AWS（Amazon Web Services）
  • Google Cloud Platform（GCP）
  • Microsoft Azure
  • Heroku（适合快速部署）
  • DigitalOcean

对于本教程，我们将以 Heroku 为例，它非常友好且适合小型应用的快速部署。

第一步：准备FastAPI应用

假设我们之前已经有一个简单的FastAPI应用，代码如下：

1
2
3
4
5
6
7
8
9
10
11
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

将以上代码保存为 main.py 文件。

第二步：创建依赖描述文件

在Heroku上运行Python应用需要一个依赖文件，通常为 requirements.txt 文件。你可以使用以下命令自动生成：

1
pip freeze > requirements.txt

确保 requirements.txt 文件中包含 fastapi 和 uvicorn 这两个必要的库。例如：

1
2
fastapi==0.75.0
uvicorn==0.17.0

此外，我们需要定义一个 Procfile 文件，以告诉Heroku如何启动应用。创建一个名为 Procfile 的文件，内容如下：

1
web: uvicorn main:app --host 0.0.0.0 --port ${PORT:-5000}

在上面的命令中，main 是包含 FastAPI 应用的 Python 文件名（不带 .py 后缀），app 是 FastAPI 应用实例的名字。

第三步：创建Heroku账号并安装Heroku CLI

访问 Heroku官网 注册一个账户，并按以下步骤安装Heroku CLI：

  1. 下载并安装Heroku CLI：Heroku CLI下载链接
  2. 安装后，使用以下命令登录：
1
heroku login

第四步：创建Heroku应用

通过以下命令在 Heroku 上创建一个新的应用：

1
heroku create my-fastapi-app

my-fastapi-app 可以替换为你想要的任何应用名称，全局唯一。

第五步：部署FastAPI应用

接下来，我们将代码提交到Heroku。根据以下步骤操作：

  1. 初始化Git仓库（如果还未初始化）：
1
git init
  1. 添加文件并提交：
1
2
git add .
git commit -m "Initial commit"
  1. 将Heroku远程仓库添加到本地Git：
1
heroku git:remote -a my-fastapi-app
  1. 推送代码到Heroku：
1
git push heroku master

在推送过程中，Heroku会自动检测到你的 requirements.txt 和 Procfile，并会安装依赖。

第六步：访问你的应用

一旦部署完成，Heroku会提供应用的URL，可以通过以下命令获取：

1
heroku open

在浏览器中打开后，你应该能看到返回的 JSON 响应如：

1
{"Hello": "World"}

你也可以通过访问 /items/1?q=foo 来测试不同的API路由。

第七步：监控与日志

你可以通过以下命令查看Heroku上的实时日志，以监控应用的运行状况：

1
heroku logs --tail

小结

在本教程中，我们介绍了如何在云平台（以Heroku为例）上部署FastAPI应用。我们从创建FastAPI应用开始，经过创建必要的依赖文件，最终实现应用的部署和访问。这一过程展示了在云环境中快速上线应用的便利性，未来我们将讨论如何在其他云平台（如AWS、GCP等）上部署FastAPI应用。

分享转发

×