使用aiobotocore上下文管理器时__aenter__/__aexit__未执行问题

你的代码存在4个核心问题，导致上下文逻辑不执行、凭证配置不生效：

• 最根本的原因：异步上下文管理器必须通过async with语法调用。直接手动实例化类只会执行__init__构造方法，Python不会自动触发__aenter__和__aexit__的生命周期逻辑，这也是运行后只打印__init__的直接原因。
• 客户端初始化逻辑错误：AioSession是会话类，你没有做实例化就直接调用create_client方法会触发参数异常；同时你在__init__中接收的AWS访问凭证，根本没有传入客户端初始化参数，凭证配置完全没有生效。
• 上下文返回值缺失：__aenter__方法没有返回实例对象，后续即使正确使用async with语法，也无法拿到客户端实例调用S3操作方法。
• 额外的拼写错误：构造方法参数名写为aws_acces_key_id，和正确拼写aws_access_key_id差了一个s，容易导致后续传参混乱。

import asyncio
from contextlib import AsyncExitStack
from aiobotocore.session import AioSession
from credentials import aws_access_key_id, aws_secret_access_key


class AWSConnectionManager:
    def __init__(self, aws_access_key_id, aws_secret_access_key, region_name="us-east-1"):
        self.aws_access_key_id = aws_access_key_id
        self.aws_secret_access_key = aws_secret_access_key
        self.region_name = region_name
        self._exit_stack = AsyncExitStack()
        self._client = None

    async def __aenter__(self):
        # 先实例化会话对象
        session = AioSession()
        # 创建S3客户端时显式传入凭证、区域配置
        self._client = await self._exit_stack.enter_async_context(
            session.create_client(
                "s3",
                aws_access_key_id=self.aws_access_key_id,
                aws_secret_access_key=self.aws_secret_access_key,
                region_name=self.region_name
            )
        )
        # 返回自身实例，供外部调用封装的操作方法
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # 自动释放上下文栈管理的所有资源
        await self._exit_stack.__aexit__(exc_type, exc_val, exc_tb)

    # 封装常用S3操作
    async def put_object(self, bucket: str, key: str, data: bytes):
        return await self._client.put_object(Bucket=bucket, Key=key, Body=data)

    async def read_object(self, bucket: str, key: str) -> bytes:
        resp = await self._client.get_object(Bucket=bucket, Key=key)
        return await resp["Body"].read()

    async def delete_object(self, bucket: str, key: str):
        return await self._client.delete_object(Bucket=bucket, Key=key)


async def main():
    # 必须使用async with语法调用异步上下文管理器
    async with AWSConnectionManager(
        aws_access_key_id=aws_access_key_id,
        aws_secret_access_key=aws_secret_access_key
    ) as s3_client:
        # 测试上传
        await s3_client.put_object("your-bucket-name", "test.txt", b"hello world")
        # 测试读取
        content = await s3_client.read_object("your-bucket-name", "test.txt")
        print("读取到的文件内容：", content)
        # 测试删除
        await s3_client.delete_object("your-bucket-name", "test.txt")


if __name__ == "__main__":
    asyncio.run(main())

• 异步上下文管理器必须在async def定义的协程函数内通过async with调用，最后通过asyncio.run()启动事件循环执行，不能直接在同步顶层代码中实例化使用。
• 代码中默认区域设置为us-east-1，可以根据实际部署情况替换为自己的S3桶所在区域。
• 如果你的运行环境已经配置了环境变量、~/.aws/credentials凭证文件或者绑定了有权限的IAM角色，可以不用手动传入aws_access_key_id和aws_secret_access_key参数，aiobotocore会自动按照AWS默认凭证链查找可用凭证。

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