Skip to main content

The Repository Pattern Architecture

The repository layer in this application provides a clean abstraction over the persistence of bookmarks, tags, and collections. By centralizing data access within the BookmarkRepository class, the codebase decouples business logic from the underlying storage mechanism, allowing services to interact with domain models without concern for how they are stored or indexed.

The Central Data Store​

The BookmarkRepository (found in app/db/repository.py) serves as the single source of truth for the application's state. In its current implementation, it manages three primary entity types using in-memory dictionaries:

  • Bookmarks: Stored in self._bookmarks (keyed by ID).
  • Tags: Stored in self._tags (keyed by ID).
  • Collections: Stored in self._collections (keyed by ID).
# 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] = {}

Because the storage is in-memory, all data is volatile and will be lost when the application process restarts. This design choice simplifies the initial implementation while providing a clear interface that could be extended to support a persistent database like PostgreSQL or SQLite in the future.

Entity Management and CRUD​

The repository provides a standard set of CRUD (Create, Read, Update, Delete) operations for each entity type. These methods are designed to be "immediate"—since the storage is in-memory, saving an object updates the dictionary instantly.

Bookmark Operations​

The repository handles the lifecycle of Bookmark objects, including retrieval and hard deletion:

def save_bookmark(self, bookmark: Bookmark) -> None:
    """Insert or update a bookmark."""
    self._bookmarks[bookmark.id] = bookmark

def get_bookmark(self, bookmark_id: str) -> Optional[Bookmark]:
    """Retrieve a bookmark by ID, or None."""
    return self._bookmarks.get(bookmark_id)

def delete_bookmark(self, bookmark_id: str) -> bool:
    """Hard-delete a bookmark. Returns True if it existed."""
    return self._bookmarks.pop(bookmark_id, None) is not None

Querying and Pagination​

Beyond simple ID-based lookups, the repository implements logic for listing and filtering entities. The list_bookmarks method is the most complex query operation, supporting pagination and status-based filtering.

Paginated Listing​

The list_bookmarks method performs in-memory sorting and slicing. It sorts bookmarks by their created_at timestamp in descending order (newest first) before applying pagination logic.

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 (active, archived, trashed)
    if status:
        try:
            target = BookmarkStatus(status)
            items = [b for b in items if b.status == target]
        except ValueError:
            pass
            
    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

Integration with Application Services​

The repository is rarely accessed directly by route handlers. Instead, it is injected into higher-level services that coordinate business logic, caching, and search indexing.

BookmarkService Coordination​

The BookmarkService (in app/services/bookmark_service.py) uses the repository to persist changes while simultaneously updating the search index and invalidating the cache.

# app/services/bookmark_service.py

def create_bookmark(self, data: Dict[str, Any]) -> Tuple[Optional[Bookmark], Optional[str]]:
    # ... validation ...
    bookmark = Bookmark.from_dict(data)
    self._repo.save_bookmark(bookmark)  # Persist to repository
    self._search.index_bookmark(bookmark) # Update search index
    self._cache.invalidate(bookmark.id)   # Clear cache
    return bookmark, None

Search Index Resolution​

The SearchIndex (in app/services/search_service.py) maintains an inverted index of keywords to bookmark IDs. When a search is performed, it uses the repository to resolve those IDs back into full Bookmark objects.

# app/services/search_service.py

def search(self, query: str, limit: int = 20) -> List[Bookmark]:
    # ... tokenization and candidate lookup ...
    results = []
    for bid in candidate_ids:
        bookmark = self._repo.get_bookmark(bid) # Resolve ID to Entity
        if bookmark:
            results.append(bookmark)
    return self._rank_results(results, tokens)[:limit]

Architectural Considerations​

Lack of Transactions​

Because the repository is in-memory and uses standard Python dictionaries, it does not currently support transactions. If an operation requires updating multiple entities (e.g., deleting a tag and updating all associated bookmarks), the BookmarkService must handle this coordination. If a failure occurs mid-operation, the state may become inconsistent.

Testability​

The repository includes internal helpers specifically designed for testing, such as _clear_all() and _count_all(). These allow tests to reset the application state between runs without restarting the process.

def _clear_all(self) -> None:
    """Wipe all data. Test use only."""
    self._bookmarks.clear()
    self._tags.clear()
    self._collections.clear()

This architecture ensures that while the current storage is simple, the rest of the application is built against a robust interface that respects the separation of concerns between domain logic and data persistence.