prod-wag-backend-automate-services-2/api_services/api_controllers

Database Handlers

This directory contains database handlers for MongoDB and PostgreSQL used in the backend automate services.

Overview

The database handlers provide a consistent interface for interacting with different database systems. They implement:

• Connection pooling
• Retry mechanisms
• Error handling
• Thread safety
• Context managers for resource management

MongoDB Handler

The MongoDB handler is implemented as a singleton pattern to ensure efficient connection management across the application. It provides:

• Connection pooling via PyMongo's built-in connection pool
• Automatic retry capabilities for MongoDB operations
• Context manager for MongoDB collections to ensure connections are properly closed
• Thread safety for concurrent operations

MongoDB Performance

The MongoDB handler has been tested with a concurrent load test:

Concurrent Operation Test Results:
Total threads: 100
Passed: 100
Failed: 0
Execution time: 0.73 seconds
Operations per second: 137.61

PostgreSQL Handler

The PostgreSQL handler leverages SQLAlchemy for ORM capabilities and connection management. It provides:

• Connection pooling via SQLAlchemy's connection pool
• ORM models with CRUD operations
• Filter methods for querying data
• Transaction management

PostgreSQL Performance

The PostgreSQL handler has been tested with a concurrent load test:

Concurrent Operation Test Results:
Total threads: 100
Passed: 100
Failed: 0
Execution time: 0.30 seconds
Operations per second: 332.11

Usage Examples

MongoDB Example

from Controllers.Mongo.database import mongo_handler

# Using the context manager for automatic connection management
with mongo_handler.collection("users") as users_collection:
    # Perform operations
    users_collection.insert_one({"name": "John", "email": "john@example.com"})
    user = users_collection.find_one({"email": "john@example.com"})

PostgreSQL Example

from Controllers.Postgres.schema import EndpointRestriction

# Using the session context manager
with EndpointRestriction.new_session() as db_session:
    # Create a new record
    new_endpoint = EndpointRestriction(
        endpoint_code="TEST_API",
        endpoint_name="Test API",
        endpoint_method="GET",
        endpoint_function="test_function",
        endpoint_desc="Test description",
        is_confirmed=True
    )
    new_endpoint.save(db=db_session)
    
    # Query records
    result = EndpointRestriction.filter_one(
        EndpointRestriction.endpoint_code == "TEST_API",
        db=db_session
    ).data

Configuration

Both handlers are configured via environment variables:

MongoDB Configuration

• MONGO_ENGINE: Database engine (mongodb)
• MONGO_HOST: Database host
• MONGO_PORT: Database port
• MONGO_USER: Database username
• MONGO_PASSWORD: Database password
• MONGO_DB: Database name
• MONGO_AUTH_DB: Authentication database

PostgreSQL Configuration

• POSTGRES_ENGINE: Database engine (postgresql+psycopg2)
• POSTGRES_HOST: Database host
• POSTGRES_PORT: Database port
• POSTGRES_USER: Database username
• POSTGRES_PASSWORD: Database password
• POSTGRES_DB: Database name
• POSTGRES_POOL_SIZE: Connection pool size
• POSTGRES_POOL_PRE_PING: Whether to ping the database before using a connection

Testing

Both handlers include comprehensive test suites that verify:

• Basic CRUD operations
• Complex queries
• Nested documents (MongoDB)
• Array operations (MongoDB)
• Aggregation (MongoDB)
• Index operations
• Concurrent operations

To run the tests:

# MongoDB tests
python -m Controllers.Mongo.implementations

# PostgreSQL tests
python -m Controllers.Postgres.implementations