MongoDB connector
Document-oriented database for collections, document queries, and change streams.
Use this as an implementation starting point: adapt the worked example to the requested dataset and the source documentation. See Custom API for the shared SDK pattern and document-upload workflow.
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.
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 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
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
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 for registration and configuration.
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
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.
vortexify==0.1.1
pymongo>=4.10,<5
pandas>=2,<3PyMongo supplies the MongoDB driver. The URI, TLS trust, DNS/SRV support, and network rules must match the deployment.
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
This file contains a JSON Schema, not actual credential values. Provide those through platform configuration.
{
"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
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 = NoneSync schema: syncs/snapshot/config.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
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
Register the connector before the sync and commit each resource separately:
vortexify create connector mongodb
vortexify create sync mongodb/snapshotEdit the files, inspect vortexify status, and use the CLI guide 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
- Vendor API and data reference — Resource definitions, authentication, fields, and source-specific limits. Check the selected resource's reference when extending the example.
- Vortexify connector and sync contracts — Client ownership, discovery, preview isolation, and DataFrame extraction.
- Vortexify CLI workflow — Project paths, resource registration, source commits, and credential configuration.