Source code for sqlspec.core.cache

"""Caching system for SQL statement processing.

This module provides a caching system with LRU eviction and TTL support for
SQL statement processing and SQLGlot expression caching.

Components:
    - CacheKey: Immutable cache key
    - LRUCache: LRU + TTL cache implementation
    - NamespacedCache: Namespace-aware cache wrapper for statement processing
"""

import logging
import threading
import time
from typing import TYPE_CHECKING, Any, Final

from mypy_extensions import mypyc_attr
from typing_extensions import TypeVar

from sqlspec.core.pipeline import (
    configure_statement_pipeline_cache,
    get_statement_pipeline_metrics,
    reset_statement_pipeline_cache,
)
from sqlspec.utils.logging import get_logger, log_with_context
from sqlspec.utils.type_guards import has_field_name, has_filter_attributes

if TYPE_CHECKING:
    from collections.abc import Callable, Iterator

    import sqlglot.expressions as exp


__all__ = (
    "CacheKey",
    "CacheStats",
    "CachedStatement",
    "FiltersView",
    "LRUCache",
    "NamespacedCache",
    "create_cache_key",
    "get_cache",
    "get_cache_config",
    "get_cache_instances",
    "get_default_cache",
    "get_pipeline_metrics",
    "reset_pipeline_registry",
    "set_cache_instances",
)

logger = get_logger("sqlspec.cache")

T = TypeVar("T")
CacheValueT = TypeVar("CacheValueT")


DEFAULT_MAX_SIZE: Final = 10000
DEFAULT_TTL_SECONDS: Final = 3600
CACHE_STATS_UPDATE_INTERVAL: Final = 100


CACHE_NODE_SLOTS: Final = ("key", "value", "prev", "next", "timestamp", "access_count")
LRU_CACHE_SLOTS: Final = ("_cache", "_lock", "_max_size", "_ttl", "_head", "_tail", "_stats", "_namespace")
CACHE_STATS_SLOTS: Final = ("hits", "misses", "evictions", "total_operations", "memory_usage")



[docs]
@mypyc_attr(allow_interpreted_subclasses=False)
class CacheKey:
    """Immutable cache key.

    Args:
        key_data: Tuple of hashable values that uniquely identify the cached item
    """

    __slots__ = ("_hash", "_key_data")


[docs]
    def __init__(self, key_data: "tuple[Any, ...]") -> None:
        """Initialize cache key.

        Args:
            key_data: Tuple of hashable values for the cache key
        """
        self._key_data = key_data
        self._hash = hash(key_data)