aiobotocore

Warnings

• breaking create_client() MUST be used as an async context manager (async with). Calling it without a context manager and manually closing is unreliable, and since v3.0.0 creating a new ClientSession after the client exits its context is explicitly forbidden and raises an error.
  Fix: Always use: async with session.create_client('s3', ...) as client: ...
• breaking The aiobotocore[boto3] and aiobotocore[awscli] packaging extras were removed in v3.0.0. Installing them will raise a pip error about unknown extras.
  Fix: Install separately: pip install aiobotocore boto3 or pip install aiobotocore awscli
• breaking Credential properties (credentials.access_key, credentials.secret_key, credentials.token) raise NotImplementedError because they do not trigger an async refresh cycle.
  Fix: Use the async method: frozen = await credentials.get_frozen_credentials()
• gotcha aiobotocore pins botocore to a narrow version range that changes with every release. Installing aiobotocore alongside boto3, awscli, s3fs, or moto frequently causes pip dependency conflicts because those packages independently pin botocore.
  Fix: Pin all AWS-related packages together (aiobotocore, boto3, botocore) to a mutually compatible set. Check the aiobotocore release notes for the supported botocore range. Use a lockfile (pip-tools, poetry, uv) to resolve conflicts upfront.
• gotcha get_object() response Body must be consumed inside an async context manager. Reading Body outside of 'async with response["Body"] as stream:' can leave the underlying TCP connection dangling and cause connection pool exhaustion.
  Fix: async with response['Body'] as stream: data = await stream.read()
• gotcha Paginators are async iterators in aiobotocore. You must use 'async for result in paginator.paginate(...):', NOT a regular 'for' loop. A regular for loop will not await pages and silently yields nothing or raises TypeError.
  Fix: paginator = client.get_paginator('list_objects_v2') async for page in paginator.paginate(Bucket='my-bucket'): for obj in page.get('Contents', []): print(obj['Key'])
• deprecated Passing the 'loop' parameter to get_session() is a legacy pattern from pre-asyncio.run() era and is no longer supported. Python's asyncio manages the event loop implicitly.
  Fix: Use get_session() with no arguments, and run coroutines with asyncio.run(main()).

Install

• pip install aiobotocore Core install
• pip install aiobotocore boto3 With boto3 (install separately — extras removed in v3)
• pip install 'types-aiobotocore[essential]' Type stubs for common services

Imports

• get_session

  from aiobotocore.session import get_session

  Top-level aiobotocore.get_session(loop=...) is the old pre-1.0 pattern. The loop parameter was removed. Always import from aiobotocore.session.
• AioSession

  from aiobotocore.session import AioSession

  Use when you need to subclass or explicitly type the session object.
• AioConfig

  from aiobotocore.config import AioConfig

  Required to pass connector_args (keepalive_timeout, use_dns_cache, etc.) or to set the experimental httpx backend.
• AIOHTTPSession

  from aiobotocore.httpsession import AIOHTTPSession

  Default HTTP session class; import only needed when customising the session class directly.
• HttpxSession

  from aiobotocore.httpxsession import HttpxSession

  Experimental httpx backend. Pass via AioConfig(http_session_cls=HttpxSession). Not fully tested; some aiohttp features are unported.
• ClientError

  import botocore.exceptions; botocore.exceptions.ClientError

  Error types live in botocore, not aiobotocore. Always catch botocore.exceptions.ClientError for AWS API errors.

Quickstart

List S3 buckets using an async context-managed client with credentials from environment variables.

import asyncio
import os
from aiobotocore.session import get_session
import botocore.exceptions

async def main():
    session = get_session()
    async with session.create_client(
        's3',
        region_name='us-east-1',
        aws_access_key_id=os.environ.get('AWS_ACCESS_KEY_ID', ''),
        aws_secret_access_key=os.environ.get('AWS_SECRET_ACCESS_KEY', ''),
    ) as client:
        try:
            response = await client.list_buckets()
            for bucket in response.get('Buckets', []):
                print(bucket['Name'])
        except botocore.exceptions.ClientError as e:
            print(f"AWS error: {e.response['Error']['Code']} - {e.response['Error']['Message']}")
        except botocore.exceptions.NoCredentialsError:
            print('No credentials found')

asyncio.run(main())