Practice Now
AWS
Boto3
Asyncio
Python
Cloud Computing

How to query aws resources using boto3 and asyncio? Is this possible?

System Design practice on Codemia

Work through 120+ system design problems with detailed solutions, from rate limiters to multi-region storage.

Practice system design
AWS & CloudPython

Introduction

boto3 is synchronous by default — every API call blocks the calling thread until the response arrives. Python's asyncio provides cooperative concurrency for I/O-bound tasks, but boto3 does not natively support async/await. To use boto3 with asyncio, you have three options: run synchronous boto3 calls in a thread pool via asyncio.to_thread(), use the third-party aiobotocore library (which provides true async AWS API calls), or use aioboto3 (a higher-level async wrapper). The thread pool approach works with standard boto3 and requires no additional dependencies.

Method 1: asyncio.to_thread() (Python 3.9+)

Run synchronous boto3 calls in the default thread pool to avoid blocking the event loop.

python
1import asyncio
2import boto3
3
4async def list_s3_buckets():
5    s3 = boto3.client('s3')
6    # Run the blocking call in a thread
7    response = await asyncio.to_thread(s3.list_buckets)
8    return [b['Name'] for b in response['Buckets']]
9
10async def list_ec2_instances():
11    ec2 = boto3.client('ec2')
12    response = await asyncio.to_thread(ec2.describe_instances)
13    instances = []
14    for reservation in response['Reservations']:
15        for instance in reservation['Instances']:
16            instances.append(instance['InstanceId'])
17    return instances
18
19async def main():
20    # Run both queries concurrently
21    buckets, instances = await asyncio.gather(
22        list_s3_buckets(),
23        list_ec2_instances()
24    )
25    print(f"S3 Buckets: {buckets}")
26    print(f"EC2 Instances: {instances}")
27
28asyncio.run(main())

For Python 3.7-3.8, use loop.run_in_executor() instead:

python
1import asyncio
2import boto3
3from functools import partial
4
5async def list_s3_buckets():
6    s3 = boto3.client('s3')
7    loop = asyncio.get_event_loop()
8    response = await loop.run_in_executor(None, s3.list_buckets)
9    return [b['Name'] for b in response['Buckets']]

Method 2: aiobotocore (True Async)

aiobotocore is a thin async wrapper around botocore (the library underlying boto3) that provides native async/await support without threads.

bash
pip install aiobotocore
python
1import asyncio
2from aiobotocore.session import get_session
3
4async def list_s3_objects(bucket_name):
5    session = get_session()
6    async with session.create_client('s3') as s3:
7        response = await s3.list_objects_v2(Bucket=bucket_name)
8        return [obj['Key'] for obj in response.get('Contents', [])]
9
10async def describe_rds_instances():
11    session = get_session()
12    async with session.create_client('rds') as rds:
13        response = await rds.describe_db_instances()
14        return [db['DBInstanceIdentifier'] for db in response['DBInstances']]
15
16async def main():
17    objects, databases = await asyncio.gather(
18        list_s3_objects('my-bucket'),
19        describe_rds_instances()
20    )
21    print(f"S3 Objects: {objects}")
22    print(f"RDS Instances: {databases}")
23
24asyncio.run(main())

Method 3: aioboto3 (High-Level Async)

aioboto3 wraps aiobotocore to provide an interface similar to boto3, including resource-level abstractions.

bash
pip install aioboto3
python
1import asyncio
2import aioboto3
3
4async def upload_file(bucket, key, data):
5    session = aioboto3.Session()
6    async with session.client('s3') as s3:
7        await s3.put_object(Bucket=bucket, Key=key, Body=data)
8        print(f"Uploaded {key}")
9
10async def scan_dynamodb_table(table_name):
11    session = aioboto3.Session()
12    async with session.resource('dynamodb') as dynamodb:
13        table = await dynamodb.Table(table_name)
14        response = await table.scan()
15        return response['Items']
16
17async def main():
18    # Upload multiple files concurrently
19    tasks = [
20        upload_file('my-bucket', f'file_{i}.txt', f'content {i}'.encode())
21        for i in range(10)
22    ]
23    await asyncio.gather(*tasks)
24
25asyncio.run(main())

Paginating Async Results

Many AWS APIs return paginated results. Handle pagination in async code:

python
1from aiobotocore.session import get_session
2
3async def list_all_s3_objects(bucket_name):
4    session = get_session()
5    all_objects = []
6    async with session.create_client('s3') as s3:
7        paginator = s3.get_paginator('list_objects_v2')
8        async for page in paginator.paginate(Bucket=bucket_name):
9            for obj in page.get('Contents', []):
10                all_objects.append(obj['Key'])
11    return all_objects
12
13# With boto3 + asyncio.to_thread
14async def list_all_s3_objects_sync(bucket_name):
15    s3 = boto3.client('s3')
16    paginator = s3.get_paginator('list_objects_v2')
17
18    all_objects = []
19    def _paginate():
20        for page in paginator.paginate(Bucket=bucket_name):
21            for obj in page.get('Contents', []):
22                all_objects.append(obj['Key'])
23        return all_objects
24
25    return await asyncio.to_thread(_paginate)

Common Pitfalls

  • Creating boto3 clients inside the event loop without threading: boto3.client() and boto3.resource() are synchronous and may perform network calls (e.g., STS for credentials). Creating them inside an async function without to_thread() blocks the event loop. Create clients in a thread or before starting the event loop.
  • Sharing a single boto3 client across tasks: boto3 clients are not thread-safe. When using asyncio.to_thread() with asyncio.gather(), each task should create its own client, or use a thread-local client pattern.
  • Mixing aiobotocore and boto3 in the same project: aiobotocore pins a specific version of botocore which may conflict with the version boto3 requires. This causes version conflicts during pip install. Choose one approach per project.
  • Not using async with for aiobotocore clients: aiobotocore clients must be used within async with blocks to properly close HTTP connections. Forgetting the context manager leaks connections and eventually causes ConnectionError or resource exhaustion.
  • Rate limiting not handled: Running many AWS API calls concurrently with asyncio.gather() can trigger AWS throttling. Use asyncio.Semaphore to limit concurrency: sem = asyncio.Semaphore(10) and async with sem: await api_call().

Summary

  • boto3 is synchronous — use asyncio.to_thread() to run it concurrently without blocking the event loop
  • Use aiobotocore for true async AWS API calls without threads
  • Use aioboto3 for a higher-level async interface similar to boto3
  • Always use async with context managers for aiobotocore/aioboto3 clients
  • Limit concurrent API calls with asyncio.Semaphore to avoid AWS throttling
Related reading
Course
Beginner
27 lessons
10 hours
System Design Fundamentals

Build a strong foundation in designing scalable, reliable distributed systems.

View the course
Track what you have practised

A free account saves your progress, solutions and study plan across every problem on Codemia.

System Design practice on Codemia

Work through 120+ system design problems with detailed solutions, from rate limiters to multi-region storage.

Practice system design
Codemia
© 2026 Codemia
Resources
Legal
Social
All Rights Reserved.