WorkerSQL Python SDK

WorkerSQL Python SDK

A Python SDK for WorkerSQL - bringing MySQL-compatible database operations to the edge with Cloudflare Workers.

Features

• 🚀 Edge-Native: Run SQL queries at the edge for ultra-low latency
• 🔒 Secure: Built-in SQL injection prevention and schema validation
• 📊 MySQL Compatible: Familiar SQL syntax with MySQL compatibility
• 🔄 Async Support: Modern async/await support with httpx
• 📝 Type Safe: Full type hints and Pydantic validation
• 🧪 Well Tested: Comprehensive test coverage with pytest
• 📚 Well Documented: Complete API documentation and examples

Installation

From PyPI (Recommended)

pip install workersql-python-sdk

From Source

git clone https://github.com/healthfees-org/workersql.git
cd workersql/sdk/python
pip install -e .

Development Installation

git clone https://github.com/healthfees-org/workersql.git
cd workersql/sdk/python
pip install -e ".[dev]"

Quick Start

Using DSN String

from workersql_client import WorkerSQLClient

# Connect using DSN
client = WorkerSQLClient(dsn='workersql://username:[email protected]:443/mydb?apiKey=your-key')

# Execute a query
result = client.query("SELECT * FROM users WHERE id = ?", [1])
print(f"User: {result.data}")

# Close the connection
client.close()

Using Configuration Dict

from workersql_client import WorkerSQLClient

# Initialize client
config = {
    "host": "api.workersql.com",
    "port": 443,
    "database": "mydb",
    "username": "myuser",
    "password": "mypass",
    "api_key": "your-api-key",
    "ssl": True,
    "pooling": {
        "enabled": True,
        "min_connections": 2,
        "max_connections": 10
    }
}

with WorkerSQLClient(config=config) as client:
    # Execute queries
    users = client.query("SELECT * FROM users")
    print(f"Users: {users.data}")

    # Execute batch queries
    queries = [
        {"sql": "INSERT INTO users (name, email) VALUES (?, ?)", "params": ["John Doe", "[email protected]"]},
        {"sql": "SELECT * FROM users WHERE email = ?", "params": ["[email protected]"]},
    ]
    batch_result = client.batch_query(queries)
    print(f"Batch result: {batch_result}")

Configuration

Environment Variables

Create a .env file in your project root:

WORKERSQL_API_ENDPOINT=https://your-endpoint.com/api
WORKERSQL_API_KEY=your-api-key
WORKERSQL_HOST=your-database-host
WORKERSQL_USERNAME=your-username
WORKERSQL_PASSWORD=your-password
WORKERSQL_DATABASE=your-database
WORKERSQL_SSL=true
WORKERSQL_TIMEOUT=30000

Programmatic Configuration

from workersql_client import WorkerSQLClient

config = {
    "api_endpoint": "https://your-endpoint.com/api",
    "api_key": "your-api-key",  # Optional
    "host": "localhost",
    "username": "user",
    "password": "password",
    "database": "mydb",
    "port": 3306,
    "ssl": True,
    "timeout": 30000,
    "retry_attempts": 3,
    "retry_delay": 1.0,
}

API Reference

WorkerSQLClient

Methods

• query(sql, params=None, timeout=None, cache=None) - Execute a single SQL query
• batch_query(queries, transaction=False, stop_on_error=True) - Execute multiple queries
• health_check() - Check service health
• close() - Close the client connection

Query Parameters

• sql: SQL query string
• params: Query parameters (list)
• timeout: Query timeout in milliseconds (default: 30000)
• cache: Cache options (optional)

Cache Options

cache_options = {
    "enabled": True,
    "ttl": 300,  # seconds
    "key": "custom-cache-key"  # optional
}

Error Handling

from workersql_client import WorkerSQLClient, ValidationError

try:
    with WorkerSQLClient(config) as client:
        result = client.query("SELECT * FROM invalid_table")
except ValidationError as e:
    print(f"Validation error: {e.code} - {e.message}")
except Exception as e:
    print(f"Unexpected error: {e}")

Async Support

import asyncio
import httpx
from workersql_client import WorkerSQLClient

async def async_query():
    # Using httpx for async HTTP requests
    async with httpx.AsyncClient() as http_client:
        # Configure client to use async HTTP client
        config["http_client"] = http_client
        client = WorkerSQLClient(config)

        result = await client.query_async("SELECT * FROM users")
        print(f"Async result: {result}")

asyncio.run(async_query())

Schema Validation

The SDK includes built-in schema validation using JSON Schema:

from workersql_client import SchemaValidator

# Validate query request
request = {
    "sql": "SELECT * FROM users",
    "params": [],
    "timeout": 30000
}

try:
    validated = SchemaValidator.validate_query_request(request)
    print("Request is valid!")
except ValidationError as e:
    print(f"Invalid request: {e}")

Development

Running Tests

# Run all tests
pytest

# Run with coverage
pytest --cov=workersql_client

# Run specific test file
pytest tests/test_client.py

Code Quality

# Format code
black .

# Sort imports
isort .

# Type checking
mypy .

# Linting
ruff check .

# Security scanning
bandit -r .

Building Documentation

pip install -e ".[docs]"
sphinx-build docs _build/html

Contributing

1. Fork the repository
2. Create a feature branch
3. Make your changes
4. Add tests for new functionality
5. Ensure all tests pass
6. Submit a pull request

License

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

Support

• 📖 Documentation
• 🐛 Issue Tracker
• 💬 Discussions

Changelog

See CHANGELOG.md for release notes.