Implementation:Mage ai Mage ai DynamoDB Source
| Knowledge Sources | |
|---|---|
| Domains | Data_Integration, DynamoDB, Source_Connector, AWS |
| Last Updated | 2026-02-09 00:00 GMT |
Overview
Concrete tool for extracting data from Amazon DynamoDB tables provided by the Mage integrations source connector framework.
Description
The DynamoDb source connector extends the base Source class to implement data extraction from Amazon DynamoDB. It connects via boto3 with retry configuration (up to 10 attempts) and supports two authentication modes: direct AWS credentials (aws_access_key_id/aws_secret_access_key) and IAM role assumption via STS for cross-account access. Discovery uses list_tables() with pagination to enumerate all tables, then describe_table() to read key schemas and attribute definitions. Each DynamoDB attribute type is mapped: S (string), N (number), and B (binary). Key properties are extracted from the table's KeySchema. The load_data() method uses scan() with a limit of 100 items per page and supports bookmark-based pagination via ExclusiveStartKey/LastEvaluatedKey. Records are converted from DynamoDB's typed attribute format to flat dictionaries using the shared row_to_singer_record() utility from the MongoDB tap module. The test_connection() method calls describe_endpoints() to verify connectivity. Tables that return an AccessDeniedException during discovery are logged and skipped. The replication method is full-table, and the database type is set to DATABASE_TYPE_DYNAMODB.
Usage
Use this source connector when building a Mage data pipeline that needs to extract data from DynamoDB tables. Configure with AWS credentials (aws_access_key_id, aws_secret_access_key) or role_arn for IAM role assumption, plus an optional aws_region.
Code Reference
Source Location
- Repository: mage-ai
- File: mage_integrations/mage_integrations/sources/dynamodb/__init__.py
- Lines: 1-225
Signature
class DynamoDb(Source):
@property
def region(self) -> str:
...
def discover_streams(self) -> List[Dict]:
...
def discover_stream(self, client, table_name):
...
def discover(self, streams: List[str] = None) -> Catalog:
...
def build_client(self):
...
def test_connection(self) -> None:
...
def scan_table(self, table_name, last_evaluated_key):
...
def load_data(self, stream, bookmarks: Dict = None, query: Dict = None, sample_data: bool = False, start_date: datetime = None, **kwargs) -> Generator[List[Dict], None, None]:
...
Import
from mage_integrations.sources.dynamodb import DynamoDb
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| config | dict | Yes | Configuration dictionary with AWS credentials and region |
| catalog | Catalog | No | Singer catalog specifying streams to extract |
| state | dict | No | Previous sync state containing last evaluated key for pagination |
Configuration Parameters
| Name | Type | Required | Description |
|---|---|---|---|
| aws_access_key_id | str | No | AWS access key ID (not required when using role_arn) |
| aws_secret_access_key | str | No | AWS secret access key (not required when using role_arn) |
| aws_region | str | No | AWS region (defaults to us-west-2) |
| role_arn | str | No | IAM role ARN for cross-account STS role assumption |
| role_session_name | str | No | Session name for STS role assumption (defaults to mage-data-integration) |
Outputs
| Name | Type | Description |
|---|---|---|
| catalog | Catalog | Discovered tables with schemas from DynamoDB attribute definitions (from discover()) |
| records | Generator[List[Dict]] | Batches of scanned records converted from DynamoDB format (from load_data()) |
Usage Examples
from mage_integrations.sources.dynamodb import DynamoDb
config = {
"aws_access_key_id": "AKIAIOSFODNN7EXAMPLE",
"aws_secret_access_key": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
"aws_region": "us-east-1",
}
source = DynamoDb(config=config)
# Discover available streams
catalog = source.discover()
# Test connection
source.test_connection()