aiodynamo: Asyncio DynamoDB Client
aiodynamo is an asynchronous Python client for Amazon DynamoDB, built using asyncio. It provides a clean, Pythonic API for interacting with DynamoDB, including support for expressions, transactions, and various credential providers. The current version is 24.7, and releases occur somewhat irregularly, often monthly or quarterly, incorporating bug fixes and new features.
Warnings
- breaking Support for Python 3.7 was dropped in version 23.10.
- gotcha aiodynamo requires an HTTP client library (like httpx or aiohttp) to be installed separately. The base `pip install aiodynamo` does not include one.
- gotcha Version 23.10 introduced a timing issue with refreshable credentials that was fixed in 23.10.1. Older versions might exhibit unstable credential refreshing.
- gotcha Logging for requests and responses was separated into distinct named loggers in version 24.7.
- gotcha Versions 21.9 and newer require `httpx>=0.15.0` when using the httpx client backend.
Install
-
pip install aiodynamo -
pip install aiodynamo[httpx] -
pip install aiodynamo[aiohttp]
Imports
- Client
from aiodynamo import Client
- AWSCredentials
from aiodynamo.credentials import AWSCredentials
- F
from aiodynamo.expressions import F
Quickstart
import asyncio
import os
from aiodynamo import Client, AWSCredentials
async def main():
# Configure AWS credentials and region via environment variables
# or replace with your actual values.
# AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION
client = Client(
region=os.environ.get("AWS_REGION", "us-east-1"),
credentials=AWSCredentials(
aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID", ""),
aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY", ""),
)
)
print("Attempting to connect to DynamoDB...")
try:
# Example: List tables (requires appropriate IAM permissions)
response = await client.list_tables()
print(f"DynamoDB Tables: {response.table_names}")
except Exception as e:
print(f"Error interacting with DynamoDB: {e}")
finally:
await client.close()
print("Client closed.")
if __name__ == "__main__":
asyncio.run(main())