织梦开发小说网站教程,电脑优化,泰安新闻联播,c 网站开发人员工具依赖注入常用于以下场景#xff1a; 共享业务逻辑#xff08;复用相同的代码逻辑#xff09; 共享数据库连接 实现安全、验证、角色权限 等……
上述场景均可以使用依赖注入#xff0c;将代码重复最小化。
创建依赖项
依赖项就是一个函数#xff0c;且可以使用与路…依赖注入常用于以下场景 共享业务逻辑复用相同的代码逻辑 共享数据库连接 实现安全、验证、角色权限 等……
上述场景均可以使用依赖注入将代码重复最小化。
创建依赖项
依赖项就是一个函数且可以使用与路径操作函数相同的参数依赖项函数的形式和结构与路径操作函数一样。因此可以把依赖项当作没有「装饰器」即没有 app.get(/some-path) 的路径操作函数。
from typing import Unionimport uvicorn
from fastapi import FastAPI, Dependsapp FastAPI()# 依赖的组件函数
async def common_parameters(q: Union[str, None] None,skip: int 0,limit: int 100,
):return {skip: skip, limit: limit, q: q}app.get(/items/)
async def read_items(commons: dict Depends(common_parameters)):return commonsapp.get(/users/)
async def read_users(commons: dict Depends(common_parameters)):return commonsif __name__ __main__:uvicorn.run(app)
实现过程
1、导入 Depends
from fastapi import Depends
2、定义依赖的组件函数
# 依赖的组件函数
async def common_parameters(q: Union[str, None] None,skip: int 0,limit: int 100,
):return {skip: skip, limit: limit, q: q}
3、声明依赖项
声明依赖项需要使用 Depends 和一个新的参数
async def read_items(commons: dict Depends(common_parameters)):
其他与「依赖注入」概念相同的术语为 资源Resource 提供方Provider 服务Service 可注入Injectable 组件Component
依赖注入系统如此简洁的特性让 FastAPI 可以与下列系统兼容 关系型数据库 NoSQL 数据库 外部支持库 外部 API 认证和鉴权系统 API 使用监控系统 响应数据注入系统 等等……
类作为依赖项
依赖项时一个可调用的对象Python 中的 可调用对象 是指任何 Python 可以像函数一样 调用 的对象。Python 类也是 可调用对象。因此在 FastAPI 中你可以使用一个 Python 类作为一个依赖项。
实际上 FastAPI 检查的是它是一个 可调用对象函数类或其他任何类型以及定义的参数。
import uvicorn
from fastapi import FastAPI, Dependsapp FastAPI()fake_items_db [{item_name: Foo}, {item_name: Bar}, {item_name: Baz}]class CommonQueryParams:def __init__(self, q: str | None None, skip: int 0, limit: int 100):self.q qself.skip skipself.limit limit# FastAPI 调用 CommonQueryParams 类。这将创建该类的一个 实例该实例将作为参数 commons 被传递给你的函数。
app.get(/items/)
async def read_items(commons: CommonQueryParams Depends(CommonQueryParams)):response {}if commons.q:response.update({q: commons.q})items fake_items_db[commons.skip: commons.skip commons.limit]response.update({items: items})return responseif __name__ __main__:uvicorn.run(app, host127.0.0.1, port8000)
FastAPI 为这些情况提供了一个快捷方式在这些情况下依赖项 明确地 是一个类FastAPI 将 调用 它来创建类本身的一个实例。
快捷写法
commons: CommonQueryParams Depends()
分层依赖注入
假设是用户管理模块创建如下的目录结构以及python文件 schemas/user_schema.py
from pydantic import BaseModel, Field
from datetime import datetimeclass User(BaseModel):id: int | None Field(description主键ID)username: str | None Field(description用户名)age: int | None Field(description年龄)is_vip: int | None Field(description是否会员 0-不是1-是)create_time: datetime Field(description创建时间)
dao/user_dao.py
from datetime import datetime
from user.schemas.user_schema import Userclass UserDAO:数据层def __init__(self):passdef get_user_by_username(self, username)- User:根据用户名查询用户信息:param username::return:user User(id1, usernamealice, age25, is_vip0, create_timedatetime.now())return user
service/user_service.py
from fastapi import Dependsfrom user.dao.user_dao import UserDAO
from user.schemas.user_schema import Userclass UserService:业务逻辑处理层def __init__(self, user_dao: UserDAO Depends()):self.user_dao user_daodef find_by_username(self, username: str)- User:根据用户名查询用户信息:param username::return:return self.user_dao.get_user_by_username(username)
routers/user_routers.py
from fastapi import APIRouter, Dependsfrom commons.result import Result
from user.exceptions.user_exception import UserNotFoundException
from user.service.user_serivce import UserServiceusers APIRouter()users.get(path/find/{username}, summary根据用户名查询用户, response_modelResult)
async def find_user_by_username(username: str, user_service: UserService Depends()):user user_service.find_by_username(username)# 用户不存在if user is None:raise UserNotFoundExceptionresult Result()result.data userreturn result
commoms/result.py
from typing import Anyfrom pydantic import BaseModel, Fieldclass Result(BaseModel):code: int Field(description响应状态码, default200)message: str Field(description响应结果信息, default成功)data: Any Field(description响应结果数据, defaultNone)
