Multi-Database Connection Manager

Build a database connection manager that can connect to multiple types of databases, validate connections, and load data with appropriate protocol selection.

Requirements

Implement a module that provides the following functionality:

1. Connection Configuration: Accept database configurations for multiple database types (PostgreSQL, MySQL, SQLite)
2. Connection Validation: Test whether database connections are valid before attempting data loads
3. Protocol Selection: Support configuring different transfer protocols for optimal performance
4. Multi-Database Query Execution: Execute queries across different databases and collect results

Your implementation should:

• Accept connection parameters as dictionaries containing connection strings and optional protocol specifications
• Validate that connections can be established to each database
• Execute a provided SQL query on each valid database connection
• Return results in a structured format indicating which databases succeeded and which failed
• Support protocol overrides for performance optimization (e.g., using 'csv' protocol for bandwidth-constrained scenarios)

Dependencies { .dependencies }

connectorx { .dependency }

Provides high-performance database connectivity and data loading capabilities.

Implementation Details

Create a Python module with a DatabaseConnectionManager class that:

1. Initializes with a list of database configurations
2. Provides a method to validate all connections
3. Provides a method to execute queries across all valid databases
4. Returns results along with connection metadata (database type, success status)

Each database configuration should include:

• name: A friendly name for the database connection
• connection_string: The database connection string
• protocol: (Optional) Protocol to use for data transfer

Test Cases { .test }

Test 1: SQLite Connection { .test_case }

# test_connection_manager.test.py
from connection_manager import DatabaseConnectionManager

def test_sqlite_connection():
    """Test basic SQLite connection and query execution"""
    config = [{
        "name": "test_db",
        "connection_string": "sqlite:///test.db",
    }]

    manager = DatabaseConnectionManager(config)
    assert manager.validate_connections()["test_db"]["is_valid"] == True

Test 2: Multiple Database Connections { .test_case }

# test_connection_manager.test.py
def test_multiple_databases():
    """Test connecting to multiple databases"""
    config = [
        {"name": "db1", "connection_string": "sqlite:///db1.db"},
        {"name": "db2", "connection_string": "sqlite:///db2.db"},
    ]

    manager = DatabaseConnectionManager(config)
    results = manager.validate_connections()

    assert len(results) == 2
    assert all(result["is_valid"] for result in results.values())

Test 3: Protocol Specification { .test_case }

# test_connection_manager.test.py
def test_protocol_specification():
    """Test that protocol can be specified for connections"""
    config = [{
        "name": "test_db",
        "connection_string": "sqlite:///test.db",
        "protocol": "binary"
    }]

    manager = DatabaseConnectionManager(config)
    assert manager.config[0]["protocol"] == "binary"

Test 4: Invalid Connection Handling { .test_case }

# test_connection_manager.test.py
def test_invalid_connection():
    """Test handling of invalid connection strings"""
    config = [{
        "name": "bad_db",
        "connection_string": "invalid://connection",
    }]

    manager = DatabaseConnectionManager(config)
    results = manager.validate_connections()

    assert results["bad_db"]["is_valid"] == False
    assert "error" in results["bad_db"]

Expected Outputs

Your implementation should:

• Successfully validate and connect to multiple databases
• Handle connection errors gracefully without crashing
• Support different protocols when specified
• Return structured results indicating success/failure for each connection