加载搜索数据…
= self.max_requests: return JSONResponse( status_code=429, content={"detail": "请求过于频繁,请稍后再试"}, headers={"Retry-After": str(self.window)}, ) self.buckets[client_ip].append(now) return await call_next(request) ``` 注册使用: ```python app.add_middleware(RateLimitMiddleware, max_requests=20, window=1) ``` ### 纯 ASGI 中间件:更高粒度控制 如果需要完全控制 ASGI 事件(如处理 WebSocket),可以直接写纯 ASGI 中间件: ```python from starlette.types import ASGIApp, Scope, Receive, Send class CustomASGIMiddleware: def __init__(self, app: ASGIApp): self.app = app async def __call__(self, scope: Scope, receive: Receive, send: Send): if scope["type"] == "http": # 仅在 HTTP 请求时干预 print(f"ASGI 中间件:{scope['method']} {scope['path']}") await self.app(scope, receive, send) ``` ## 中间件的顺序问题 中间件的执行顺序与 `add_middleware` 的调用顺序**相同**——先添加的在外层,先拦截请求、最后放行响应。 ```python # 洋葱模型:A 在最外层 app.add_middleware(MiddlewareA) # 第 1 层(最外层) app.add_middleware(MiddlewareB) # 第 2 层 app.add_middleware(MiddlewareC) # 第 3 层(最内层,最靠近路由) ``` 请求链路:`A → B → C → 路由`,响应链路:`路由 → C → B → A`。 **最佳实践排序**: 1. **最外层**:日志、耗时统计(尽量覆盖所有请求) 2. **次外层**:CORS、安全头 3. **次内层**:认证鉴权 4. **最内层**:请求数据预处理、响应格式化 ## 实际场景:响应头注入 一个常见的需求是为所有响应统一注入安全头: ```python @app.middleware("http") async def security_headers_middleware(request: Request, call_next): response = await call_next(request) response.headers["X-Content-Type-Options"] = "nosniff" response.headers["X-Frame-Options"] = "DENY" response.headers["X-XSS-Protection"] = "1; mode=block" response.headers["Strict-Transport-Security"] = "max-age=31536000" return response ``` ## 中间件 vs 依赖注入 很多初学者会困惑:都是请求前执行的逻辑,中间件和 Depends 依赖注入有何区别? | 维度 | 中间件 | Depends 依赖注入 | |------|--------|-----------------| | **作用范围** | 全局,覆盖所有路由 | 可按路由、路由分组、全局配置 | | **执行时机** | 在路由匹配之前 | 路由匹配之后 | | **请求信息** | 仅有 Request 对象 | 有解析后的路径/查询/请求体参数 | | **修改响应** | 可以直接修改 Response | 不能直接修改 Response | | **适用场景** | 日志、CORS、安全头、限流 | 鉴权、数据库会话、参数校验 | 一句话总结:**影响"所有请求"用中间件,影响"特定业务逻辑"用 Depends**。 ## 小结 FastAPI 的中间件系统继承自 Starlette,简洁而强大。日常开发中: - **优先级最高**的中间件是 CORS,前后端分离几乎必配 - 函数式 `@app.middleware("http")` 覆盖 80% 的场景 - 需要参数化或复杂逻辑时,使用类式 `BaseHTTPMiddleware` - 牢记洋葱模型,合理安排中间件添加顺序 掌握中间件,你就掌握了 FastAPI 请求生命周期的"总开关"。 ">

FastAPI 中间件开发详解

详解 FastAPI 中间件机制,涵盖内置中间件(CORS、HTTPSRedirect、TrustedHost)、自定义中间件的编写、请求/响应拦截、中间件执行顺序、以及实际应用场景(请求日志、耗时统计、请求限流)。

同一个请求内,对同一个依赖 FastAPI 默认会**缓存**结果。也就是说,若一个请求中多个地方依赖了同一个函数,它只会被执行一次,后续直接返回缓存值。 --- ## 类作为依赖 除了函数,**任何可调用对象**(类、实例)都可以作为依赖。用类来定义依赖,结构更清晰: ```python class CommonQueryParams: def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100): self.q = q self.skip = skip self.limit = limit @app.get("/items/") def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)): return {"q": commons.q, "skip": commons.skip, "limit": commons.limit} ``` `Depends(CommonQueryParams)` 中传的参数是类本身(可调用),FastAPI 会实例化它,并把构造函数参数识别为查询参数。 ### 简写形式 当依赖「类型」和「可调用对象」相同时,可以简写: ```python # 这三种写法等价 def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)): ... def read_items(commons: CommonQueryParams = Depends()): ... def read_items(commons: CommonQueryParams = Depends()): ... ``` `Depends()` 不传参时,FastAPI 会自动用参数的类型注解(这里是 `CommonQueryParams`)作为依赖。 --- ## 子依赖(嵌套依赖) 依赖可以层层嵌套,FastAPI 会按顺序解析整个依赖树: ```python def query_extractor(q: str | None = None): return q def query_checker(q: str = Depends(query_extractor)): if q == "admin": raise ValueError("禁止使用保留关键字") return q @app.get("/search/") def search(result: str = Depends(query_checker)): return {"q": result} ``` 执行顺序:`query_extractor` → `query_checker` → `search`。如果某一层依赖抛出异常,请求会立即中断并返回对应错误。 --- ## 全局依赖与路由组依赖 有些依赖(如鉴权、CORS 校验)需要应用到整个应用或一组路由上,无需在每个函数里写 `Depends`。 ### 应用级依赖 ```python async def verify_token(x_token: str = Header(...)): if x_token != "secret-token": raise HTTPException(status_code=400, detail="X-Token 错误") app = FastAPI(dependencies=[Depends(verify_token)]) ``` 这样**所有路由**都会先执行 `verify_token`,但它的返回值不会传入路由函数。 ### APIRouter 级依赖 ```python from fastapi import APIRouter admin_router = APIRouter(dependencies=[Depends(verify_token)]) @admin_router.get("/dashboard/") def dashboard(): return {"data": "仅管理员可见"} app.include_router(admin_router, prefix="/admin") ``` ### 单个路由的专属依赖 ```python @app.get("/items/", dependencies=[Depends(verify_token)]) def read_items(): return {"items": []} ``` `dependencies` 列表中的依赖会执行,但结果不注入函数——适合用于「拦截 / 校验」类需求。 --- ## 使用 yield 的依赖(资源管理) 类似上下文管理器,依赖函数可以使用 `yield` 来在请求结束后执行清理逻辑: ```python def get_db(): db = SessionLocal() try: yield db finally: db.close() @app.get("/users/") def list_users(db = Depends(get_db)): # 使用 db 查询 return db.execute("SELECT * FROM users").fetchall() ``` 执行流程: 1. `get_db` 执行到 `yield db`,把 `db` 注入路由函数; 2. 路由函数执行完毕并返回响应; 3. FastAPI 继续执行 `yield` 之后的 `finally` 代码,关闭连接。 这种方式非常适合管理数据库会话、Redis 连接、文件句柄等资源。也可以配合 `with` 语句使用: ```python def get_file(): with open("data.txt") as f: yield f ``` --- ## 实战示例:分页参数 + 鉴权 把依赖注入用到真实业务中。下面演示「分页依赖」和「当前用户依赖」的组合: ```python from fastapi import FastAPI, Depends, HTTPException, Header app = FastAPI() class Pagination: def __init__(self, page: int = 1, size: int = 10): self.page = page self.size = size def get_current_user(authorization: str = Header(...)): if not authorization.startswith("Bearer "): raise HTTPException(status_code=401, detail="无效的认证信息") token = authorization[7:] user = verify_jwt(token) # 假设的解密函数 if not user: raise HTTPException(status_code=401, detail="用户不存在") return user @app.get("/posts/") def list_posts( pager: Pagination = Depends(Pagination), current_user: dict = Depends(get_current_user), ): return { "page": pager.page, "size": pager.size, "user": current_user["username"], } ``` 一个路由函数同时依赖两个组件:`Pagination` 负责解析分页参数,`get_current_user` 负责鉴权。两者互不耦合,各自独立可测试、可复用。 --- ## 小结 | 用法 | 关键写法 | 适用场景 | |------|---------|---------| | 函数依赖 | `Depends(func)` | 抽取公共参数 / 逻辑 | | 类依赖 | `Depends(Class)` 或 `Depends()` | 结构化参数对象 | | 子依赖 | 依赖内部再调用 `Depends` | 多层校验、拆分逻辑 | | 全局依赖 | `FastAPI(dependencies=[...])` | 全局拦截(鉴权、日志) | | 路由组依赖 | `APIRouter(dependencies=[...])` | 一组接口共享前置逻辑 | | yield 依赖 | 函数中 `yield` | 数据库会话等资源管理 | 依赖注入是 FastAPI 区别于其他框架的核心特性之一。善用 `Depends`,可以让接口代码保持「只关心业务本身」,把参数解析、权限校验、资源管理等工作交给可复用的依赖组件,显著提升代码的可维护性和可测试性。 ">

FastAPI 依赖注入(Depends)详解

详解 FastAPI 中的依赖注入机制,包括 Depends 基本用法、类作为依赖、嵌套依赖、全局依赖、yield 依赖与资源释放、路径操作专属依赖等核心知识点。

str: if not v.isalnum(): raise ValueError('用户名只能包含字母和数字') return v @validator('confirm_password') def passwords_match(cls, v: str, values: dict) -> str: if 'password' in values and v != values['password']: raise ValueError('两次输入的密码不一致') return v ``` `values` 参数包含前面已验证过的字段值,因此可以在 `confirm_password` 的验证器中拿到 `password` 进行比较。 > **注意**:Pydantic V2 中 `@validator` 已改为 `@field_validator`,用法类似但参数名略有不同。 ## 结合路径参数与查询参数 请求体、路径参数、查询参数可以同时存在,FastAPI 根据参数声明自动区分: ```python from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class ItemUpdate(BaseModel): name: str = Field(..., min_length=1) price: float = Field(..., gt=0) is_offer: bool = False @app.put("/items/{item_id}") async def update_item( item_id: int, # 路径参数 item: ItemUpdate, # 请求体(Pydantic 模型 = body) force: bool = False, # 查询参数 ): return { "item_id": item_id, "force": force, "update": item.dict() } ``` 识别规则很简单: - 在路径中也声明的 → **路径参数** - 是 Pydantic 模型类型的 → **请求体** - 是简单类型(str/int/float/bool)且不在路径中 → **查询参数** ## 请求体中的额外字段 默认情况下,如果客户端多传了模型中未定义的字段,Pydantic V1 会忽略它们。如果需要严格模式(拒绝多余字段),可以在模型内部配置: ```python from pydantic import BaseModel class StrictItem(BaseModel): name: str price: float class Config: extra = "forbid" # 禁止额外字段 ``` 此时多传字段会直接返回 422 错误。 ## 小结 FastAPI 结合 Pydantic 的请求体处理机制让数据校验变得极其简洁: - 用 `BaseModel` 定义数据结构,类型即校验规则 - 用 `Field` 添加约束(长度、范围、正则) - 用嵌套模型处理复杂 JSON 结构 - 用 `@validator` 处理跨字段校验和自定义逻辑 这套机制让开发者无需手写校验代码,同时自动产出清晰的 API 文档,显著提升开发效率和数据质量。 ">

FastAPI 请求体与 Pydantic 模型校验

详解 FastAPI 中请求体的使用方式,Pydantic 模型的字段类型、校验规则、嵌套模型、自定义验证器以及结合路径参数和查询参数的完整实战。

= 1 le=10000, # <= 10000 ) ): return {"article_id": article_id} ``` `Path` 常用校验参数: | 参数 | 说明 | |------|------| | `...` | 必填(省略号) | | `ge` / `gt` | 大于等于 / 大于 | | `le` / `lt` | 小于等于 / 小于 | | `min_length` / `max_length` | 字符串最小/最大长度 | | `regex` | 正则匹配(FastAPI 旧版) | | `pattern` | 正则匹配(Pydantic v2) | | `title` / `description` | 文档展示用 | ### 枚举类型的路径参数 ```python from enum import Enum class CategoryEnum(str, Enum): tech = "tech" life = "life" travel = "travel" @app.get("/posts/{category}") async def get_posts(category: CategoryEnum): # 自动生成文档,且只接受这三个值 return {"category": category} ``` 访问 `/posts/tech` 正常返回;访问 `/posts/music` 返回 422。 --- ## 查询参数 查询参数是 URL 中 `?` 后面的键值对,如 `?page=1&size=20`。FastAPI 中,**函数参数不在路径中且不是 `Body` 类型,就会被自动识别为查询参数**。 ### 基本用法 ```python @app.get("/posts") async def list_posts(page: int = 1, size: int = 10): return {"page": page, "size": size} ``` 访问 `/posts?page=2&size=5`,`page=2, size=5`;不传参数则使用默认值。 ### 可选查询参数 ```python from typing import Optional @app.get("/search") async def search( keyword: Optional[str] = None, category: Optional[str] = None, ): result = {"keyword": keyword, "category": category} if keyword: result["match"] = f"搜索: {keyword}" return result ``` `Optional[str] = None` 表示该参数可选,不传时为 `None`。 ### 使用 Query 做额外校验 ```python from fastapi import Query @app.get("/items") async def list_items( offset: int = Query(default=0, ge=0, description="偏移量"), limit: int = Query(default=10, ge=1, le=100, description="每页数量"), sort: Optional[str] = Query(default=None, pattern=r"^(asc|desc)$", description="排序方式"), ): return {"offset": offset, "limit": limit, "sort": sort} ``` `Query` 支持与 `Path` 几乎相同的校验参数。 ### 接收多个同名查询参数 ```python from typing import List @app.get("/filter") async def filter_items(tags: List[str] = Query(default=[])): """ 示例: /filter?tags=python&tags=fastapi&tags=web 结果: tags = ["python", "fastapi", "web"] """ return {"tags": tags} ``` ### 布尔类型查询参数 ```python @app.get("/products") async def list_products(in_stock: bool = False): """ 访问: /products?in_stock=true → True /products?in_stock=1 → True /products?in_stock=yes → True /products?in_stock=on → True /products → False(默认值) """ return {"in_stock": in_stock} ``` --- ## 路径参数与查询参数的组合 最常见的场景是路径参数标识资源,查询参数做筛选: ```python @app.get("/users/{user_id}/posts") async def get_user_posts( user_id: int = Path(..., ge=1), page: int = Query(default=1, ge=1), size: int = Query(default=20, ge=1, le=100), status: Optional[str] = Query(default=None, pattern=r"^(draft|published|archived)$"), ): """ 获取某个用户的文章列表 - **user_id**: 用户ID(路径参数) - **page**: 页码(查询参数,默认第1页) - **size**: 每页数量(查询参数,默认20,最大100) - **status**: 文章状态过滤(可选) """ return { "user_id": user_id, "page": page, "size": size, "status": status, } ``` 请求示例:`/users/42/posts?page=2&size=5&status=published` --- ## 小结 | 特性 | 路径参数 | 查询参数 | |------|---------|---------| | 位置 | URL 路径中 `/users/{id}` | URL `?` 后面 `?key=val` | | 用途 | 标识资源 | 过滤、排序、分页 | | 必填 | 默认必填 | 默认可选 | | 校验工具 | `Path()` | `Query()` | | 类型 | str / int / UUID / Enum | str / int / bool / List 等 | 掌握路径参数和查询参数是 FastAPI 开发的基础,配合 `Path` 和 `Query` 的校验能力,可以在不写额外逻辑的情况下完成大部分参数校验工作,并自动生成清晰的 API 文档。 ">

FastAPI 路径参数与查询参数详解

详解 FastAPI 中的路径参数和查询参数,包括类型校验、默认值、可选参数、枚举约束、参数校验以及两者的组合使用。

环境 - Python版本:Python3.8.9 - 框架:[FastApi](https://fastapi.tiangolo.com/zh/) ```shell pip install fastapi ``` - ORM工具:[tortoise-orm](https://tortoise-orm.readthedocs.io/en/latest/index.html) ```shell pip install tortoise-orm pip install aiomysql ``` - 数据库:MySQL - 迁移工具:[aerich](https://github.com/tortoise/aerich/tree/master) ```shell pip install aerich ``` - 部署工具:[uvicorn](https://www.uvicorn.org/deployment/) ```shell pip install uvicorn ``` - 数据校验:[Pydantic](https://pydantic-docs.helpmanual.io/), FastApi附带安装。 - 异步任务:[celery](https://docs.celeryq.dev/en/stable/reference/celery.html) --- # 程序构建 ## 创建程序所需文件 创建文件夹 `fastapi-app`,然后创建程序所需文件,创建完成后程序目录格式如下。目前目录创建的都是空白文件,后面再写内容。 ``` . ├── main.py ├── middlewares │ ├── __init__.py │ └── auth_middleware.py ├── models │ ├── __init__.py │ └── user.py ├── tasks │ ├── __init__.py │ └── tasks.py └── views ├── __init__.py ├── user.py └── login.py ``` ## 简单启动程序 1. 在`main.py`文件中创建`fastapi`实例。 ```python from fastapi import FastAPI def create_app(): # 创建一个实例 app = FastAPI() return app ``` 2. 更新`main.py`,添加一个路由。 ```python from fastapi import FastAPI def create_app(): # 创建一个实例 app = FastAPI() @app.get('/') def index(): return "Hello World" return app app = create_app() ``` 3. 启动程序:`uvicorn main:app --reload`。 ```shell (venv) ➜ uvicorn main:app --reload INFO: Will watch for changes in these directories: ['/Users/wxy/fastapi-app'] INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit) INFO: Started reloader process [1428] using StatReload INFO: Started server process [1430] INFO: Waiting for application startup. INFO: Application startup complete. ``` 4. 访问`http://127.0.0.1:8000`。 ![在这里插入图片描述](fastapi/hellow.png) 5. 访问`http://127.0.0.1:8000/docs`查看文档。 ![在这里插入图片描述](fastapi/doc.png) ## 程序拓展 作为实例程序,下面会完成一个用户的创建、登录,虽然只有两个接口,但是其中包括了MySQL的配置和连接、orm工具和迁移工具的使用、中间件的使用等。后面写一个完整的程序其实也就是按照这两个接口修修改改,就不过多赘述了。 ### 更新代码 1. 更新`main.py`,配置数据库参数,并且连接数据库。 ```python import uvicorn from fastapi import FastAPI from tortoise.contrib.fastapi import register_tortoise TORTOISE_ORM_CONFIG = { 'connections': { 'default': { 'engine': 'tortoise.backends.mysql', 'credentials': { 'host': 'localhost', 'port': '3306', 'user': 'root', 'password': '12345678', 'database': 'fastapp', } }, }, 'apps': { 'models': { # 数据表对应文件, `aerich.models`是迁移工具生成的数据表 'models': ['aerich.models', 'models'], 'default_connection': 'default', } } } def create_app(): # 创建一个实例 app = FastAPI() # 连接数据库 register_tortoise( app, add_exception_handlers=True, config=TORTOISE_ORM_CONFIG, # 生成模式, 自动创建数据表, generate_schemas=False, ) @app.get('/') def index(): return "Hello World" return app app = create_app() ``` 2. 打开`models/user.py`新增User的model数据。 ```python from tortoise import fields, Model class User(Model): """ 创建user表 """ # pk=True, 设置为主键 id = fields.IntField(pk=True) name = fields.CharField(max_length=64, description="用户名") password = fields.CharField(max_length=128, description="登录密码") create_at = fields.DatetimeField(auto_now_add=True, description="创建时间") modify_at = fields.DatetimeField(auto_now=True, description="更新时间") ``` 3. 把新创建的User表导入到`models/__init__.py`中。 ```python from models.user import User ``` 4. 打开`viewls/user.py`新增创建用户接口的路由。 ```python from fastapi import APIRouter # 创建一个路由 router = APIRouter( # 请求路径 prefix="/user", # 标签, 文档上显示 tags=["登录"], ) ``` 5. 更新`viewls/user.py`文件,添加请求数据的校验和响应数据格式化。 ```python from fastapi import Body from pydantic.main import BaseModel from typing import Optional class UserRequest(BaseModel): """ 创建用户接口数据校验 """ # default=..., 是指name字段为必填项, 不写default参数也是默认为必填, 这里加上只是为了更清晰 name: str = Body(default=..., description="用户名") password: str = Body(description="登录密码") # Optional[str]可选项, default=None可以不填或者是填写None email: Optional[str] = Body(default=None, description="邮箱") class UserResponse(BaseModel): """ 创建用户返回数据格式化 """ name: Optional[str] password: Optional[str] email: Optional[str] class Config: # 设置orm_mode=True, 可以在view层直接返回model实例, 并且关联的外键数据也可以直接查出来 orm_mode = True ``` 6. 更新`viewls/user.py`文件,定义创建用户的接口。 ```python # response_model=UserResponse, 指定响应数据的格式 # response_model_exclude_none=True, 如果返回的字段=None, 不显示该字段 @router.post('/', response_model=UserResponse, response_model_exclude_none=True) async def create_user(data: UserRequest): user = await User.create( name=data.name, passsword=data.password, email=data.email, ) return user ``` 7. 更新`main.py`文件,把上面定义的创建路由添加到程序中,在`create_app`方法最后添加一行`app.include_router(user.router)`。 ```python import uvicorn from fastapi import FastAPI from tortoise.contrib.fastapi import register_tortoise from views import user TORTOISE_ORM_CONFIG = { 'connections': { 'default': { 'engine': 'tortoise.backends.mysql', 'credentials': { 'host': 'localhost', 'port': '3306', 'user': 'root', 'password': '12345678', 'database': 'fastapp', } }, }, 'apps': { 'models': { # 数据表对应文件, `aerich.models`是迁移工具生成的数据表 'models': ['aerich.models', 'models'], 'default_connection': 'default', } } } def create_app(): # 创建一个实例 app = FastAPI() # 连接数据库 register_tortoise( app, add_exception_handlers=True, config=TORTOISE_ORM_CONFIG, # 生成模式, 自动创建数据表, generate_schemas=False, ) @app.get('/') def index(): return "Hello World" # 添加路由 app.include_router(user.router) return app app = create_app() ``` 到了这里创建用户的接口就算是完成了,但是现在还无法真正的运行起来,如果现在直接运行会提示 ``` tortoise.exceptions.ConfigurationError: default_connection for the model cannot be None ``` 这是因为我们创建的User用户表其实还没有在数据库里面创建,需要使用迁移工具更新数据库以后才可以使用。 ## aerich迁移工具的使用 1. 初始化配置文件和迁移位置: ```shell aerich init -t main.TORTOISE_ORM_CONFIG ``` 2. 初始化后在`main.py`同级目录下会生成一个空的`migrations`文件夹,和`pyproject.toml`文件, `pyproject.toml`内容如下: ```ini [tool.aerich] tortoise_orm = "main.TORTOISE_ORM_CONFIG" location = "./migrations" src_folder = "./." ``` 3. 初始化数据库。 ```shell aerich init-db ``` 4. 初始化后会在`migrations`文件夹下生成一个`models`文件夹, 同时生成一份`0_{datetime}_init.sql`数据库迁移文件, 并且在数据库中添加一个名为`aerich`的迁移表。 ```sql -- upgrade -- CREATE TABLE IF NOT EXISTS `aerich` ( `id` INT NOT NULL PRIMARY KEY AUTO_INCREMENT, `version` VARCHAR(255) NOT NULL, `app` VARCHAR(100) NOT NULL, `content` JSON NOT NULL ) CHARACTER SET utf8mb4; ``` 5. 更新数据库迁移文件。如果models有更新,会在`migrations/models`文件夹中新生成一份`1_{datetime}_update.sql`的数据库迁移文件。 ```shell aerich migrate ``` 6. 把迁移文件更新到数据库。 ```shell aerich upgrade ``` 7. (可选) 降级数据库1: 查看历史版本。 ```shell aerich history ``` 8. (可选) 降级数据库2: 降级到指定版本。 ```shell aerich downgrade -v [版本] ``` 9. (可选) 降级数据库3: 查看被降级的版本。 ```shell aerich heads ``` 10. (可选) [迁移文件整合、损坏修复](https://github.com/tortoise/aerich/tree/master#restore-aerich-workflow) ## 通过文档创建用户 1. 启动程序:`uvicorn main:app --reload`。 2. 访问文档地址:`http://127.0.0.1:8000/docs`。 3. 依次点击 ![在这里插入图片描述](fastapi/user.png) 4. 修改请求值(因为`email`字段非必填,所以这里没有填写`email`参数),然后点击执行按钮。 ![在这里插入图片描述](fastapi/body.png) 5. 然后在下面可以看到响应数据,说明用户已经创建完成。 ![用户创建完成](fastapi/resp.png) 用户创建的接口完成,下面开始写用户的登录接口 ## 用户登录 1. 打开文件`views/login.py`,新增一个登录接口。 ```python from fastapi import APIRouter, HTTPException, Depends, Body from pydantic import BaseModel from starlette import status from models.user import User router = APIRouter( # 请求路径 prefix="/1/login", # 标签, 文档上显示 tags=["登录"], ) class RequestUserLogin(BaseModel): """ 用户登录请求数据校验 """ phone: str = Body(..., max_length=11, description="登录手机号") password: str = Body(..., description="登录密码") class ResponseUserLogin(BaseModel): """ 用户登录响应数据 """ name: str = None access_token: str = None @router.post("/", response_model=ResponseUserLogin, response_model_exclude_unset=True) async def login(data: RequestUserLogin): """ 登录 :param data: :return: """ user = User.filter(phone=data.phone, password=data.password) if not user: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="用户名或密码错误.") # 生成一个token返回给前端, 做登录校验使用 access_token = "0123456789" return {"access_token": access_token, "name": user.name} ``` 2. 写个脚本使用`requests`调用一下登录接口,测试接口的可用性(可选)。 ```python import requests url = "http://127.0.0.1:8000/1/login/" payload = { "name": "用户名", "password": "密码123" } headers = { 'Authorization': '0123456789', 'Content-Type': 'application/json' } response = requests.request("POST", url, headers=headers, json=payload) print(response.json()) ``` 返回以下内容表示登录成功: ```python { "name": "用户名", "access_token": "0123456789" } ``` ---">

FastApi程序搭建

使用FastApi框架完成一个用户的创建、登录,虽然只有两个接口,其中包括了MySQL的配置和连接、orm工具和迁移工具的使用、中间件的使用等。

删除字符串中的空格 ```python s = " 123abc " # 删除开头的空格 print(s.lstrip()) # "123 abc " # 删除结尾的空格 print(s.rstrip()) # " 123 abc" # 删除开头和结尾的空格 print(s.strip()) # "123 abc" # 删除字符串中所有的空格 print(s.replace(' ', '')) # "123abc" ``` ### 删除字符串中的所有符号,只保留数字和英文字母 ```python import re s = "123,abc .?/&?》^_^dddA。" # 把所有编码非\u0030-\u0039(数字)、\u0041-\u007a(英文字母)的字符替换为空字符串 rs = re.sub("([^\u0030-\u0039\u0041-\u007a])", '', s) print(rs) # "123abcdddA" ``` ### 只字符串中的保留汉字 ```python import re s = "我爱中国🇨🇳,I love China。" # 把所有编码非\u4e00-\u9fa5(汉字)的字符替换为空字符串 rs = re.sub("([^\u4e00-\u9fa5])", '', s) print(rs) # "我爱中国" ``` ### 对应的unicode编码范围 |说明|unicode范围| |--|--| |数字|\u0030-\u0039| |汉字|\u4e00-\u9fa5| |大写字母|\u0041-\u005a| |小写字母|\u0061-\u007a| |英文字母|\u0041-\u007a| |韩文|\uAC00-\uD7AF| |日文|\u3040-\u31FF|">

Python删除字符串中的符号

Python删除字符串中的指定符号、空格、保留指定类型字符。

简单使用`return_value`返回固定的数据 - 方法一,使用装饰器。 ```python import mock def add(a, b): return a + b # 使用mock的return_value参数改变add返回的数据 @mock.patch('add', return_value=10) def test_add(): print(add(1, 2)) # 10 ``` - 方法二,使用mock.path方法。 ```python import mock def test_add(): with mock.patch('add', return_value=10): print(add(1, 2)) # 10 ``` ## 使用`side_effect`返回可变的数据 - 当一个方法被同一个测试调用多次,而自己又不想得到同一个返回值,就轮到`side_effect`上场了。 - 依次返回list里的数据。 ```python # 方法一 @mock.patch('add', side_effect=[10, 20]) def test_add(): print(add(1, 2)) # 10 print(add(1, 2)) # 20 # 方法二 def test_add(): with mock.patch('add', side_effect=[10, 20]): print(add(1, 2)) # 10 print(add(1, 2)) # 20 ``` - 根据传入的参数返回指定的数据。 ```python # 方法一 @mock.patch('add', side_effect={(1, 2): 10, (2, 3): 20}) def test_add(): print(add(2, 3)) # 20 print(add(1, 2)) # 10 # 方法二 def test_add(): with mock.patch('add', side_effect={(1, 2): 10, (2, 3): 20}): print(add(2, 3)) # 20 print(add(1, 2)) # 10 ``` ---">

pytest mock的使用

pytest mock的使用方法