Core services for MongoDB access, error handling, encryption, notifications, and model management.

​
Database Services

​
MongoDBClient

Singleton MongoDB connection manager using environment variables. Environment Variables:
  • DB_USERNAME
  • DB_PASSWORD
  • DB_HOST (default: localhost)
  • DB_PORT (default: 27017)
  • DB_DATABASE (default: customai)
  • DB_CONNECTION (default: mongodb)
Methods:
  • get_database(db_name=None) - Returns MongoDB database instance
db = MongoDBClient().get_database()

​
Database Utilities

​
get_db_instance(db_name=None)

Retrieves singleton database instance across the application. 
db = get_db_instance()

​
mongodb_connection_error_handler(exc)

Global error handler for MongoDB connection issues. 
@app.exception_handler(MongoDBConnectionError)
async def handler(request, exc):
    return mongodb_connection_error_handler(exc)

​
Callback Handlers

​
MongoDBCallbackHandler

Post-processing handler for LLM operations. Methods:
  • on_llm_end(response: LLMResult) - Handles token tracking, memory updates, and chat history
Features:
  • Usage tracking
  • Credit updates
  • Memory management
  • Chat history persistence

​
Encryption Services

​
MessageEncryptor

AES encryption utility for plaintext messages. 
encryptor = MessageEncryptor(key=b'sixteenbytekey123')
encrypted = encryptor.encrypt("plaintext message")
Methods:
  • encrypt(plaintext: str) -> str - Returns Base64-encoded ciphertext

​
MessageDecryptor

AES decryption utility for encrypted messages. 
decryptor = MessageDecryptor(key=b'sixteenbytekey123')
plaintext = decryptor.decrypt("base64_ciphertext")
Methods:
  • decrypt(ciphertext: str) -> str - Returns decrypted plaintext

​
Model Repositories

​
DefaultGeminiModelRepository

Manages default Gemini model configurations per company. Methods:
  • get_default_model_key()
  • get_encrypt_key()
  • get_decrypt_key()

​
DefaultAnthropicModelRepository

Handles default Anthropic model configurations. Methods:
  • get_default_model_key()
  • get_encrypt_key()
  • get_decrypt_key()

​
DefaultOpenAIModelRepository

Manages default OpenAI model configurations. Methods:
  • get_default_model_key()

​
Notification Services

​
EmailService

Email sending abstraction for AWS SES or SMTP. 
service = EmailService()
service.send_email(email, subject, username, content_body, slug, template_name)
Provider selection via EMAIL_PROVIDER environment variable.

​
Firebase

Push notification service using Firebase Cloud Messaging. 
firebase = Firebase()
firebase.send_push_notification(
    tokens=["token1", "token2"], 
    title="Alert", 
    body="Notification message"
)
Methods:
  • send_push_notification(tokens: list, title: str, body: str = None)

​
Security Services

​
LLMAPIKeyDecryptionHandler

Secure API key decryption from MongoDB. 
handler = LLMAPIKeyDecryptionHandler()
handler.initialization("key-id", "keys-collection")
decrypted_key = handler.decrypt()
Methods:
  • initialization(api_key_id: str, collection_name: str)
  • decrypt() -> bytes
Ensure correct decrypt key is configured in environment variables.

​
Streaming Utilities

​
StreamingResponseWithStatusCode

Enhanced FastAPI StreamingResponse with dynamic status code updates. 
return StreamingResponseWithStatusCode(generator_func())
Features:
  • Dynamic status code changes during streaming
  • Mid-stream error handling
  • Compatible with FastAPI streaming responses
Methods:
  • stream_response(send) - Streams content with status code management

​
User Services

​
get_user_data(user_id: str)

Retrieves user profile information by user ID. Returns: Dictionary containing user information 
user_info = get_user_data("user_123")