# MongoDB connector

Source: /docs/connectors/mongodb.md

Use this as an implementation starting point: adapt the worked example to the requested dataset and the source documentation. See [Custom API](/docs/connectors/custom-api) for the shared SDK pattern and document-upload workflow.

## What this connector is for [#what-this-connector-is-for]

Document-oriented database for collections, document queries, and change streams.

A connector owns access to the source. Separate syncs shape selected resources into tables for analysis and downstream transformations using the [Python SDK](/docs/reference/python/connectors-and-syncs).

## Data and tables you can pull [#data-and-tables-you-can-pull]

These are representative dataset families to consider, not a claim that the worked example implements every resource. Access, fields, and availability depend on the source account, product edition, and permissions. Use the [vendor reference](https://www.mongodb.com/docs/manual/) to select concrete resources.

| Suggested table | One row represents |
| --- | --- |
| `databases` | a database |
| `collections` | a collection |
| `documents` | a document |
| `index_definitions` | a index definition |

### Relationships and extraction considerations [#relationships-and-extraction-considerations]

A collection can contain documents with different shapes. Preserve document IDs and choose how to normalize nested arrays. Snapshot extraction and cursor-based incremental polling are different from change-stream consumption.

## Connecting to MongoDB [#connecting-to-mongodb]

`MongoDBConnector` creates and caches a PyMongo client using a secret `uri`
(`mongodb://` or `mongodb+srv://`) containing authentication and TLS options.
`database` selects the database used for discovery and extraction. Connection
and server selection timeouts are 10 seconds. `check()` issues a database ping;
collection listing and reads still require appropriate permissions.

`discover()` lists ordinary, non-system collections, excluding views and time
series collections. Each has a stable output schema: `_id` and `document`, both
canonical Extended JSON strings. The document includes all original fields;
BSON values such as ObjectIds, dates, and decimals retain type wrappers.

Use the connector as a context manager, or call `close()` when finished.

Additional resources may require additional scopes. Connection settings belong to the connector; dataset filters and resource selection belong to each sync. See the [CLI workflow](/docs/reference/cli/overview) for registration and configuration.

### Connection settings used by the example [#connection-settings-used-by-the-example]

| Setting | Requirement | Purpose |
| --- | --- | --- |
| `uri` | Required; secret | MongoDB connection string, including authentication and TLS options. Supports mongodb:// and mongodb+srv:// URIs. |
| `database` | Required | Database whose collections are available for discovery and syncing. |

### requirements.in [#requirementsin]

Save these dependencies in `requirements.in` alongside `connector.py`; the example includes packages used by its sync. Install the published `vortexify` package for local development. Keep sync-specific dependencies in the sync's own `requirements.in` when splitting the implementation into separate resources.

```text
vortexify==0.1.1
pymongo>=4.10,<5
pandas>=2,<3
```

PyMongo supplies the MongoDB driver. The URI, TLS trust, DNS/SRV support, and network rules must match the deployment.

## Worked example: sync `snapshot` [#worked-example-sync-snapshot]

The selected sync reads the resource shown below.

The complete files below implement only this example resource. They illustrate the current SDK signatures, a separate preview, bounded DataFrame output, and source-specific extraction. They do not establish complete source coverage. Live credentials and platform ingestion have not been used to validate this guide.

With default paths, place the connector files in `connectors/mongodb/` and sync files in `connectors/mongodb/syncs/snapshot/`. Use your project configuration if it specifies different paths.

### Connection schema: `config.json` [#connection-schema-configjson]

This file contains a JSON Schema, not actual credential values. Provide those through platform configuration.

```json
{
  "type": "object",
  "properties": {
    "uri": {
      "title": "Connection URI",
      "description": "MongoDB connection string, including authentication and TLS options. Supports mongodb:// and mongodb+srv:// URIs.",
      "type": "string",
      "minLength": 1,
      "examples": [
        "mongodb://username:password@localhost:27017/?authSource=admin"
      ],
      "x-vortexify-secret": true,
      "order": 0
    },
    "database": {
      "title": "Database",
      "description": "Database whose collections are available for discovery and syncing.",
      "type": "string",
      "minLength": 1,
      "examples": [
        "production"
      ],
      "order": 1
    }
  },
  "required": [
    "uri",
    "database"
  ]
}
```

### Connector: `connector.py` [#connector-connectorpy]

```python
from pymongo import MongoClient
from vortexify import Column, Connector, DiscoveredTable


class MongoDBConnector(Connector):
    vendor_domain = "mongodb.com"
    "Expose each collection as an ID plus its complete Extended JSON document."

    def client(self) -> MongoClient:
        if self._client is None:
            self._client = MongoClient(
                self.config["uri"],
                serverSelectionTimeoutMS=10000,
                connectTimeoutMS=10000,
                tz_aware=True,
            )
        return self._client

    def check(self) -> bool:
        result = self.client()[self.config["database"]].command("ping")
        return result["ok"] == 1

    def discover(self) -> list[DiscoveredTable]:
        database = self.client()[self.config["database"]]
        resources = []
        with database.list_collections(filter={"type": "collection"}) as collections:
            names = sorted((collection["name"] for collection in collections))
        for name in names:
            if name.startswith("system."):
                continue
            resources.append(
                DiscoveredTable(
                    name=f"{database.name}.{name}",
                    schema=[
                        Column(name="_id", type="string", nullable=False),
                        Column(name="document", type="string", nullable=False),
                    ],
                )
            )
        return resources

    def close(self) -> None:
        if self._client is not None:
            try:
                self._client.close()
            finally:
                self._client = None
```

### Sync schema: `syncs/snapshot/config.json` [#sync-schema-syncssnapshotconfigjson]

```json
{
  "type": "object",
  "properties": {
    "collection": {
      "title": "Collection",
      "description": "Collection to scan in the connector's configured database. Output contains _id and document as canonical Extended JSON strings. A full scan is not a point-in-time snapshot during concurrent writes.",
      "type": "string",
      "minLength": 1,
      "examples": [
        "orders"
      ],
      "order": 0
    }
  },
  "required": [
    "collection"
  ]
}
```

### Sync: `syncs/snapshot/sync.py` [#sync-syncssnapshotsyncpy]

```python
from collections.abc import Iterator
from math import isfinite
import pandas as pd
from bson import json_util
from vortexify import Sync


class MongoDBSnapshotSync(Sync):
    """Scan a collection in bounded batches; concurrent writes are not isolated."""

    def preview(self) -> pd.DataFrame:
        database = self.client[self.connector.config["database"]]
        collection = database[self.config["collection"]]
        query = {}
        rows = []
        with collection.find(query).limit(100).batch_size(100) as cursor:
            for document in cursor:
                rows.append(
                    {
                        "_id": json_util.dumps(
                            document["_id"],
                            json_options=json_util.CANONICAL_JSON_OPTIONS,
                        ),
                        "document": json_util.dumps(
                            document, json_options=json_util.CANONICAL_JSON_OPTIONS
                        ),
                    }
                )
        return pd.DataFrame.from_records(rows, columns=["_id", "document"])

    def extract(
        self, *, avg_row_size_bytes: float | None = None
    ) -> Iterator[pd.DataFrame]:
        max_batch_size = 1000
        memory_mib = self.compute.memory if self.compute else None
        memory_mib = memory_mib if memory_mib is not None and memory_mib > 0 else 1024
        target_bytes = min(16 * 1024 * 1024, max(1, memory_mib * 1024 * 1024 // 20))
        batch_size = 100
        if avg_row_size_bytes is not None:
            if not isfinite(avg_row_size_bytes) or avg_row_size_bytes <= 0:
                raise ValueError("avg_row_size_bytes must be finite and positive")
            batch_size = min(
                max_batch_size,
                max(1, int(target_bytes / max(1.0, avg_row_size_bytes) / 4)),
            )
        database = self.client[self.connector.config["database"]]
        collection = database[self.config["collection"]]
        rows = []
        with collection.find({}).batch_size(batch_size) as cursor:
            for document in cursor:
                rows.append(
                    {
                        "_id": json_util.dumps(
                            document["_id"],
                            json_options=json_util.CANONICAL_JSON_OPTIONS,
                        ),
                        "document": json_util.dumps(
                            document, json_options=json_util.CANONICAL_JSON_OPTIONS
                        ),
                    }
                )
                if len(rows) == batch_size:
                    yield pd.DataFrame.from_records(rows, columns=["_id", "document"])
                    rows = []
            if rows:
                yield pd.DataFrame.from_records(rows, columns=["_id", "document"])
```

### Running and extending the example [#running-and-extending-the-example]

Register the connector before the sync and commit each resource separately:

```sh
vortexify create connector mongodb
vortexify create sync mongodb/snapshot
```

Edit the files, inspect `vortexify status`, and use the [CLI guide](/docs/reference/cli/overview) to commit and configure them. Registration alone does not run extraction or deploy the connector.

Preview must run with a fresh connector and disposable state. The extraction implementation determines full/incremental behavior; merely setting a mode does not implement change capture. API/driver failures propagate unless explicitly handled in the code. Check source limits and single-record/response sizes as well as output batch size.

## Sources and further reading [#sources-and-further-reading]

- [Vendor API and data reference](https://www.mongodb.com/docs/manual/) — Resource definitions, authentication, fields, and source-specific limits. Check the selected resource's reference when extending the example.
- [Vortexify connector and sync contracts](/docs/reference/python/connectors-and-syncs) — Client ownership, discovery, preview isolation, and DataFrame extraction.
- [Vortexify CLI workflow](/docs/reference/cli/overview) — Project paths, resource registration, source commits, and credential configuration.