Production Caching Strategy¶

from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional, Tuple

class CacheBackend(ABC):
    """Abstract base class for cache backends"""

    @abstractmethod
    async def get(self, key: str) -> Optional[Any]:
        """Retrieve value from cache"""
        pass

    @abstractmethod
    async def set(self, key: str, value: Any, ttl: Optional[int] = None) -> bool:
        """Store value in cache with optional TTL in seconds"""
        pass

    @abstractmethod
    async def delete(self, key: str) -> bool:
        """Delete key from cache"""
        pass

    @abstractmethod
    async def exists(self, key: str) -> bool:
        """Check if key exists in cache"""
        pass

    @abstractmethod
    async def clear(self, pattern: Optional[str] = None) -> int:
        """Clear cache entries matching pattern"""
        pass

    @abstractmethod
    async def get_stats(self) -> Dict[str, Any]:
        """Get cache statistics"""
        pass

    # Optional methods
    async def get_metadata(self, key: str) -> Optional[Dict[str, Any]]:
        """Get metadata for a key (optional)"""
        return None

    async def list_keys(
        self, pattern: Optional[str] = None, include_metadata: bool = False
    ) -> List[Tuple[str, Optional[Dict[str, Any]]]]:
        """List keys matching pattern (optional)"""
        return []

    async def cleanup_expired(self) -> int:
        """Clean up expired entries (optional)"""
        return 0

from cogniverse_core.common.cache.base import CacheManager, CacheConfig

# Configuration
cache_config = CacheConfig(
    backends=[
        {
            "backend_type": "structured_filesystem",
            "priority": 0,
            "enabled": True,
            "base_path": "/var/cache/cogniverse",
            "enable_ttl": True,
            "cleanup_on_startup": True
        }
    ],
    default_ttl=604800,  # 7 days
    enable_compression=True,
    serialization_format="pickle"
)

# Initialize
manager = CacheManager(cache_config)

# Usage
await manager.set("my_key", data, ttl=3600)
value = await manager.get("my_key")
await manager.delete("my_key")
stats = await manager.get_stats()

from cogniverse_core.common.cache.pipeline_cache import PipelineArtifactCache

# Profile is set to tenant-specific schema name
cache = PipelineArtifactCache(
    cache_manager=manager,
    ttl=604800,
    profile="video_frames_acme"  # Tenant-specific schema name
)

# Keys are automatically prefixed with the profile and a SHA-256 digest
# of the canonicalized video path (see Key Structure below):
# "video_frames_acme:video:a3f2e9d8c7b6a5f4:keyframes:..."

from cogniverse_core.common.cache.pipeline_cache import PipelineArtifactCache

cache = PipelineArtifactCache(
    cache_manager=manager,
    ttl=604800,  # 7 days
    profile=schema_name  # Tenant isolation
)

# Cache keyframes
await cache.set_keyframes(
    video_path="/path/to/video.mp4",
    keyframes_metadata=keyframes_data,
    keyframe_images=None,  # Optional
    strategy="similarity",
    threshold=0.999,
    max_frames=3000
)

# Retrieve cached keyframes (with default parameters)
cached = await cache.get_keyframes(
    video_path="/path/to/video.mp4",
    strategy="similarity",
    threshold=0.999,
    max_frames=3000
)

# Cache transcript
await cache.set_transcript(
    video_path="/path/to/video.mp4",
    transcript_data=transcript_data,
    model_size="base",
    language=None
)

# Cache VLM frame descriptions
await cache.set_descriptions(
    video_path="/path/to/video.mp4",
    descriptions_data=descriptions_data,
    model_name="Qwen/Qwen2-VL-2B-Instruct",
    batch_size=500
)
descriptions = await cache.get_descriptions(
    video_path="/path/to/video.mp4",
    model_name="Qwen/Qwen2-VL-2B-Instruct",
    batch_size=500
)

# Cache temporal segment frames (e.g. for chunked/segment-based profiles)
await cache.set_segment_frames(
    video_path="/path/to/video.mp4",
    segment_id=0,
    start_time=0.0,
    end_time=6.0,
    frames=frame_list,
    timestamps=timestamp_list,
    sampling_fps=2.0,
    max_frames=12
)
segment = await cache.get_segment_frames(
    video_path="/path/to/video.mp4",
    segment_id=0,
    start_time=0.0,
    end_time=6.0
)

# Invalidate all cache entries for a video
await cache.invalidate_video(video_path="/path/to/video.mp4")

from cogniverse_core.common.cache.backends.structured_filesystem import (
    StructuredFilesystemBackend,
    StructuredFilesystemConfig
)

config = StructuredFilesystemConfig(
    backend_type="structured_filesystem",
    base_path="/var/cache/cogniverse",  # Override default (~/.cache/cogniverse/pipeline)
    serialization_format="pickle",
    enabled=True,
    priority=0,
    enable_ttl=True,
    cleanup_on_startup=True
)

backend = StructuredFilesystemBackend(config)

# Metadata stored alongside cached files
# Supports pattern-based listing and cleanup
# TTL enforcement and cleanup on startup

from cogniverse_core.common.cache.backends.s3 import S3CacheBackend, S3CacheBackendConfig

config = S3CacheBackendConfig(
    backend_type="s3",
    # endpoint/access_key/secret_key default to MINIO_* env vars when None
    endpoint=None,          # falls back to MINIO_ENDPOINT
    access_key=None,        # falls back to MINIO_ACCESS_KEY
    secret_key=None,        # falls back to MINIO_SECRET_KEY
    bucket="cogniverse-pipeline-cache",
    key_prefix="pipeline/",
    priority=1,
    enabled=True,
)

backend = S3CacheBackend(config)
# Uses asyncio.to_thread for async I/O; envelope metadata stored in S3
# object Metadata["cache-meta"] as JSON.

from cogniverse_runtime.ingestion.pipeline import VideoIngestionPipeline
from cogniverse_foundation.config.utils import create_default_config_manager

# Initialize config manager with cache configuration
config_manager = create_default_config_manager()

# Pipeline automatically initializes cache from config_manager
pipeline = VideoIngestionPipeline(
    tenant_id="my_tenant",
    config_manager=config_manager,
    schema_name="video_frames_acme"
)

# Cache configuration is read from config_manager's app_config:
# {
#     "pipeline_cache": {
#         "enabled": True,
#         "backends": [
#             {
#                 "backend_type": "structured_filesystem",
#                 "base_path": "/var/cache/cogniverse",
#                 "priority": 0,
#                 "enabled": True,
#                 "enable_ttl": True,
#                 "cleanup_on_startup": True
#             }
#         ],
#         "default_ttl": 604800,
#         "enable_compression": True,
#         "serialization_format": "pickle"
#     }
# }

# Cache is used automatically during processing:
# - Keyframes cached after extraction
# - Transcriptions cached after audio processing
# - Frame descriptions cached after VLM inference
# - Segment frames cached for chunked/segment-based profiles
#
# Note: embeddings themselves are not cached by PipelineArtifactCache —
# embedding generation is not part of this cache's supported artifact set.

stats = await manager.get_stats()

# Returns:
{
    "manager": {
        "hits": 1234,
        "misses": 56,
        "sets": 789,
        "deletes": 12,
        "hit_rate": 0.956,
        "size_bytes": 52428800,
        "total_files": 456
    },
    "backends": {
        "StructuredFilesystemBackend": {
            "hits": 0,
            "misses": 0,
            "sets": 0,
            "deletes": 0,
            "evictions": 0,
            "size_bytes": 52428800,
            "total_files": 456,
            "metadata_files": 456
        }
    }
}

from cogniverse_core.common.cache import CacheBackendRegistry

registry = CacheBackendRegistry()
# Registered backend types:
#   "structured_filesystem" → StructuredFilesystemBackend (CONFIG_CLASS = StructuredFilesystemConfig)
#   "s3"                    → S3CacheBackend              (CONFIG_CLASS = S3CacheBackendConfig)

backend = registry.create({"backend_type": "s3", "bucket": "my-bucket"})

from cogniverse_foundation.caching import TenantLRUCache

cache: TenantLRUCache[MyPerTenantObject] = TenantLRUCache(
    capacity=16,
    on_evict=lambda tenant_id, instance: instance.close(),  # optional cleanup
)

# Atomic get-or-build: concurrent callers share one instance per tenant
instance = cache.get_or_set("tenant_acme", lambda: MyPerTenantObject(tenant_id="tenant_acme"))

# Direct access
cache.set("tenant_acme", instance)
cached = cache.get("tenant_acme")          # None if absent, moves entry to MRU
cache.pop("tenant_acme")                    # remove without triggering on_evict
list(cache.keys())                          # tenant ids, LRU order
list(cache.values())                        # snapshot of cached instances
cache.clear()                               # evicts everything, invoking on_evict per entry