A comprehensive business intelligence platform with unified database connectivity.

Unified Database Connector Library

A powerful, unified interface for connecting to and querying multiple database types including SQL, NoSQL, and key-value stores.

🚀 Features

• Unified Interface: Single API for all database operations
• Multiple Database Support:
  • SQL: PostgreSQL, MySQL, SQLite
  • NoSQL: MongoDB, Elasticsearch
  • Key-Value: Redis
  • Analytics: ClickHouse
• Query Builder: High-level abstraction for common operations
• Connection Management: Built-in connection pooling and context managers
• Type Safety: Strongly typed with proper error handling
• Transaction Support: ACID transactions for SQL databases
• Flexible Queries: Support for both raw queries and ORM-like operations

📦 Installation

Install the package with database drivers:

# Install all database drivers
uv sync --group all-databases

# Or install specific drivers
uv add psycopg2-binary  # PostgreSQL
uv add pymysql          # MySQL
uv add pymongo          # MongoDB
uv add redis            # Redis
uv add clickhouse-driver # ClickHouse
uv add elasticsearch    # Elasticsearch
# Note: SQLite is included in Python standard library

🎯 Quick Start

Basic Connection

from datasource import create_connection

# Connect to PostgreSQL
conn = create_connection(
    'postgresql',
    host='localhost',
    database='mydb',
    username='user',
    password='password'
)

# Use with context manager (auto-connect/disconnect)
with conn:
    results = conn.fetch_all("SELECT * FROM users")
    print(results)

Using Query Builder

from datasource import create_connection, QueryBuilder

conn = create_connection('postgresql', host='localhost', database='mydb')

with conn:
    qb = QueryBuilder(conn)
    
    # SELECT with conditions
    result = qb.select(
        'users',
        columns=['id', 'name', 'email'],
        where={'active': True},
        limit=10,
        order_by='created_at DESC'
    )
    print(result.data)
    
    # INSERT
    qb.insert('users', {
        'name': 'John Doe',
        'email': 'john@example.com'
    })
    
    # UPDATE
    qb.update('users', 
              data={'status': 'active'},
              where={'email': 'john@example.com'})
    
    # DELETE
    qb.delete('users', where={'status': 'inactive'})
    
    # COUNT
    count = qb.count('users', where={'age': {'$gte': 18}})

💾 Database-Specific Examples

PostgreSQL

from datasource import create_connection

conn = create_connection('postgresql', 
                        host='localhost',
                        database='testdb')

with conn:
    # Get schema information
    tables = conn.get_tables()
    columns = conn.get_columns('users')
    
    # Transaction support
    with conn.transaction():
        conn.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
        conn.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
    
    # Parameterized queries
    results = conn.fetch_all(
        "SELECT * FROM users WHERE age > %s",
        (25,)
    )

MySQL

conn = create_connection('mysql', 
                        host='localhost',
                        database='testdb')

with conn:
    # Create table
    conn.create_table('users', {
        'id': 'INT AUTO_INCREMENT',
        'name': 'VARCHAR(100)',
        'email': 'VARCHAR(100) UNIQUE'
    }, primary_key='id', engine='InnoDB')
    
    # Batch insert
    users = [
        ('Alice', 'alice@example.com'),
        ('Bob', 'bob@example.com')
    ]
    conn.execute_many(
        "INSERT INTO users (name, email) VALUES (%s, %s)",
        users
    )

SQLite

conn = create_connection('sqlite', database='./mydb.db')

with conn:
    # Create table
    conn.execute("""
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            email TEXT UNIQUE
        )
    """)
    
    # Insert and query
    conn.execute("INSERT INTO users (name, email) VALUES (?, ?)",
                ('John', 'john@example.com'))
    
    users = conn.fetch_all("SELECT * FROM users")
    
    # Database utilities
    conn.backup('./backup.db')
    conn.vacuum()  # Optimize database

MongoDB

conn = create_connection('mongodb',
                        host='localhost',
                        database='testdb')

with conn:
    # Insert documents
    conn.insert_one('users', {
        'name': 'John Doe',
        'age': 30,
        'tags': ['python', 'mongodb']
    })
    
    # Find documents
    users = conn.find_many('users', 
                          {'age': {'$gte': 25}},
                          limit=10)
    
    # Update
    conn.update_many('users',
                    filter_dict={'status': 'pending'},
                    update_dict={'$set': {'status': 'active'}})
    
    # Aggregation
    pipeline = [
        {'$match': {'age': {'$gte': 18}}},
        {'$group': {'_id': '$country', 'count': {'$sum': 1}}}
    ]
    results = conn.aggregate('users', pipeline)

Redis

conn = create_connection('redis', host='localhost')

with conn:
    # Key-value operations
    conn.set('user:1:name', 'John Doe')
    conn.set('session:abc', 'data', expire=3600)
    
    name = conn.get('user:1:name')
    
    # Hash operations
    conn.hset('user:2', 'name', 'Alice')
    conn.hset('user:2', 'age', '25')
    user = conn.hgetall('user:2')
    
    # List operations
    conn.lpush('queue', 'task1', 'task2')
    tasks = conn.lrange('queue', 0, -1)
    
    # Set operations
    conn.sadd('tags', 'python', 'redis')
    tags = conn.smembers('tags')
    
    # Sorted set (leaderboard)
    conn.zadd('scores', {'player1': 100, 'player2': 200})
    top = conn.zrange('scores', 0, 9, withscores=True)

ClickHouse

conn = create_connection('clickhouse',
                        host='localhost',
                        database='default')

with conn:
    # Create table
    conn.create_table('events', {
        'date': 'Date',
        'user_id': 'UInt32',
        'event_type': 'String',
        'value': 'Float32'
    }, engine='MergeTree()', order_by='(date, user_id)')
    
    # Batch insert
    events = [
        ('2024-01-01', 1, 'click', 1.0),
        ('2024-01-02', 2, 'view', 1.0)
    ]
    conn.execute_many(
        "INSERT INTO events VALUES",
        events
    )
    
    # Analytics query
    result = conn.fetch_all("""
        SELECT 
            event_type,
            count() as count,
            avg(value) as avg_value
        FROM events
        GROUP BY event_type
    """)
    
    # Optimize table
    conn.optimize_table('events')

Elasticsearch

conn = create_connection('elasticsearch',
                        host='localhost',
                        port=9200)

with conn:
    # Create index
    conn.create_index('products', mappings={
        'properties': {
            'name': {'type': 'text'},
            'price': {'type': 'float'},
            'category': {'type': 'keyword'}
        }
    })
    
    # Insert documents
    conn.insert_one('products', {
        'name': 'Laptop',
        'price': 999.99,
        'category': 'electronics'
    })
    
    # Full-text search
    results = conn.search('products', {
        'match': {'name': 'laptop'}
    }, size=10)
    
    # Aggregations
    agg = conn.aggregate('products', {
        'avg_price': {'avg': {'field': 'price'}},
        'categories': {'terms': {'field': 'category'}}
    })

🔧 Advanced Features

Managing Multiple Connections

from datasource import DataSourceManager

manager = DataSourceManager()

# Add connections
manager.add_connection('pg', 'postgresql', {
    'host': 'localhost',
    'database': 'prod_db'
})

manager.add_connection('mongo', 'mongodb', {
    'host': 'localhost',
    'database': 'analytics'
})

# Use all connections
with manager:
    pg_conn = manager.get_connection('pg')
    mongo_conn = manager.get_connection('mongo')
    
    # Query both databases
    users = pg_conn.fetch_all("SELECT * FROM users")
    events = mongo_conn.find_many('events', limit=100)

# All connections automatically closed

Factory Pattern

from datasource import DataSourceFactory, ConnectionConfig, DatabaseType

# Create connection using factory
config = ConnectionConfig(
    host='localhost',
    database='mydb',
    username='user',
    password='password'
)

conn = DataSourceFactory.create(DatabaseType.POSTGRESQL, config)

# List supported databases
databases = DataSourceFactory.get_supported_databases()
print(databases)
# ['postgresql', 'mysql', 'sqlite', 'mongodb', 'redis', 'clickhouse', 'elasticsearch']

Error Handling

from datasource import create_connection

try:
    conn = create_connection('postgresql', host='localhost')
    
    # Test connection
    if conn.test_connection():
        print("Connected successfully!")
    
    with conn:
        results = conn.fetch_all("SELECT * FROM users")
        
except Exception as e:
    print(f"Database error: {e}")
finally:
    if conn.is_connected():
        conn.disconnect()

📚 API Reference

Base Classes

• BaseConnector: Abstract base for all connectors
• SQLConnector: Base for SQL databases (PostgreSQL, MySQL, SQLite, ClickHouse)
• NoSQLConnector: Base for NoSQL databases (MongoDB, Elasticsearch)
• KeyValueConnector: Base for key-value stores (Redis)

Core Methods

All connectors implement:

• connect(): Establish connection
• disconnect(): Close connection
• is_connected(): Check connection status
• execute(query, params): Execute query
• execute_many(query, params_list): Batch execution
• fetch_one(query, params): Fetch single row
• fetch_all(query, params): Fetch all rows
• fetch_many(query, size, params): Fetch N rows

SQL-Specific Methods

• get_tables(): List tables
• get_columns(table): Get column info
• table_exists(table): Check if table exists
• create_table(table, columns, **options): Create table
• drop_table(table): Drop table

NoSQL-Specific Methods

• get_collections(): List collections/indices
• collection_exists(name): Check if collection exists
• insert_one(collection, document): Insert document
• insert_many(collection, documents): Batch insert
• find_one(collection, filter): Find document
• find_many(collection, filter, limit): Find documents
• update_one/update_many(): Update documents
• delete_one/delete_many(): Delete documents

🧪 Running Examples

See comprehensive examples in src/examples.py:

python src/examples.py

🤝 Contributing

Contributions are welcome! Areas for improvement:

• Additional database connectors
• Enhanced query builder features
• Performance optimizations
• More comprehensive tests

📄 License

This project is part of YAPBI (Yet Another Power BI) platform.

🔗 Related

• Architecture Documentation
• API Reference
• Examples

---

Note: Make sure to install the appropriate database drivers for the databases you want to use. See the Installation section for details.