metronome-pulse-sqlite 0.1.0

High-performance SQLite connector for DataPulse - async-first, local storage, and development features

DataPulse SQLite

High-performance, async-first SQLite connector for the DataPulse ecosystem.

DataPulse SQLite provides enterprise-grade connectivity to SQLite databases with advanced features like efficient bulk operations, comprehensive error handling, and local storage optimization. Perfect for development, testing, and local data processing scenarios.

✨ Features

• ⚡ Async-First: Built with async/await patterns for modern Python
• 🔌 Local Storage: Optimized for local development and testing
• 📊 High-Performance Operations: Bulk insert, efficient queries, and custom SQL
• 🔄 Transaction Support: Full ACID compliance with rollback
• 🛡️ Type Safe: Full type hints and runtime validation
• 📈 Performance Monitoring: Built-in metrics and observability
• 🔧 Flexible Configuration: Support for complex operations and custom SQL
• 📋 Schema Management: Automatic table creation and schema validation

🚀 Quick Start

Installation

pip install metronome-pulse-sqlite

Basic Usage

import asyncio
from metronome_pulse_sqlite import SQLitePulse

async def main():
    # Initialize connector
    pulse = SQLitePulse(database_path="my_database.db")
    
    # Connect to database
    await pulse.connect()
    
    try:
        # Simple query
        users = await pulse.query("SELECT * FROM users WHERE active = ?", [True])
        print(f"Found {len(users)} active users")
        
        # Bulk insert
        new_users = [
            {"name": "Alice", "email": "[e-mail]", "active": True},
            {"name": "Bob", "email": "[e-mail]", "active": True}
        ]
        await pulse.write(new_users, "users")
        print("Users inserted successfully")
        
        # Get table information
        table_info = await pulse.get_table_info("users")
        print(f"Table schema: {table_info}")
        
    finally:
        await pulse.close()

# Run the async function
asyncio.run(main())

🔧 Advanced Features

High-Performance Bulk Operations

# Efficient bulk insert
await pulse.write(data, "users", {
    "batch_size": 1000,
    "use_transaction": True
})

# Custom SQL operations
await pulse.execute("""
    INSERT INTO users (name, email, created_at) 
    VALUES (?, ?, ?)
""", ["John", "[e-mail]", "2024-01-01"])

# Bulk copy operations
await pulse.copy_records("users", user_records)

Read-Only Operations

from metronome_pulse_sqlite import SQLiteReadonlyPulse

# Read-only connector for analytics
readonly = SQLiteReadonlyPulse("analytics.db")
await readonly.connect()

# Complex queries
results = await readonly.query_with_params("""
    SELECT 
        user_id,
        COUNT(*) as login_count,
        MAX(login_time) as last_login
    FROM user_logins 
    WHERE login_time >= ? 
    GROUP BY user_id 
    HAVING COUNT(*) > ?
""", ["2024-01-01", 5])

await readonly.close()

Write-Only Operations

from metronome_pulse_sqlite import SQLiteWriteonlyPulse

# Write-only connector for data ingestion
writeonly = SQLiteWriteonlyPulse("data_warehouse.db")
await writeonly.connect()

# High-volume data writing
await writeonly.write(log_data, "event_logs", {
    "batch_size": 5000,
    "use_transaction": True
})

await writeonly.close()

🏗️ Architecture

The SQLite connector follows the DataPulse architecture pattern:

• SQLitePulse: Full-featured connector implementing both read and write operations
• SQLiteReadonlyPulse: Optimized for read-only operations and analytics
• SQLiteWriteonlyPulse: Specialized for high-volume data ingestion

All connectors implement the core DataPulse interfaces:

• Pulse: Base connection management
• Readable: Query and data retrieval operations
• Writable: Data insertion and modification operations

🔧 Configuration

Connection Options

# Basic configuration
pulse = SQLitePulse(database_path="path/to/database.db")

# Advanced configuration with custom settings
pulse = SQLitePulse(
    database_path=":memory:",  # In-memory database
)

Performance Tuning

# Optimize for bulk operations
await pulse.write(data, "table_name", {
    "batch_size": 10000,        # Large batch size for efficiency
    "use_transaction": True,     # Wrap in transaction
    "pragma_settings": {         # SQLite performance pragmas
        "journal_mode": "WAL",
        "synchronous": "NORMAL",
        "cache_size": 10000
    }
})

🧪 Testing

Running Tests

# Install test dependencies
pip install -r requirements-test.txt

# Run all tests
make test

# Run with coverage
make test-cov

# Run specific test types
make test-unit        # Fast unit tests
make test-integration # Slower integration tests

Test Structure

• Unit Tests: Fast, isolated tests for individual components
• Integration Tests: Database interaction tests with proper setup/teardown
• Performance Tests: Benchmark and stress testing

📊 Performance Characteristics

• Bulk Insert: 10,000+ records/second on SSD
• Query Performance: Optimized for analytical workloads
• Memory Usage: Efficient memory management for large datasets
• Concurrent Access: Thread-safe operations with proper locking

🔒 Security Features

• Parameterized Queries: Protection against SQL injection
• Input Validation: Comprehensive data validation
• Error Handling: Secure error messages without information leakage
• Transaction Safety: ACID compliance for data integrity

🤝 Contributing

We welcome contributions! Please see our Contributing Guide for details.

Development Setup

# Clone the repository
git clone https://github.com/datametronome/metronome-pulse-sqlite.git
cd metronome-pulse-sqlite

# Install in development mode
make install-dev

# Run linting and formatting
make lint
make format

# Run tests
make test

📚 API Reference

Core Classes

SQLitePulse

Main connector class implementing full read/write capabilities.

class SQLitePulse(Pulse, Readable, Writable):
    def __init__(self, database_path: str = "datametronome.db")
    async def connect(self) -> None
    async def close(self) -> None
    async def is_connected(self) -> bool

SQLiteReadonlyPulse

Read-only connector optimized for analytics and reporting.

class SQLiteReadonlyPulse(Pulse, Readable):
    async def query(self, query_config: str | dict[str, Any]) -> list
    async def query_with_params(self, sql: str, params: list[Any]) -> list[dict[str, Any]]
    async def get_table_info(self, table_name: str) -> list[dict[str, Any]]
    async def list_tables(self) -> list[str]

SQLiteWriteonlyPulse

Write-only connector optimized for data ingestion.

class SQLiteWriteonlyPulse(Pulse, Writable):
    async def write(self, data: list[dict[str, Any]], config: dict[str, Any] | None = None) -> None
    async def execute(self, sql: str, params: list[Any] | None = None) -> bool
    async def copy_records(self, table_name: str, records: list[dict[str, Any]]) -> bool

📄 License

This project is licensed under the MIT License - see the LICENSE file for details.

🆘 Support

• Documentation: https://datametronome.dev/docs/pulse-sqlite
• Issues: GitHub Issues
• Discussions: GitHub Discussions

🔗 Related Projects

• DataPulse Core - Core interfaces and base classes
• DataPulse PostgreSQL - PostgreSQL connector
• DataMetronome - Main platform