Python异步嵌套类场景下的设计模式方案求助

核心思路：把Session管理逻辑抽成可复用模块 #

别让HttpClient和ServiceClient各自写一遍两种模式的逻辑，把「独立创建Session」「上下文复用Session」这部分通用代码抽成基类或者混入类，让两个类都继承它；或者让ServiceClient直接组合HttpClient实例，用上下文管理器控制Session生命周期，这样就能避免重复代码。

方案1：用混入类(Mixin)封装Session管理 #

先写一个专门管Session的混入类，把两种模式的逻辑封装好：

import aiohttp
from typing import Optional

class SessionManagerMixin:
    def __init__(self, http_client: Optional['HttpClient'] = None):
        self._http_client = http_client
        self._own_session = False

    async def __aenter__(self):
        # 上下文模式：如果没传现成的HttpClient，就自己创建并初始化Session
        if self._http_client is None:
            self._http_client = HttpClient()
            await self._http_client.__aenter__()
            self._own_session = True
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # 只有自己创建的Session才需要清理，别碰外部传入的
        if self._own_session and self._http_client:
            await self._http_client.__aexit__(exc_type, exc_val, exc_tb)
            self._http_client = None
            self._own_session = False

    def _get_http_client(self) -> 'HttpClient':
        # 用的时候自动判断：有复用的就用复用的，没有就临时创建新的
        if self._http_client:
            return self._http_client
        return HttpClient()

然后是你的HttpClient（补个完整示例，和你已实现的逻辑对齐）：

class HttpClient:
    def __init__(self):
        self._session = None

    async def __aenter__(self):
        self._session = aiohttp.ClientSession()
        await self._session.__aenter__()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self._session:
            await self._session.__aexit__(exc_type, exc_val, exc_tb)
            self._session = None

    # 示例GET方法，自动适配两种模式
    async def get(self, url: str):
        if not self._session:
            # 非上下文模式：临时创建Session用完就关
            async with aiohttp.ClientSession() as session:
                async with session.get(url) as resp:
                    return await resp.json()
        # 上下文模式：复用已有的Session
        async with self._session.get(url) as resp:
            return await resp.json()

最后ServiceClient直接继承这个混入类，不用管Session的事儿，专注写业务方法就行：

class ServiceClient(SessionManagerMixin):
    async def fetch_user(self, user_id: int):
        http_client = self._get_http_client()
        return await http_client.get(f"https://api.example.com/users/{user_id}")

    async def fetch_orders(self, user_id: int):
        http_client = self._get_http_client()
        return await http_client.get(f"https://api.example.com/users/{user_id}/orders")

两种使用方式验证 #

1. 直接调用（自动创建独立Session）：

async def test_direct():
    service = ServiceClient()
    user = await service.fetch_user(123)
    orders = await service.fetch_orders(123)
    print(user, orders)

2. 上下文复用Session：

async def test_context():
    async with ServiceClient() as service:
        user = await service.fetch_user(123)
        orders = await service.fetch_orders(123)
        print(user, orders)
    # 退出上下文后Session自动关闭，不用手动处理

方案2：组合HttpClient实例（更灵活） #

如果不想用继承，让ServiceClient直接持有HttpClient实例，用同样的思路控制生命周期：

class ServiceClient:
    def __init__(self, http_client: Optional[HttpClient] = None):
        self._http_client = http_client
        self._own_client = False

    async def __aenter__(self):
        if self._http_client is None:
            self._http_client = HttpClient()
            await self._http_client.__aenter__()
            self._own_client = True
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self._own_client and self._http_client:
            await self._http_client.__aexit__(exc_type, exc_val, exc_tb)
            self._http_client = None
            self._own_client = False

    async def _get_client(self) -> HttpClient:
        if self._http_client:
            return self._http_client
        # 非上下文模式：临时创建HttpClient，内部会自动生成独立Session
        return HttpClient()

    async def fetch_user(self, user_id: int):
        client = await self._get_client()
        return await client.get(f"https://api.example.com/users/{user_id}")

关键注意事项 #

• 别乱清理资源：只有自己创建的Session/HttpClient才在__aexit__里关闭，避免把外部传入的资源搞炸。
• 职责分开：HttpClient只管HTTP请求，ServiceClient只管业务逻辑，别混在一起。
• 复用逻辑集中：把Session管理的代码只写一遍，让多个类共享，避免重复造轮子。

内容的提问来源于stack exchange，提问作者kriska