Data Persistence

Data persistence in this application is managed through a centralized Repository pattern, providing a clean abstraction between the business logic and the underlying storage mechanism. The primary interface for data access is the BookmarkRepository, which currently implements an in-memory storage system designed for high-performance retrieval and easy testing.

The Repository Pattern

The BookmarkRepository class, located in app/db/repository.py, serves as the single source of truth for all application entities: Bookmarks, Tags, and Collections. It encapsulates the logic for storing and retrieving these objects, ensuring that the service layer (specifically BookmarkService) does not need to manage the complexities of data storage.

The repository is initialized as a singleton component within the BookmarkService during its bootstrap phase:

# app/services/bookmark_service.py

def _init_services(self) -> None:
    """Bootstrap repository, cache, and search index."""
    self._repo = BookmarkRepository()
    self._cache: LRUCache[Bookmark] = LRUCache(max_size=256)
    self._search = SearchIndex(self._repo)

In-Memory Storage

Currently, the BookmarkRepository uses Python dictionaries to store entities in memory. This approach provides rapid access but means that data is not persistent across process restarts.

# app/db/repository.py

class BookmarkRepository:
    def __init__(self) -> None:
        self._bookmarks: Dict[str, Bookmark] = {}
        self._tags: Dict[str, Tag] = {}
        self._collections: Dict[str, Collection] = {}

Core Domain Models

Entities are represented as rich domain models that handle their own serialization and internal state transitions.

Bookmark

The Bookmark model (app/models/bookmark.py) represents a saved URL. It includes metadata such as title, description, and status (Active, Archived, or Trashed).

Tag

The Tag model (app/models/tag.py) allows for organizing bookmarks. It tracks its own usage_count, which is updated by the BookmarkService when tags are attached to or removed from bookmarks.

Collection

The Collection model (app/models/collection.py) supports two types of grouping:

1. Manual: Users explicitly add or remove bookmark IDs.
2. Smart: Bookmarks are dynamically included based on a filter_rule.

Smart collections evaluate their rules in-memory using the _apply_filter method:

# app/models/collection.py

def _apply_filter(self, bookmarks: list) -> List[str]:
    """Evaluate the filter_rule against a list of bookmarks."""
    if not self.filter_rule:
        return []
    keyword = self.filter_rule.lower()
    return [b.id for b in bookmarks if keyword in b.title.lower() or keyword in b.description.lower()]

Data Operations and Pagination

The repository provides standard CRUD operations. For bookmark retrieval, it implements a 1-based pagination system in the list_bookmarks method.

# app/db/repository.py

def list_bookmarks(
    self,
    page: int = 1,
    per_page: int = 25,
    status: Optional[str] = None,
) -> Tuple[List[Bookmark], int]:
    items = list(self._bookmarks.values())
    # ... filtering by status ...
    items.sort(key=lambda b: b.created_at, reverse=True)
    total = len(items)
    start = (page - 1) * per_page
    return items[start : start + per_page], total

Serialization

To facilitate API communication and potential future database persistence, each model implements to_dict() and from_dict() methods. This ensures a consistent format for data exchange.

# Example from app/models/tag.py
def to_dict(self) -> Dict[str, Any]:
    return {
        "id": self.id,
        "name": self.name,
        "color": self.color.value,
        "usage_count": self.usage_count,
    }

Future Scalability: Connection Pooling

While the current implementation is in-memory, the codebase includes an internal _ConnectionPool in app/db/_connection.py. This module is a thread-safe stub designed to manage a pool of reusable database connections, signaling the architecture's readiness for integration with a persistent SQL backend.

The _ConnectionPool manages a set of _Connection objects, handling acquisition and release with a threading.Lock to ensure thread safety:

# app/db/_connection.py

class _ConnectionPool:
    def acquire(self) -> _Connection:
        with self._lock:
            if self._available:
                conn = self._available.pop()
            elif len(self._in_use) < self._config.max_pool:
                conn = _Connection(self._config)
                conn.open()
            else:
                raise RuntimeError("Connection pool exhausted")
            self._in_use.append(conn)
            return conn

This internal infrastructure allows the BookmarkRepository to be swapped for a database-backed implementation without altering the public API used by the service layer.