Implementation:Mage ai Mage ai Amazon S3 Source

Knowledge Sources	Mage Integrations Amazon S3
Domains	Data_Integration, Amazon_S3, Source_Connector, File_Based
Last Updated	2026-02-09 00:00 GMT

Overview

Concrete tool for extracting data from Amazon S3 buckets by reading CSV and Parquet files provided by the Mage integrations source connector framework.

Description

The AmazonS3 source connector extends the base Source class to implement data extraction from Amazon S3 buckets. It reads CSV and Parquet files from a configured bucket and prefix, infers schemas using pandas DataFrame type detection, and yields records as dictionaries. The connector supports two authentication modes: direct AWS access key/secret key credentials, and IAM role assumption via STS for cross-account access. It uses the boto3 client with automatic retry configuration (up to 10 attempts). Discovery reads the first matching file under the configured prefix to infer column types, including a special _s3_last_modified datetime column appended to track file modification times. The connector supports both single-stream and multi-stream configurations via table_configs, where each table config specifies its own prefix, search pattern, and table name. Files can be filtered by regex search pattern. During load_data(), bookmark-based incremental loading is supported using the S3 object's LastModified timestamp. The replication method is full-table. A custom S3-compatible endpoint URL can also be configured.

Usage

Use this source connector when building a Mage data pipeline that needs to extract data from Amazon S3 buckets containing CSV or Parquet files. Configure with bucket, prefix, and either aws_access_key_id/aws_secret_access_key or role_arn for IAM role assumption.

Code Reference

Source Location

• Repository: mage-ai
• File: mage_integrations/mage_integrations/sources/amazon_s3/__init__.py
• Lines: 1-288

Signature

class AmazonS3(Source):
    @property
    def bucket(self) -> str:
        ...
    @property
    def file_type(self) -> str:
        ...
    @property
    def prefix(self) -> str:
        ...
    @property
    def region(self) -> str:
        ...
    @property
    def search_pattern(self) -> str:
        ...
    @property
    def single_stream_in_prefix(self) -> bool:
        ...
    @property
    def endpoint(self) -> str:
        ...
    @property
    def table_configs(self):
        ...
    def build_client(self):
        ...
    def discover(self, streams: List[str] = None) -> Catalog:
        ...
    def discover_stream(self, prefix: str, search_pattern: str, table_name: str = None):
        ...
    def list_objects(self, prefix, search_pattern):
        ...
    def load_data(self, stream, bookmarks: Dict = None, *args, **kwargs) -> Generator[List[Dict], None, None]:
        ...
    def test_connection(self) -> None:
        ...

Import

from mage_integrations.sources.amazon_s3 import AmazonS3

I/O Contract

Inputs

Name	Type	Required	Description
config	dict	Yes	Configuration dictionary with S3 credentials and bucket settings
catalog	Catalog	No	Singer catalog specifying streams to extract
state	dict	No	Previous sync state for incremental extraction

Configuration Parameters

Name	Type	Required	Description
bucket	str	Yes	S3 bucket name
prefix	str	Yes	S3 key prefix to filter objects
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)
search_pattern	str	No	Regex pattern to filter S3 object keys
file_type	str	No	Explicit file type (csv or parquet); auto-detected from extension if omitted
single_stream_in_prefix	bool	No	Whether all files in the prefix belong to a single stream
aws_endpoint	str	No	Custom S3-compatible endpoint URL
table_configs	list	No	List of table configs for multi-stream mode, each with prefix, search_pattern, and table_name

Outputs

Name	Type	Description
catalog	Catalog	Discovered streams with schemas inferred from file contents (from discover())
records	Generator[List[Dict]]	Batches of records from CSV/Parquet files with _s3_last_modified column (from load_data())

Usage Examples

from mage_integrations.sources.amazon_s3 import AmazonS3

config = {
    "bucket": "my-data-bucket",
    "prefix": "data/exports/",
    "aws_access_key_id": "AKIAIOSFODNN7EXAMPLE",
    "aws_secret_access_key": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
    "aws_region": "us-east-1",
    "file_type": "csv",
}

source = AmazonS3(config=config)

# Discover available streams
catalog = source.discover()

# Test connection
source.test_connection()

Related Pages

Implements Principle

• Principle:Mage_ai_Mage_ai_Source_Lifecycle_Orchestration

Requires Environment

• Environment:Mage_ai_Mage_ai_Python_3_9_Runtime