purreal 0.1.2

Production-grade async connection pooler for SurrealDB

Purreal

Quick Test (After Git Pull)

# 1. Start SurrealDB
surreal start --bind 0.0.0.0:8000 --user root --pass root

# 2. Install purreal
pip install -e .

# 3. Test connectivity (takes 5 seconds)
python tests/test_connectivity.py

# 4. Test 500 concurrent connections (optional)
python tests/stress_test_simple.py 500

Quick test runner:

# Linux/Mac
./test.sh

# Windows
test.bat

See QUICKSTART.md for full setup guide.

Production-Grade SurrealDB Connection Pooling

Overview

Purreal is a production-grade async connection pooler for SurrealDB that solves the critical websockets.exceptions.ConcurrencyError: cannot call recv while another coro is calling recv issue in high-concurrency Python applications.

By ensuring exclusive connection leasing and sophisticated connection lifecycle management, Purreal enables your async applications to safely handle thousands of concurrent database operations without race conditions or connection conflicts.

Why Purreal?

The Problem: SurrealDB's Python client connections cannot safely handle concurrent operations from multiple coroutines. Attempting to share a single connection results in ConcurrencyError crashes.

The Solution: Purreal provides:

• Exclusive connection leasing - each coroutine gets its own connection
• Automatic pool management - connections created/destroyed based on demand
• Robust error handling - connections auto-replaced on failure
• Production-ready - battle-tested with comprehensive logging and stats

Key Features

Core Pooling

• Exclusive Connection Leasing: Prevents ConcurrencyError by ensuring only one coroutine uses a connection at a time
• Dynamic Pool Sizing: Auto-scales between min_connections and max_connections based on load
• Connection Queueing: Fair FIFO queue when pool exhausted, with configurable timeout
• Async Context Manager: Safe async with pattern guarantees connection return

Reliability & Health

• Automatic Health Checks: Background maintenance loop validates connection health
• Connection Lifecycle Management: Tracks usage count, age, and health status
• Intelligent Retry Logic: Exponential backoff with jitter for transient failures
• Graceful Degradation: Continues operating even when some connections fail

Production Features

• Connection State Reset: Optional reset_on_return to clear session state
• Schema Initialization: Auto-execute .surql schema files on connection creation
• Comprehensive Stats: Track acquisitions, timeouts, errors, peak usage
• Detailed Logging: Debug-level connection lifecycle tracing
• Query Logging: Optional request/response logging for debugging

Installation

pip install purreal

Requirements:

• Python 3.11+
• surrealdb >= 0.3.0

Quick Start

Basic Usage

import asyncio
from purreal import SurrealDBConnectionPool

async def main():
    # Initialize pool
    pool = SurrealDBConnectionPool(
        uri="ws://localhost:8000/rpc",
        credentials={"username": "root", "password": "root"},
        namespace="test",
        database="test",
        min_connections=5,
        max_connections=20,
    )
    
    # Use async context manager (handles init + cleanup)
    async with pool:
        # Acquire exclusive connection
        async with pool.acquire() as conn:
            result = await conn.query("SELECT * FROM users")
            print(result)

asyncio.run(main())

Production Example with Error Handling

import asyncio
import logging
from purreal import SurrealDBConnectionPool

logging.basicConfig(level=logging.INFO)

async def process_user_batch(pool, user_ids):
    """Process multiple users concurrently using the pool."""
    async def process_user(user_id):
        try:
            async with pool.acquire() as conn:
                # Each coroutine gets exclusive connection access
                user = await conn.query(f"SELECT * FROM user:{user_id}")
                await conn.query(
                    f"UPDATE user:{user_id} SET last_accessed = time::now()"
                )
                return user
        except asyncio.TimeoutError:
            logging.warning(f"Timeout acquiring connection for user {user_id}")
            return None
        except Exception as e:
            logging.error(f"Error processing user {user_id}: {e}")
            return None
    
    # Process all users concurrently - pool handles queueing
    results = await asyncio.gather(*[process_user(uid) for uid in user_ids])
    return [r for r in results if r is not None]

async def main():
    pool = SurrealDBConnectionPool(
        uri="wss://mydb.surreal.cloud",
        credentials={"username": "admin", "password": "secure_pass"},
        namespace="production",
        database="app",
        min_connections=10,
        max_connections=50,
        acquisition_timeout=30.0,  # Wait up to 30s for connection
        health_check_interval=60.0,  # Check health every 60s
        log_queries=True,  # Log all queries (disable in prod)
        schema_file="schema.surql",  # Auto-execute on new connections
    )
    
    async with pool:
        # Simulate high-concurrency workload
        user_ids = [f"user_{i}" for i in range(100)]
        results = await process_user_batch(pool, user_ids)
        
        # Check pool stats
        stats = await pool.get_stats()
        logging.info(f"Pool stats: {stats}")

asyncio.run(main())

Configuration Options

Required Parameters

• uri (str): SurrealDB connection URI (e.g., ws://localhost:8000/rpc, wss://cloud.surreal.io)
• credentials (dict): Authentication credentials {"username": "...", "password": "..."}
• namespace (str): SurrealDB namespace
• database (str): SurrealDB database name

Pool Sizing

• min_connections (int, default: 4): Minimum connections maintained in pool
• max_connections (int, default: 10): Maximum connections allowed
• acquisition_timeout (float, default: 10.0): Seconds to wait for available connection

Connection Lifecycle

• max_idle_time (float, default: 300.0): Seconds before idle connection is recycled
• max_usage_count (int, default: 1000): Max queries before connection recycled
• connection_timeout (float, default: 25.0): Seconds to establish new connection
• connection_retry_attempts (int, default: 3): Retries for failed connections
• connection_retry_delay (float, default: 1.0): Base delay between retries (exponential backoff)

Health & Maintenance

• health_check_interval (float, default: 30.0): Seconds between health checks
• reset_on_return (bool, default: True): Reset connection state on release

Advanced

• schema_file (str, optional): Path to .surql file to execute on new connections
• on_connection_create (callable, optional): Async callback when connection created
• log_queries (bool, default: False): Log all queries (useful for debugging)

API Reference

SurrealDBConnectionPool

Initialization

pool = SurrealDBConnectionPool(
    uri, credentials, namespace, database,
    min_connections=4, max_connections=10,
    acquisition_timeout=10.0, **kwargs
)

Context Manager (Recommended)

async with pool:  # Calls initialize() and close() automatically
    async with pool.acquire() as conn:
        await conn.query("SELECT * FROM table")

Methods

async with acquire() -> SurrealConnection

• Acquires exclusive connection from pool
• Must use with async with to ensure proper release
• Raises asyncio.TimeoutError if pool exhausted for acquisition_timeout seconds
• Raises RuntimeError if pool closed or not initialized

async initialize()

• Creates minimum connections and starts maintenance loop
• Safe to call multiple times (idempotent)
• Automatically called by async with pool

async close()

• Gracefully closes all connections
• Cancels maintenance loop
• Notifies waiting coroutines
• Safe to call multiple times
• Automatically called by async with pool exit

async get_stats() -> dict

• Returns pool statistics:

  {
      "total_connections_created": int,
      "total_connections_closed": int,
      "total_acquisitions": int,
      "total_releases": int,
      "acquisition_timeouts": int,
      "connection_errors": int,
      "health_check_failures": int,
      "peak_connections": int,
      "peak_waiters": int,
      "current_pool_size": int,
      "current_available": int,
      "current_in_use": int,
      "current_waiters": int,
  }
  

async execute_query(query: str, params: dict = None) -> Any

• Convenience method: acquires connection, executes query, releases
• Equivalent to:

  async with pool.acquire() as conn:
      return await conn.query(query, params)
  

Best Practices

✅ DO

# Use async context managers for guaranteed cleanup
async with pool:
    async with pool.acquire() as conn:
        await conn.query("SELECT * FROM users")

# Set appropriate pool sizes for your workload
pool = SurrealDBConnectionPool(
    min_connections=10,  # Based on baseline load
    max_connections=50,  # Based on peak load
)

# Handle timeouts gracefully
try:
    async with pool.acquire() as conn:
        result = await conn.query("SLOW QUERY")
except asyncio.TimeoutError:
    logger.warning("Pool exhausted, consider scaling")

❌ DON'T

# Don't share connections between coroutines
conn = await pool.acquire()  # Missing async with!
await asyncio.gather(
    conn.query("SELECT 1"),  # ❌ ConcurrencyError!
    conn.query("SELECT 2"),
)

# Don't forget to close the pool
pool = SurrealDBConnectionPool(...)
await pool.initialize()
# ... use pool ...
# ❌ Missing: await pool.close()

# Don't set pool sizes too small
pool = SurrealDBConnectionPool(
    min_connections=1,   # ❌ Too small for production
    max_connections=2,   # ❌ Will bottleneck quickly
)

Troubleshooting

ConcurrencyError: cannot call recv

Cause: Sharing a connection between coroutines
Solution: Always use async with pool.acquire() - never store or share the connection object

asyncio.TimeoutError during acquisition

Cause: Pool exhausted (all connections in use)
Solutions:

• Increase max_connections
• Increase acquisition_timeout
• Reduce query execution time
• Check for connection leaks (not releasing connections)

Connections failing health checks

Cause: Network issues, database restart, or connection timeout
Solution: Purreal auto-replaces unhealthy connections. Check logs for patterns:

# Enable debug logging to diagnose
logging.getLogger('purreal').setLevel(logging.DEBUG)

Memory usage growing

Cause: Connections not being released
Solution: Verify all pool.acquire() uses are in async with blocks

Performance Tips

Pool Sizing

# Calculate based on your workload
concurrent_requests = 100  # Peak concurrent users
avg_query_time = 0.1       # Average query time in seconds
connections_needed = concurrent_requests * avg_query_time

pool = SurrealDBConnectionPool(
    min_connections=int(connections_needed * 0.5),  # 50% for baseline
    max_connections=int(connections_needed * 2),    # 200% for peaks
)

Connection Lifecycle

# Tune based on your database workload
pool = SurrealDBConnectionPool(
    max_usage_count=10000,      # Higher for read-heavy workloads
    max_idle_time=600,          # Longer for stable connections
    health_check_interval=60,   # More frequent for critical apps
)

Monitoring

# Periodically check pool health
async def monitor_pool(pool):
    while True:
        stats = await pool.get_stats()
        if stats['acquisition_timeouts'] > 10:
            logger.warning(f"High timeout rate: {stats}")
        if stats['in_use_connections'] / stats['current_connections'] > 0.8:
            logger.warning("Pool utilization above 80%")
        await asyncio.sleep(60)

asyncio.create_task(monitor_pool(pool))

Known Limitations

Burst Load > max_connections

Issue: When burst traffic exceeds max_connections, waiting tasks may timeout instead of queuing properly.

Example:

• Pool: max_connections=50
• Burst: 100 concurrent requests
• Result: 50 succeed, 50 timeout (instead of queuing)

Workaround:

# Size pool for peak burst load
pool = SurrealDBConnectionPool(
    max_connections=150,  # 1.5x expected peak
    ...
)

# Or use semaphore to limit concurrency
semaphore = asyncio.Semaphore(50)

async def limited_query():
    async with semaphore:
        async with pool.acquire() as conn:
            await conn.query(...)

Status: Known issue in v0.1.0. See KNOWN_ISSUES.md for details.

Impact: Sustained load and gradual ramp-up work fine. Only affects sudden bursts >> pool size.

Contributing

We welcome contributions! Please feel free to submit a Pull Request.

Guidelines

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Make your changes
4. Write tests for your changes (pytest tests/)
5. Ensure all tests pass (pytest)
6. Submit a pull request

Development Setup

git clone https://github.com/dyleeeeeeee/purreal.git
cd purreal
pip install -e ".[dev]"
pytest

License

This project is licensed under the GNU General Public License v3 (GPLv3) - see the LICENSE file for details.

Acknowledgments

• Solves the critical ConcurrencyError issue affecting SurrealDB Python applications
• Built for production use in high-concurrency async applications
• Thanks to the SurrealDB team for an excellent database