Version: 1.0.0 Last Updated: 2026-05-14 Status: Active

bast-ai-mongo¶

Version: 0.1.15 Module: bast_ai_mongo Path: src/python/mongo

Overview¶

MongoDB integration library providing connection management, CRUD operations, and query utilities for all Bast AI services that interact with MongoDB.

Features¶

Connection Management¶

Connection pooling
Automatic reconnection
Multiple database support
Atlas and self-hosted support

CRUD Operations¶

Document insert/update/delete
Bulk operations
Upsert support
Query building utilities

Advanced Features¶

Atlas Search integration
Atlas Vector Search integration
Aggregation pipeline support
Index management

Dependencies¶

Internal Dependencies¶

Package	Purpose
bast-ai-config	Configuration management

External Dependencies¶

Package	Version	Purpose
pymongo	^4.13	MongoDB driver

Usage Examples¶

Basic Connection¶

from bast_ai_mongo import MongoClient

client = MongoClient(catname="mycat")
db = client.get_database()
collection = db["users"]

CRUD Operations¶

from bast_ai_mongo import MongoService

mongo = MongoService(catname="mycat")

# Insert
doc_id = mongo.insert_one("users", {"name": "John", "email": "john@example.com"})

# Find
users = mongo.find("users", {"status": "active"}, limit=10)

# Update
mongo.update_one("users", {"_id": doc_id}, {"$set": {"status": "inactive"}})

# Delete
mongo.delete_one("users", {"_id": doc_id})

Bulk Operations¶

# Bulk insert
docs = [{"name": f"User {i}"} for i in range(100)]
mongo.insert_many("users", docs)

# Bulk update
mongo.update_many("users", {"status": "pending"}, {"$set": {"status": "active"}})

Aggregation Pipeline¶

pipeline = [
    {"$match": {"status": "active"}},
    {"$group": {"_id": "$department", "count": {"$sum": 1}}},
    {"$sort": {"count": -1}}
]

results = mongo.aggregate("users", pipeline)

Atlas Search¶

# Text search using Atlas Search
results = mongo.atlas_search(
    collection="documents",
    query="machine learning",
    index_name="default",
    limit=10
)

Atlas Vector Search¶

# Vector similarity search
results = mongo.vector_search(
    collection="documents",
    vector=embedding,  # Your query embedding
    index_name="vector_index",
    num_candidates=100,
    limit=10
)

Configuration¶

# resources/config/{catname}.yaml
mongodb:
  uri: ${MONGODB_URI:mongodb://localhost:27017}
  database: mycat_db
  options:
    maxPoolSize: 50
    minPoolSize: 10
    maxIdleTimeMS: 30000

  # Atlas Search settings
  atlas:
    search_index: default
    vector_index: vector_index

Testing¶

Unit Tests (1 file)¶

test_svc_api.py - Service API tests

Integration Tests (2 files)¶

cat_config.py - CAT configuration tests
auth_user.py - User authentication tests

Running Tests¶

# Unit tests
pants test src/python/mongo:: -- -s --no-header

# Integration tests (requires MongoDB)
pants run src/python/mongo/integration/cat_config.py

Issues & Improvements¶

Current Issues¶

Limited connection pool tuning: Pool settings not exposed
No query caching: Repeated queries hit database
Missing retry logic: Transient failures not handled

Suggested Improvements¶

Connection pool configuration: Expose pool settings in config
Query result caching: Add Redis caching layer
Automatic retries: Add retry logic for transient failures
Query logging: Add optional query logging for debugging
Schema validation: Support MongoDB schema validation
Change streams: Add change stream support for real-time updates

Best Practices¶

Index Management¶

# Create indexes on startup
mongo.create_index("users", [("email", 1)], unique=True)
mongo.create_index("documents", [("created_at", -1)])
mongo.create_text_index("documents", ["content", "title"])

Error Handling¶

from pymongo.errors import DuplicateKeyError, ConnectionFailure

try:
    mongo.insert_one("users", {"email": "test@example.com"})
except DuplicateKeyError:
    logger.warning("User already exists")
except ConnectionFailure:
    logger.error("MongoDB connection failed")

Connection Cleanup¶

# Use context manager
with MongoClient(catname="mycat") as client:
    db = client.get_database()
    # ... operations ...
# Connection automatically closed

Changelog¶

v0.1.15 (Current)¶

Updated pymongo to ^4.13
Improved Atlas Vector Search support

v0.1.14¶

Added Atlas Search integration
Connection pooling improvements

v0.1.13¶

Bulk operation support
Better error handling