martinholovsky/claude-skills-generator/surrealdb-expert-2

# tests/test_user_repository.py
import pytest
from surrealdb import Surreal
@pytest.fixture
async def db():
    """Set up test database connection."""
    client = Surreal("ws://localhost:8000/rpc")
    await client.connect()
    await client.use("test", "test_db")
    await client.signin({"user": "root", "pass": "root"})
    yield client
    # Cleanup
    await client.query("DELETE user;")
    await client.close()
@pytest.mark.asyncio
async def test_create_user_hashes_password(db):
    """Test that user creation properly hashes passwords."""
    # This test should FAIL initially - no implementation yet
    result = await db.query(
        """
        CREATE user CONTENT {
            email: $email,
            password: crypto::argon2::generate($password)
        } RETURN id, email, password;
        """,
        {"email": "test@example.com", "password": "secret123"}
    )
    user = result[0]["result"][0]
    assert user["email"] == "test@example.com"
    # Password should be hashed, not plain text
    assert user["password"] != "secret123"
    assert user["password"].startswith("$argon2")
@pytest.mark.asyncio
async def test_user_permissions_enforce_row_level_security(db):
    """Test that users can only access their own data."""
    # Create schema with row-level security
    await db.query("""
        DEFINE TABLE user SCHEMAFULL
            PERMISSIONS
                FOR select, update, delete WHERE id = $auth.id
                FOR create WHERE $auth.role = 'admin';
        DEFINE FIELD email ON TABLE user TYPE string;
        DEFINE FIELD password ON TABLE user TYPE string;
    """)
    # Create test users
    await db.query("""
        CREATE user:1 CONTENT { email: 'user1@test.com', password: 'hash1' };
        CREATE user:2 CONTENT { email: 'user2@test.com', password: 'hash2' };
    """)
    # Verify row-level security works
    # This requires proper auth context setup
    assert True  # Placeholder - implement auth context test
@pytest.mark.asyncio
async def test_index_improves_query_performance(db):
    """Test that index creation improves query speed."""
    # Create table and data without index
    await db.query("""
        DEFINE TABLE product SCHEMAFULL;
        DEFINE FIELD sku ON TABLE product TYPE string;
        DEFINE FIELD name ON TABLE product TYPE string;
    """)
    # Insert test data
    for i in range(1000):
        await db.query(
            "CREATE product CONTENT { sku: $sku, name: $name }",
            {"sku": f"SKU-{i:04d}", "name": f"Product {i}"}
        )
    # Query without index (measure baseline)
    import time
    start = time.time()
    await db.query("SELECT * FROM product WHERE sku = 'SKU-0500'")
    time_without_index = time.time() - start
    # Create index
    await db.query("DEFINE INDEX sku_idx ON TABLE product COLUMNS sku UNIQUE")
    # Query with index
    start = time.time()
    await db.query("SELECT * FROM product WHERE sku = 'SKU-0500'")
    time_with_index = time.time() - start
    # Index should improve performance
    assert time_with_index <= time_without_index

# src/repositories/user_repository.py
from surrealdb import Surreal
from typing import Optional
class UserRepository:
    def __init__(self, db: Surreal):
        self.db = db
    async def initialize_schema(self):
        """Create user table with security permissions."""
        await self.db.query("""
            DEFINE TABLE user SCHEMAFULL
                PERMISSIONS
                    FOR select, update, delete WHERE id = $auth.id
                    FOR create WHERE $auth.id != NONE;
            DEFINE FIELD email ON TABLE user TYPE string
                ASSERT string::is::email($value);
            DEFINE FIELD password ON TABLE user TYPE string
                VALUE crypto::argon2::generate($value);
            DEFINE FIELD created_at ON TABLE user TYPE datetime
                DEFAULT time::now();
            DEFINE INDEX email_idx ON TABLE user COLUMNS email UNIQUE;
        """)
    async def create(self, email: str, password: str) -> dict:
        """Create user with hashed password."""
        result = await self.db.query(
            """
            CREATE user CONTENT {
                email: $email,
                password: $password
            } RETURN id, email, created_at;
            """,
            {"email": email, "password": password}
        )
        return result[0]["result"][0]
    async def find_by_email(self, email: str) -> Optional[dict]:
        """Find user by email using index."""
        result = await self.db.query(
            "SELECT * FROM user WHERE email = $email",
            {"email": email}
        )
        users = result[0]["result"]
        return users[0] if users else None

# Refactored with connection pooling and better error handling
from contextlib import asynccontextmanager
from surrealdb import Surreal
import asyncio
class SurrealDBPool:
    """Connection pool for SurrealDB."""
    def __init__(self, url: str, ns: str, db: str, size: int = 10):
        self.url = url
        self.ns = ns
        self.db = db
        self.size = size
        self._pool: asyncio.Queue = asyncio.Queue(maxsize=size)
        self._initialized = False
    async def initialize(self):
        """Initialize connection pool."""
        for _ in range(self.size):
            conn = Surreal(self.url)
            await conn.connect()
            await conn.use(self.ns, self.db)
            await self._pool.put(conn)
        self._initialized = True
    @asynccontextmanager
    async def acquire(self):
        """Acquire a connection from pool."""
        if not self._initialized:
            await self.initialize()
        conn = await self._pool.get()
        try:
            yield conn
        finally:
            await self._pool.put(conn)
    async def close(self):
        """Close all connections in pool."""
        while not self._pool.empty():
            conn = await self._pool.get()
            await conn.close()

# Run all SurrealDB tests
pytest tests/test_surrealdb/ -v --asyncio-mode=auto
 
# Run with coverage
pytest tests/test_surrealdb/ --cov=src/repositories --cov-report=term-missing
 
# Run specific test file
pytest tests/test_user_repository.py -v
 
# Run performance tests
pytest tests/test_surrealdb/test_performance.py -v --benchmark-only

-- ✅ Good: Index on frequently queried fields
DEFINE INDEX email_idx ON TABLE user COLUMNS email UNIQUE;
DEFINE INDEX created_idx ON TABLE post COLUMNS created_at;
DEFINE INDEX composite_idx ON TABLE order COLUMNS user_id, status;
 
-- ✅ Good: Full-text search index
DEFINE INDEX search_idx ON TABLE article
    COLUMNS title, content
    SEARCH ANALYZER simple BM25;
 
-- Query using search index
SELECT * FROM article WHERE title @@ 'database' OR content @@ 'performance';
 
-- ❌ Bad: No indexes on queried fields
SELECT * FROM user WHERE email = $email;  -- Full table scan!
SELECT * FROM post WHERE created_at > $date;  -- Slow without index

-- ✅ Good: Single query with graph traversal (avoids N+1)
SELECT
    *,
    ->authored->post.* AS posts,
    ->follows->user.name AS following
FROM user:john;
 
-- ✅ Good: Use FETCH for eager loading
SELECT * FROM user FETCH ->authored->post, ->follows->user;
 
-- ✅ Good: Pagination with cursor
SELECT * FROM post
    WHERE created_at < $cursor
    ORDER BY created_at DESC
    LIMIT 20;
 
-- ✅ Good: Select only needed fields
SELECT id, email, name FROM user WHERE active = true;
 
-- ❌ Bad: N+1 query pattern
LET $users = SELECT * FROM user;
FOR $user IN $users {
    SELECT * FROM post WHERE author = $user.id;  -- N additional queries!
};
 
-- ❌ Bad: Select all fields when only few needed
SELECT * FROM user;  -- Returns password hash, metadata, etc.

# ✅ Good: Connection pool with proper management
import asyncio
from contextlib import asynccontextmanager
from surrealdb import Surreal
class SurrealDBPool:
    def __init__(self, url: str, ns: str, db: str, pool_size: int = 10):
        self.url = url
        self.ns = ns
        self.db = db
        self.pool_size = pool_size
        self._pool: asyncio.Queue = asyncio.Queue(maxsize=pool_size)
        self._semaphore = asyncio.Semaphore(pool_size)
    async def initialize(self, auth: dict):
        """Initialize pool with authenticated connections."""
        for _ in range(self.pool_size):
            conn = Surreal(self.url)
            await conn.connect()
            await conn.use(self.ns, self.db)
            await conn.signin(auth)
            await self._pool.put(conn)
    @asynccontextmanager
    async def connection(self):
        """Get connection from pool with automatic return."""
        async with self._semaphore:
            conn = await self._pool.get()
            try:
                yield conn
            except Exception as e:
                # Reconnect on error
                await conn.close()
                conn = Surreal(self.url)
                await conn.connect()
                raise e
            finally:
                await self._pool.put(conn)
    async def close_all(self):
        """Gracefully close all connections."""
        while not self._pool.empty():
            conn = await self._pool.get()
            await conn.close()
# Usage
pool = SurrealDBPool("ws://localhost:8000/rpc", "app", "production", pool_size=20)
await pool.initialize({"user": "admin", "pass": "secure"})