database-engine-implementation

Overview

Database engines provide read-only access to SQL and NoSQL databases as iterable data sources. They wrap database drivers to provide a unified BaseIterable interface.

Architecture

• DBDriver (iterable/db/base.py) - Abstract base class for database drivers
• DatabaseIterable (iterable/db/iterable.py) - Wrapper that makes DBDriver work as BaseIterable
• Driver Registry (iterable/db/__init__.py) - Registry for mapping engine names to driver classes

Adding New Database Engines

Step-by-Step Process

1. Create driver file: iterable/db/<engine>.py
2. Implement DBDriver: Inherit from DBDriver in iterable/db/base.py
3. Required methods: connect(), iterate(), close() (close has default implementation)
4. Register driver: Call register_driver() in iterable/db/__init__.py
5. Add detection: Update iterable/helpers/detect.py to recognize database URLs
6. Create tests: tests/test_db_engines.py or add to existing test file
7. Update dependencies: Add optional dependency to pyproject.toml
8. Update documentation: Add engine to docs

Implementation Pattern

from collections.abc import Iterator
from typing import Any
from ..types import Row
from .base import DBDriver

class NewEngineDriver(DBDriver):
    """Database driver for NewEngine.
    
    Supports streaming queries using batch processing.
    """
    
    def __init__(self, source: str | Any, **kwargs: Any) -> None:
        """Initialize driver.
        
        Args:
            source: Connection string/URL or existing connection object
            **kwargs: Additional parameters:
                - query: Query string or table name
                - batch_size: Rows per batch (default: 10000)
                - on_error: Error handling policy ('raise', 'skip', 'warn')
        """
        super().__init__(source, **kwargs)
        self._cursor: Any = None
    
    def connect(self) -> None:
        """Establish database connection.
        
        Raises:
            ImportError: If required driver library is not installed
            ConnectionError: If connection fails
        """
        try:
            import database_library
        except ImportError:
            raise ImportError(
                "database-library is required. Install with: pip install database-library"
            ) from None
        
        # Handle existing connection object
        if hasattr(self.source, "cursor"):
            self.conn = self.source
            self._connected = True
            return
        
        # Parse connection string
        if not isinstance(self.source, str):
            raise ValueError("Source must be connection string or connection object")
        
        try:
            self.conn = database_library.connect(self.source, **self.kwargs.get("connect_args", {}))
            self._connected = True
        except Exception as e:
            self._connected = False
            raise ConnectionError(f"Failed to connect: {e}") from e
    
    def iterate(self) -> Iterator[Row]:
        """Return iterator of dict rows.
        
        Yields:
            dict: Database row as dictionary
            
        Raises:
            RuntimeError: If not connected
        """
        if not self._connected:
            raise RuntimeError("Not connected. Call connect() first.")
        
        self._start_metrics()
        batch_size = self.kwargs.get("batch_size", 10000)
        
        try:
            query = self._build_query()
            cursor = self.conn.cursor()
            
            # Execute query with batching
            cursor.execute(query)
            
            while True:
                rows = cursor.fetchmany(batch_size)
                if not rows:
                    break
                
                # Convert rows to dicts
                column_names = [desc[0] for desc in cursor.description]
                for row in rows:
                    row_dict = dict(zip(column_names, row))
                    self._update_metrics(rows_read=1)
                    yield row_dict
                    
        except Exception as e:
            self._handle_error(e, "during iteration")
            if self._on_error == "raise":
                raise

Key Implementation Details

Connection Handling

• Support both connection strings and existing connection objects
• Parse connection strings appropriately (URL format)
• Handle connection errors gracefully with clear messages
• Set self._connected = True on success

Query Building

• Accept query parameter (SQL query or table name)
• If table name provided, build SELECT query
• Support filtering, projection, sorting via kwargs
• Use parameterized queries to prevent SQL injection

Iteration

• Use batch processing (batch_size parameter)
• Convert database rows to dict objects
• Update metrics: self._update_metrics(rows_read=1)
• Handle errors according to on_error policy
• Start metrics: self._start_metrics() before iteration

Error Handling

Three policies (set via on_error kwarg):

• 'raise' - Raise exceptions (default)
• 'skip' - Skip problematic rows
• 'warn' - Warn and continue

Use self._handle_error(error, context) for consistent handling.

Metrics Tracking

Automatic metrics available via self.metrics:

• rows_read - Number of rows read
• bytes_read - Bytes read (may be None)
• elapsed_seconds - Time elapsed

Update during iteration: self._update_metrics(rows_read=count)

Registration

Register driver in iterable/db/__init__.py:

from .newengine import NewEngineDriver

register_driver("newengine", NewEngineDriver)

Format Detection

Update iterable/helpers/detect.py to recognize database URLs:

def detect_file_type(filename, content=None):
    # Database URL detection
    if filename.startswith(("postgresql://", "postgres://")):
        return "database"
    if filename.startswith("newengine://"):
        return "database"
    # ... existing detection

Testing Requirements

Test Structure

import pytest
from iterable.db.newengine import NewEngineDriver

class TestNewEngineDriver:
    def test_connect(self):
        driver = NewEngineDriver("newengine://localhost/db")
        driver.connect()
        assert driver.is_connected
        driver.close()
    
    def test_iterate(self):
        driver = NewEngineDriver("newengine://localhost/db", query="SELECT * FROM table")
        driver.connect()
        rows = list(driver.iterate())
        assert len(rows) > 0
        assert isinstance(rows[0], dict)
        driver.close()
    
    def test_batch_processing(self):
        driver = NewEngineDriver("newengine://localhost/db", query="SELECT * FROM table", batch_size=100)
        driver.connect()
        # Verify batching works correctly
        driver.close()
    
    def test_error_handling(self):
        driver = NewEngineDriver("newengine://invalid", query="SELECT * FROM table", on_error="skip")
        # Test error handling policies

Test Coverage

• Connection with string and object sources
• Query building (table name vs SQL)
• Batch processing
• Error handling policies
• Metrics tracking
• Context manager support (with statement)
• Missing dependencies (should raise ImportError)

Dependencies

Optional Dependencies

Add to pyproject.toml:

[project.optional-dependencies]
newengine = ["database-library>=1.0.0"]

Import Handling

Handle missing dependencies:

try:
    import database_library
except ImportError:
    raise ImportError(
        "newengine support requires 'database-library'. "
        "Install with: pip install iterabledata[newengine]"
    ) from None

Examples

Look at existing implementations:

• iterable/db/postgres.py - SQL database example (PostgreSQL)
• iterable/db/mongo.py - NoSQL database example (MongoDB)
• iterable/db/clickhouse.py - Columnar database example
• iterable/db/elasticsearch.py - Search engine example

Common Patterns

SQL Databases

• Use server-side cursors for memory efficiency
• Support read-only transactions
• Handle connection pooling
• Use parameterized queries

NoSQL Databases

• Support query filters and projections
• Handle pagination/cursors
• Support aggregation pipelines
• Batch document retrieval

Connection String Parsing

from urllib.parse import urlparse

def parse_connection_string(conn_str: str) -> dict:
    parsed = urlparse(conn_str)
    return {
        "host": parsed.hostname,
        "port": parsed.port,
        "database": parsed.path.lstrip("/"),
        "username": parsed.username,
        "password": parsed.password,
    }

Common Pitfalls

• Memory issues: Always use batch processing, don't load all rows at once
• Connection leaks: Always close connections, use context managers
• Error handling: Implement all three error policies consistently
• Metrics: Update metrics during iteration, not after
• Reset support: Most databases don't support reset (set _reset_supported = False)
• Read-only: Default to read-only access for safety