database-server-rules

Database Server Specific Development Rules

Core Database Operations

Async Database Operations

import asyncio
import aiosqlite
from typing import Dict, Any, List, Optional
import logging

logger = logging.getLogger(__name__)

class DatabaseConnection:
    """Async database connection manager"""
    
    def __init__(self, db_path: str):
        self.db_path = db_path
        self.connection: Optional[aiosqlite.Connection] = None
    
    async def connect(self) -> None:
        """Establish database connection"""
        try:
            self.connection = await aiosqlite.connect(self.db_path)
            await self.connection.execute("PRAGMA foreign_keys = ON")
            await self.connection.commit()
            logger.info(f"Connected to database: {self.db_path}")
        except Exception as e:
            logger.error(f"Database connection error: {e}")
            raise
    
    async def disconnect(self) -> None:
        """Close database connection"""
        if self.connection:
            await self.connection.close()
            self.connection = None
            logger.info("Database connection closed")

SQL Query Security

import re
from typing import Dict, Any, List

# Dangerous SQL patterns to detect
DANGEROUS_PATTERNS = [
    r'\b(DROP|DELETE|TRUNCATE|ALTER|CREATE|INSERT|UPDATE)\b',
    r';\s*--',  # SQL injection comments
    r'\bUNION\b.*\bSELECT\b',
    r'\bEXEC\b|\bEXECUTE\b',
    r'\bxp_cmdshell\b',
    r'\bsp_\w+\b'
]

def validate_sql_query(query: str, allowed_operations: List[str] = None) -> bool:
    """Validate SQL query for security"""
    if not allowed_operations:
        allowed_operations = ['SELECT']
    
    # Check for dangerous patterns
    for pattern in DANGEROUS_PATTERNS:
        if re.search(pattern, query, re.IGNORECASE):
            first_word = query.strip().split()[0].upper()
            if first_word not in allowed_operations:
                raise ValueError(f"Dangerous SQL operation detected: {first_word}")
    
    return True

def sanitize_sql_input(value: Any) -> str:
    """Sanitize SQL input to prevent injection"""
    if isinstance(value, str):
        # Remove SQL special characters
        sanitized = re.sub(r'[^\w\s.-]', '', value)
        return sanitized
    return str(value)