14个文本转图像AI API
FastAPI “类视图”管理接口
2024-12-13
这里的类视图和常看到的Python Web中的类视图不太一致（常见的类视图仅提供了POST,GET,DELETE, PUT,...等http方法），这里的类视图时将原本的router.get的操作用类来统一管理， 通过装饰器实现。from fastapi import APIRouter
router = APIRouter()
@router.get("/")
def index():
return "/index"@router.get 背后
访问起源码可以看见调到self.add_api_route传了一个func，这个func在上面的代码中就是index这个函数 / endpoint，到这一层就够了
通过上面得知每个router.xx后面都调用了add_api_route，那要实现类视图，需要步骤如下：1. 创建APIRouter示例，2. 将类中的某些方法add_api_route加到APIRouter实例中即可Controller 装饰器
这里简化了参数描述，会缺少代码提示，实际和APIRouter的参数定义一致就行
class Controller:
def __init__(self, **kwargs):
"""kwargs 等同于APIRouter 实例化入参"""
self.kwargs = kwargs
def __call__(self, cls):
# 创建router实例
router: APIRouter = APIRouter(
**self.kwargs
)
# 返回被装饰类的所有方法和属性名称
for attr_name in dir(cls):
# 通过反射拿到对应属性的值 或方法对象本身
attr = getattr(cls, attr_name)
# 简单处理，如果函数返回的是个字典并且endpoint在这里面，就添加到router上
if isinstance(attr, dict) and "endpoint" in attr:
router.add_api_route(**attr)
# 然后把router 给被装饰的类上，再返回被装饰类
cls.router = router
return clsRequestMapping 装饰器
被装饰方法返回带
endpoint的字典，让Controller能够扫描到
class RequestMapping:
"""请求"""
def __init__(self, **kwargs):
self.kwargs = kwargs
def __call__(self, func):
# 这里这个endpoint 对应的value 就是被装饰的函数
# 返回的内容其实是符合self.api_add_route的入参要求
return {"endpoint": func, **self.kwargs}测试代码 main.py
from fastapi import APIRouter
class Controller:
def __init__(self, **kwargs):
"""kwargs 等同于APIRouter 实例化入参"""
self.kwargs = kwargs
def __call__(self, cls):
# 创建router实例
router: APIRouter = APIRouter(
**self.kwargs
)
# 返回被装饰类的所有方法和属性名称
for attr_name in dir(cls):
# 通过反射拿到对应属性的值 或方法对象本身
attr = getattr(cls, attr_name)
# 简单处理，如果函数返回的是个字典并且endpoint在这里面，就添加到router上
if isinstance(attr, dict) and "endpoint" in attr:
router.add_api_route(**attr)
# 然后把router 给被装饰的类上，再返回被装饰类
cls.router = router
return cls
class RequestMapping:
"""请求"""
def __init__(self, **kwargs):
self.kwargs = kwargs
def __call__(self, func):
# 这里这个endpoint 对应的value 就是被装饰的函数
# 返回的内容其实是符合self.api_add_route的入参要求
return {"endpoint": func, **self.kwargs}
def get_db():
print("模拟db session")
return "模拟db session"
@Controller(prefix="/demo", tags=["demo"])
class Demo:
name: str = "ggbond"
db_session: str = Depends(get_db)
@RequestMapping(path="")
def get_list(self):
return [self.name, self.db_session]
from fastapi import FastAPI
app = FastAPI()
app.include_router(Demo.router)
if __name__ == '__main__':
import uvicorn
uvicorn.run(app)问题
遇到的这些问题，解决方法其实之前大佬都写好了 https://github.com/dmontagu/fastapi-utils/blob/master/fastapi_utils/cbv.py
将self识别成了查询参数
解决方法 update_endpoint_signature
将
self这个方法签名修改成Depends(cls)– 实例化当前类对象，并依赖注入（依赖注入的参数，无需用户传递哦）。
def update_endpoint_signature(endpoint, cls_obj):
"""
source：https://github.com/dmontagu/fastapi-utils/blob/master/fastapi_utils/cbv.py#L92
修改端点签名，将self参数改成关键字参数 self = Depends(cls)
:param endpoint: 端点
:param cls_obj: 被装饰的类 本身
:return:
"""
# 获取原始端点函数的签名
old_signature = inspect.signature(endpoint)
# 获取参数列表
old_parameters = list(old_signature.parameters.values())
# 获取原始端点函数的第一个参数， self
old_first_parameter = old_parameters[0]
# 将第一个参数替换为依赖注入 当前类 的参数
#
new_first_parameter = old_first_parameter.replace(default=Depends(cls_obj))
new_parameters = [new_first_parameter] + [
parameter.replace(kind=inspect.Parameter.KEYWORD_ONLY)
for parameter in old_parameters[1:]
] # 替换其他参数的类型为 KEYWORD_ONLY，以便正确执行依赖注入
new_signature = old_signature.replace(parameters=new_parameters) # 创建新的签名
setattr(endpoint, "__signature__", new_signature) # 替换端点函数的签名
依赖注入未被正确解析
解决方法
需要修改类的签名、和init方法，并在init方法时调用依赖注入的函数这里的实例
get_db，并把结果更新到db_session中去
CBV_CLASS_KEY = "__cbv_class__"
def _init_cbv(cls: type[Any]) -> None:
"""
Idempotently modifies the provided `cls`, performing the following modifications:
* The `__init__` function is updated to set any class-annotated dependencies as instance attributes
* The `__signature__` attribute is updated to indicate to FastAPI what arguments should be passed to the initializer
"""
# 实例化之后就不用实例化了
if getattr(cls, CBV_CLASS_KEY, False): # pragma: no cover
return # Already initialized
# 保留原本的 init
old_init: Callable[..., Any] = cls.__init__
# 获取原本的 实例签名
old_signature = inspect.signature(old_init)
# 删除self
old_parameters = list(old_signature.parameters.values())[1:] # drop `self` parameter
# 改成关键字参数
new_parameters = [
x for x in old_parameters if x.kind not in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD)
]
# 依赖注入列表
dependency_names: list[str] = []
# 返回类 cls 中的所有类型标注，
for name, hint in get_type_hints(cls).items():
if is_classvar(hint):
continue
parameter_kwargs = {"default": getattr(cls, name, Ellipsis)}
dependency_names.append(name)
# 加入到关键字参数中
new_parameters.append(
inspect.Parameter(name=name, kind=inspect.Parameter.KEYWORD_ONLY, annotation=hint, **parameter_kwargs)
)
# 替换签名参数
new_signature = old_signature.replace(parameters=new_parameters)
# 实例化时候的处理
def new_init(self: Any, *args: Any, **kwargs: Any) -> None:
for dep_name in dependency_names:
# pop 会拿到依赖注入的返回值
dep_value = kwargs.pop(dep_name)
# 设置成属性
setattr(self, dep_name, dep_value)
# 再调用原来的实例化方法
old_init(self, *args, **kwargs)
setattr(cls, "__signature__", new_signature)
setattr(cls, "__init__", new_init)完整实现
boot.py
"""
Source: https://github.com/dmontagu/fastapi-utils/blob/master/fastapi_utils/cbv.py
"""
from __future__ import annotations
import inspect
from collections.abc import Callable
from typing import Any, TypeVar, get_type_hints
from fastapi import APIRouter, Depends
from pydantic.v1.typing import is_classvar
T = TypeVar("T")
CBV_CLASS_KEY = "__cbv_class__"
def _init_cbv(cls: type[Any]) -> None:
"""
Idempotently modifies the provided `cls`, performing the following modifications:
* The `__init__` function is updated to set any class-annotated dependencies as instance attributes
* The `__signature__` attribute is updated to indicate to FastAPI what arguments should be passed to the initializer
"""
if getattr(cls, CBV_CLASS_KEY, False): # pragma: no cover
return # Already initialized
old_init: Callable[..., Any] = cls.__init__
old_signature = inspect.signature(old_init)
old_parameters = list(old_signature.parameters.values())[1:] # drop `self` parameter
new_parameters = [
x for x in old_parameters if x.kind not in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD)
]
dependency_names: list[str] = []
for name, hint in get_type_hints(cls).items():
if is_classvar(hint):
continue
parameter_kwargs = {"default": getattr(cls, name, Ellipsis)}
dependency_names.append(name)
new_parameters.append(
inspect.Parameter(name=name, kind=inspect.Parameter.KEYWORD_ONLY, annotation=hint, **parameter_kwargs)
)
new_signature = old_signature.replace(parameters=new_parameters)
def new_init(self: Any, *args: Any, **kwargs: Any) -> None:
for dep_name in dependency_names:
dep_value = kwargs.pop(dep_name)
setattr(self, dep_name, dep_value)
old_init(self, *args, **kwargs)
setattr(cls, "__signature__", new_signature)
setattr(cls, "__init__", new_init)
setattr(cls, CBV_CLASS_KEY, True)
def _update_cbv_route_endpoint_signature(cls: type[Any], endpoint: Callable[..., Any]) -> None:
"""
Fixes the endpoint signature for a cbv route to ensure FastAPI performs dependency injection properly.
"""
old_signature = inspect.signature(endpoint)
old_parameters: list[inspect.Parameter] = list(old_signature.parameters.values())
old_first_parameter = old_parameters[0]
new_first_parameter = old_first_parameter.replace(default=Depends(cls))
new_parameters = [new_first_parameter] + [
parameter.replace(kind=inspect.Parameter.KEYWORD_ONLY) for parameter in old_parameters[1:]
]
new_signature = old_signature.replace(parameters=new_parameters)
setattr(endpoint, "__signature__", new_signature)
class Controller:
def __init__(self, **kwargs):
"""kwargs 等同于APIRouter 实例化入参"""
self.kwargs = kwargs
def __call__(self, cls):
_init_cbv(cls)
# 创建router实例
router: APIRouter = APIRouter(
**self.kwargs
)
# 返回被装饰类的所有方法和属性名称
for attr_name in dir(cls):
# 通过反射拿到对应属性的值 或方法对象本身
attr = getattr(cls, attr_name)
# 添加到router上
if isinstance(attr, BaseRoute) and hasattr(attr, "kwargs"):
_update_cbv_route_endpoint_signature(cls, attr.kwargs["endpoint"])
if isinstance(attr, RequestMapping):
router.add_api_route(**attr.kwargs)
elif isinstance(attr, WebSocket):
router.add_websocket_route(**attr.kwargs)
else:
assert False, "Cls Type is RequestMapping or WebSocket"
cls.router = router
return cls
class BaseRoute:
pass
class RequestMapping(BaseRoute):
"""请求"""
def __init__(self, **kwargs):
self.kwargs = kwargs
def __call__(self, func):
# 这里这个endpoint 对应的value 就是被装饰的函数
# 返回的内容其实是符合self.api_add_route的入参要求
self.kwargs["endpoint"] = func
return self
class WebSocket(BaseRoute):
def __init__(self, **kwargs):
self.kwargs = kwargs
def __call__(self, func):
# 这里这个endpoint 对应的value 就是被装饰的函数
# 返回的内容其实是符合self.api_add_route的入参要求
self.kwargs["endpoint"] = func
return selfmain.py
from fastapi import Depends
from boot import Controller, RequestMapping, WebSocket
def get_db():
print("模拟db session")
return "模拟db session"
@Controller(prefix="/demo", tags=["demo"])
class Demo:
name = "ggbond"
# 带类型标注类属性会被加入到方法中如router.get(db_session: str = Depends(get_db))
db_session: str = Depends(get_db)
@RequestMapping(path="")
def get_list(self, age: int):
return [self.name, self.db_session]
@WebSocket(path="/ws")
async def ws(self):
pass
from fastapi import FastAPI
app = FastAPI()
app.include_router(Demo.router)
if __name__ == '__main__':
import uvicorn
uvicorn.run(app)
文章转自微信公众号@7y记
