Implementation:Mage ai Mage ai MongoDB Source

Knowledge Sources	Mage Integrations MongoDB
Domains	Data_Integration, MongoDB, Source_Connector
Last Updated	2026-02-09 00:00 GMT

Overview

Concrete tool for extracting data from MongoDB collections provided by the Mage integrations source connector framework.

Description

The MongoDB source connector extends the base Source class to implement data extraction from MongoDB databases. It uses pymongo (via the shared build_client() utility from the tap_mongodb module) for connections and pymongo_schema to automatically extract collection schemas. Discovery connects to the configured database, introspects all collection schemas (or specific ones if stream names are provided), and maps MongoDB types to Singer schema types: biginteger/integer become integer, boolean stays boolean, date becomes string with datetime format, dbref/oid/string become string, and float/number become number. All columns are nullable. The connector provides a custom sync() method that delegates to the tap_mongodb do_sync() function, populating stream metadata with database name, replication key, and replication method. It supports both full-table and log-based replication methods. The load_data() method builds a MongoDB find filter from bookmarks or query state and uses collection.find() with an optional sample limit of 100 documents. Records are converted from MongoDB's internal format via row_to_singer_record(). The count_records() method provides document counting with filter support. The test_connection() method calls server_info() to verify connectivity.

Usage

Use this source connector when building a Mage data pipeline that needs to extract data from MongoDB. Configure with standard MongoDB connection parameters and a database name.

Code Reference

Source Location

• Repository: mage-ai
• File: mage_integrations/mage_integrations/sources/mongodb/__init__.py
• Lines: 1-187

Signature

class MongoDB(Source):
    def discover(self, streams: List[str] = None) -> Catalog:
        ...
    def sync(self, catalog: Catalog) -> None:
        ...
    def count_records(self, stream, bookmarks: Dict = None, query: Dict = None, **kwargs) -> int:
        ...
    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]:
        ...
    def test_connection(self):
        ...

Import

from mage_integrations.sources.mongodb import MongoDB

I/O Contract

Inputs

Name	Type	Required	Description
config	dict	Yes	Configuration dictionary with MongoDB connection parameters
catalog	Catalog	No	Singer catalog specifying streams to extract
state	dict	No	Previous sync state for incremental or log-based extraction

Configuration Parameters

Name	Type	Required	Description
database	str	Yes	MongoDB database name to connect to
(connection params)	various	Yes	Standard MongoDB connection parameters passed to build_client (host, port, username, password, etc.)

Outputs

Name	Type	Description
catalog	Catalog	Discovered collections with schemas extracted via pymongo_schema (from discover())
records	Generator[List[Dict]]	Batches of records converted from MongoDB documents (from load_data())

Usage Examples

from mage_integrations.sources.mongodb import MongoDB

config = {
    "host": "mongodb://localhost:27017",
    "database": "my_database",
    "username": "admin",
    "password": "password123",
}

source = MongoDB(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