You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何通过装饰器注入AsyncSession并消除Pylance类型警告?

问题原因分析与解决方案

类型不兼容警告的核心原因

你遇到的Pylance类型警告,本质是装饰器返回的inner函数签名与你通过Protocol定义的原函数签名不匹配:

  • 你定义的Protocol要求函数必须接受一个必填的session: AsyncSession参数,调用时必须传入该参数才能符合类型规范。
  • 但装饰后的inner函数实际允许不传session(装饰器会自动实例化并注入会话),这就导致inner的参数要求与Protocol的定义冲突,类型检查器识别到这种差异后抛出不兼容警告。

举个具体的例子:假设你定义的Protocol是:

from typing import Protocol, Awaitable
from sqlalchemy.ext.asyncio import AsyncSession

class SessionFunc(Protocol):
    async def __call__(self, session: AsyncSession) -> Awaitable[Any]:
        ...

而装饰器返回的inner函数实际签名是async def inner(session: Optional[AsyncSession] = None) -> Any:,这两个类型完全不兼容——前者要求必须传session,后者允许不传,Pylance自然会在return inner处报错。

解决思路与实现方案

要同时满足「函数定义时session是必填类型」和「调用时可传可不传且无类型警告」的需求,需要用ParamSpec+TypeVar结合overload来准确标注装饰器的类型,让类型检查器理解装饰前后的函数签名变化:

from typing import overload, Callable, Awaitable, TypeVar, ParamSpec, Optional
from sqlalchemy.ext.asyncio import AsyncSession
from contextlib import asynccontextmanager

# 定义参数和返回值的类型变量
P = ParamSpec("P")
R = TypeVar("R")

class Database:
    @staticmethod
    @asynccontextmanager
    async def session() -> AsyncSession:
        # 这里替换为你的实际会话创建/销毁逻辑
        session = AsyncSession(...)
        try:
            yield session
        finally:
            await session.close()

# 用overload标注装饰器的两种调用场景
@overload
def inject_session(func: Callable[[AsyncSession, *P.args], Awaitable[R]]) -> Callable[P, Awaitable[R]]:
    ...

@overload
def inject_session(func: Callable[P, Awaitable[R]]) -> Callable[P, Awaitable[R]]:
    ...

def inject_session(func):
    async def inner(*args: P.args, **kwargs: P.kwargs) -> R:
        # 提取已传入的session,没有则自动创建
        session: Optional[AsyncSession] = kwargs.pop("session", None)
        if not session:
            async with Database.session() as session:
                return await func(session, *args, **kwargs)
        return await func(session, *args, **kwargs)
    return inner

这样使用时:

@inject_session
async def create_model(session: AsyncSession, name: str):
    # 你的业务逻辑,无需检查session是否存在
    pass

async def main():
    # 调用时不传session,无类型警告
    await create_model("test")
    # 手动传入session,也符合类型规范
    async with Database.session() as sess:
        await create_model(sess, "test2")

这个方案既保留了函数定义时session的必填类型约束,又允许调用时灵活选择是否传入会话,同时消除了Pylance的类型警告。

内容的提问来源于stack exchange,提问作者Vadim

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.30 09:00:04