CREATE: Collection Adapter

[kjaymiller]

Issue: Collection Adapter API (Merged from #3 & #4)

Summary

Both CLI and TUI need unified collection operations with backend abstraction. Consolidate collection management, ContentManager integration, caching, and search into a single adapter that works across all backends (FileSystem, PostgreSQL, custom).

Current State

CLI Implementation

• create_collection_entry() in utils.py
• Directly calls Collection.create_entry()
• No backend abstraction or caching
• Handles editor launching, content reading, metadata merging

TUI Implementation

• ContentManager class in render_engine_integration.py
• Wraps Collection's ContentManager for backend abstraction
• get_all_posts() with caching
• create_post() with frontmatter generation
• search_posts() for filtering
• Cache invalidation on create
• Multi-collection support with set_collection()

Problem

Both tools solve the same problems independently:

1. Creating entries in collections
2. Reading/parsing entries efficiently
3. Searching across entries
4. Handling different ContentManager backends

Proposed API

Create render_engine_api.collections with:

from typing import Optional, List, Dict, Any
from pathlib import Path
from render_engine import Site, Collection, Page


class CollectionAdapter:
    """Unified collection operations with backend abstraction.

    Provides a consistent interface for CRUD operations across
    FileSystem, PostgreSQL, and custom ContentManager backends.
    Includes caching, search, and validation.
    """

    def __init__(
        self,
        site: Site,
        collection_name: Optional[str] = None,
        enable_cache: bool = True
    ):
        """Initialize adapter with a Site and optional default collection.

        Args:
            site: The render-engine Site instance
            collection_name: Optional default collection to use
            enable_cache: Whether to cache parsed pages
        """

    # Collection Management

    def get_collection(self, name: str) -> Collection:
        """Get a Collection by name.

        Args:
            name: Collection name

        Returns:
            The Collection instance

        Raises:
            ValueError: If collection not found
        """

    def list_collections(self) -> List[str]:
        """List all available collection names.

        Returns:
            List of collection names
        """

    def set_default_collection(self, name: str) -> None:
        """Set the default collection for operations.

        Args:
            name: Collection name to set as default
        """

    # Backend Introspection

    def get_backend_type(self, collection: Optional[str] = None) -> str:
        """Get ContentManager backend type.

        Args:
            collection: Collection name (uses default if not specified)

        Returns:
            Backend type: 'filesystem', 'postgresql', 'custom', etc.
        """

    def supports_native_search(self, collection: Optional[str] = None) -> bool:
        """Check if backend supports native search.

        Args:
            collection: Collection name (uses default if not specified)

        Returns:
            True if backend has native search capability
        """

    def supports_transactions(self, collection: Optional[str] = None) -> bool:
        """Check if backend supports transactions.

        Args:
            collection: Collection name (uses default if not specified)

        Returns:
            True if backend supports transactional operations
        """

    # Read Operations

    def get_pages(
        self,
        collection: Optional[str] = None,
        use_cache: bool = True
    ) -> List[Page]:
        """Get all pages from a collection.

        Args:
            collection: Collection name (uses default if not specified)
            use_cache: Whether to use cached pages if available

        Returns:
            List of Page instances
        """

    def get_page_by_slug(
        self,
        slug: str,
        collection: Optional[str] = None
    ) -> Optional[Page]:
        """Get a single page by slug.

        Args:
            slug: Page slug to find
            collection: Collection name (uses default if not specified)

        Returns:
            Page instance or None if not found
        """

    # Create Operations

    def create_entry(
        self,
        collection: str,
        filepath: Path,
        content: str,
        metadata: Dict[str, Any],
        editor: Optional[str] = None,
        open_editor: bool = False
    ) -> str:
        """Create a new entry in the collection.

        Args:
            collection: Collection name
            filepath: Path where entry should be created
            content: Entry content (without frontmatter)
            metadata: Metadata dictionary for frontmatter
            editor: Editor command to use (e.g., 'vim', 'code')
            open_editor: Whether to open editor after creation

        Returns:
            The slug of the created entry

        Raises:
            ValueError: If slug/filename invalid
            FileExistsError: If entry already exists
        """

    def create_entry_interactive(
        self,
        collection: str,
        metadata: Dict[str, Any],
        editor: Optional[str] = None
    ) -> str:
        """Create entry by opening editor for content input.

        Args:
            collection: Collection name
            metadata: Initial metadata for frontmatter
            editor: Editor command (defaults to $EDITOR)

        Returns:
            The slug of the created entry
        """

    # Search Operations

    def search_pages(
        self,
        query: str,
        fields: Optional[List[str]] = None,
        collection: Optional[str] = None,
        use_native: bool = True
    ) -> List[Page]:
        """Search pages across specified fields.

        Uses backend-native search if available and use_native=True,
        otherwise falls back to in-memory search.

        Args:
            query: Search query string
            fields: Fields to search (e.g., ['title', 'content', 'tags'])
            collection: Collection name (uses default if not specified)
            use_native: Whether to use native backend search if available

        Returns:
            List of matching Page instances
        """

    # Cache Management

    def invalidate_cache(self, collection: Optional[str] = None) -> None:
        """Invalidate page cache.

        Args:
            collection: Specific collection to invalidate, or None for all
        """

    def refresh_collection(self, collection: Optional[str] = None) -> None:
        """Refresh collection data from backend.

        Invalidates cache and reloads pages.

        Args:
            collection: Collection name (uses default if not specified)
        """

    # Validation

    def validate_slug(self, slug: str, collection: Optional[str] = None) -> bool:
        """Check if slug is valid and available.

        Args:
            slug: Slug to validate
            collection: Collection name (uses default if not specified)

        Returns:
            True if slug is valid and not in use
        """

    def generate_slug(
        self,
        title: str,
        collection: Optional[str] = None
    ) -> str:
        """Generate a unique slug from title.

        Args:
            title: Title to generate slug from
            collection: Collection name (uses default if not specified)

        Returns:
            Unique slug string
        """

Backend-Specific Handling

FileSystemManager

• Read: Parses files from content_path
• Write: Creates files with frontmatter
• Search: In-memory only (no native search)
• Cache: Cache parsed Page objects
• Path: Use collection's content_path + generated filename

PostgreSQLManager (from pg-parser)

• Read: Execute read_sql from config
• Write: Execute insert_sql with parameters
• Search: Can use SQL LIKE/FTS if configured
• Cache: Cache query results
• Transaction: Support batch operations in transactions

Custom ContentManagers

• Read: Call pages property
• Write: Call create_entry() if available
• Search: In-memory fallback
• Cache: Cache pages results
• Detection: Inspect methods/properties at runtime

Benefits

1. Single Source of Truth: One implementation for all tools
2. Backend Agnostic: Works with any ContentManager
3. Performance: Built-in caching reduces redundant parsing/queries
4. Intelligent Search: Uses native backend search when available
5. Type Safety: Proper typing throughout
6. Validation: Centralized slug and metadata validation
7. Testing: Easy to mock backends
8. Extensibility: Easy to add update/delete operations

Features Consolidated

From CLI (003):

• split_args() - Parse key=value arguments
• validate_file_name_or_slug() - Slug validation
• handle_content_file() - Read content from file/stdin
• Editor integration for interactive creation

From TUI (004):

• ContentManager backend abstraction
• Cache management with invalidation
• Multi-field search
• Collection switching
• Frontmatter generation

New Features:

• Backend capability detection
• Native search with fallback
• Slug generation with uniqueness
• Transaction support (where available)
• Batch operations (future)

Migration Path

1. Create render_engine_api.collections.CollectionAdapter
2. Implement backend detection and capability checking
3. Extract and adapt cache logic from TUI
4. Extract and adapt entry creation from CLI
5. Add unified search with native/fallback logic
6. Add validation and slug generation utilities
7. Update CLI commands to use CollectionAdapter
8. Update TUI to use CollectionAdapter instead of ContentManager
9. Add comprehensive tests with multiple backend mocks
10. Document backend-specific behaviors

Example Usage

Basic Entry Creation (CLI Style)

from render_engine_api.site_loader import SiteLoader
from render_engine_api.collections import CollectionAdapter

# Load site
loader = SiteLoader()
site = loader.load_site()

# Initialize adapter
adapter = CollectionAdapter(site, collection_name="blog")

# Create entry
slug = adapter.create_entry(
    collection="blog",
    filepath=Path("content/blog/my-post.md"),
    content="Post content here",
    metadata={"title": "My Post", "date": "2025-01-30"}
)
print(f"Created entry with slug: {slug}")

Interactive Creation with Editor

# Create with editor
slug = adapter.create_entry_interactive(
    collection="blog",
    metadata={"title": "New Post", "tags": ["python", "api"]},
    editor="vim"  # Opens vim for content input
)

Search with Backend Detection (TUI Style)

# Smart search - uses native if available
results = adapter.search_pages(
    query="python",
    fields=["title", "content", "tags"],
    collection="blog"
)

for page in results:
    print(f"{page.title} ({page.slug})")

Multi-Collection Operations

# List all collections
collections = adapter.list_collections()
print(f"Available collections: {collections}")

# Work with different collections
for coll_name in collections:
    print(f"\n{coll_name}:")
    print(f"  Backend: {adapter.get_backend_type(coll_name)}")
    print(f"  Native search: {adapter.supports_native_search(coll_name)}")

    pages = adapter.get_pages(collection=coll_name)
    print(f"  Pages: {len(pages)}")

Cache Management

# Refresh after external changes
adapter.refresh_collection("blog")

# Invalidate all caches
adapter.invalidate_cache()

Implementation Details

Caching Strategy

# Cache structure
self._cache: Dict[str, List[Page]] = {}
self._cache_time: Dict[str, float] = {}

def get_pages(self, collection: str, use_cache: bool = True):
    if use_cache and collection in self._cache:
        return self._cache[collection]

    coll = self.get_collection(collection)
    pages = list(coll.content_manager.pages)

    if use_cache:
        self._cache[collection] = pages
        self._cache_time[collection] = time.time()

    return pages

Backend Detection

def get_backend_type(self, collection: str) -> str:
    coll = self.get_collection(collection)
    cm = coll.content_manager

    # Check class name or attributes
    class_name = cm.__class__.__name__.lower()

    if 'filesystem' in class_name:
        return 'filesystem'
    elif 'postgres' in class_name or 'postgresql' in class_name:
        return 'postgresql'
    else:
        return 'custom'

Search Fallback

def search_pages(
    self,
    query: str,
    fields: List[str],
    collection: str,
    use_native: bool = True
) -> List[Page]:
    if use_native and self.supports_native_search(collection):
        # Try native backend search
        return self._native_search(query, fields, collection)
    else:
        # Fall back to in-memory search
        pages = self.get_pages(collection)
        return self._memory_search(query, fields, pages)

Dependencies

• python-frontmatter - Metadata handling
• python-slugify - Slug generation
• render-engine - Core Site/Collection/Page
• render-engine-pg - PostgreSQL backend (optional)

Testing Strategy

1. Mock Backends: Create test doubles for each backend type
2. Cache Tests: Verify invalidation and refresh
3. Search Tests: Test both native and fallback search
4. Validation Tests: Slug uniqueness and format
5. Integration Tests: Real backends in test environment
6. Error Cases: Missing collections, invalid slugs, backend failures

Related Issues

• Create SiteLoader #1: Site Loader API (provides Site to adapter)
• Implement Caching Strategy #5: Search Operations API (search algorithm details)
• Create Validation Endpoint #6: Caching Strategy API (cache implementation details)
• #007: Entry Validation API (validation rules)

Future Enhancements

• Update entry operations
• Delete entry operations
• Batch operations with transactions
• Async support for database backends
• Webhook/event system for cache invalidation
• Query builder for advanced search