The Document Libraries module provides metadata management for organizing collections of Document Sets. It follows hexagonal architecture (ports & adapters pattern) to ensure clean separation of concerns and testability.
┌─────────────────────────────────────────────────────────────┐
│ API Layer │
│ (FastAPI Routes, DTOs, Mappers) │
│ - document_libraries.py (routes) │
│ - document_library_dto.py (request/response models) │
│ - document_library_mapper.py (domain ↔ DTO conversion) │
└─────────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────┐
│ Application Layer │
│ (Business Logic Orchestration) │
│ - DocumentLibraryService │
│ • Library lifecycle management │
│ • Document set relationship management │
│ • Business rule enforcement │
└─────────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────┐
│ Domain Layer │
│ (Pure Business Logic - No Dependencies) │
│ - DocumentLibrary (domain model) │
│ - DocumentLibraryRepositoryPort (interface) │
│ - Domain exceptions │
└─────────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────┐
│ Adapters Layer │
│ (Infrastructure Implementations) │
│ - DuckDBDocumentLibraryStorage (database layer) │
│ - DuckDBDocumentLibraryMetadataRepository (port implementation) │
└─────────────────────────────────────────────────────────────┘
document_libraries/
├── README.md # This file
├── __init__.py
├── domain/ # Domain Layer (Pure Python)
│ ├── __init__.py
│ ├── models/
│ │ ├── __init__.py
│ │ └── document_library.py # DocumentLibrary domain model
│ └── ports/
│ ├── __init__.py
│ └── document_library_repository.py # Repository interface
├── adapters/ # Adapters Layer (Infrastructure)
│ ├── __init__.py
│ ├── storage/
│ │ ├── __init__.py
│ │ └── duckdb_storage.py # DuckDB storage implementation
│ └── repositories/
│ ├── __init__.py
│ └── duckdb_document_library_metadata_repository.py # Repository implementation
└── application/ # Application Layer (Services)
├── __init__.py
└── services/
├── __init__.py
└── document_library_service.py # Business logic orchestration
The core domain entity representing a collection of Document Sets.
Attributes:
library_id (str): Unique identifier (UUID)name (str): Library name (3-100 characters)description (str): Optional description (max 500 characters)tags (List[str]): Optional tags for categorization (max 20 tags)document_set_ids (List[str]): References to Document Set IDscreated_at (datetime): Creation timestamplast_modified (datetime): Last modification timestamptotal_document_sets (int): Count of document setstotal_documents (int): Aggregate document counttotal_size_bytes (int): Aggregate size in bytesKey Methods:
create(): Factory method for creating new librariesvalidate(): Validates all business rulesadd_document_set(): Adds a document set referenceremove_document_set(): Removes a document set referenceupdate_aggregate_metrics(): Updates computed metricsupdate_timestamp(): Updates last_modified timestampValidation Rules:
Defines the contract for persistence operations:
CRUD Operations:
create(library): Create a new libraryget_by_id(library_id): Retrieve by IDget_by_name(name): Retrieve by namelist_all(skip, limit): List all libraries with paginationupdate(library): Update existing librarydelete(library_id): Delete libraryRelationship Management:
add_document_set(library_id, document_set_id): Add document set to libraryremove_document_set(library_id, document_set_id): Remove document set from libraryget_document_sets_for_library(library_id): Get all document sets in libraryQuery Operations:
list_by_tags(tags, skip, limit): Filter by tagssearch_by_name(name_pattern, skip, limit): Search by nameConcrete implementation using DuckDB for metadata storage only.
Database Schema:
CREATE TABLE document_libraries (
library_id VARCHAR PRIMARY KEY,
name VARCHAR NOT NULL UNIQUE,
description VARCHAR,
created_at TIMESTAMP NOT NULL,
last_modified TIMESTAMP NOT NULL,
total_document_sets INTEGER NOT NULL DEFAULT 0,
total_documents INTEGER NOT NULL DEFAULT 0,
total_size_bytes INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE library_documentset_junction (
library_id VARCHAR NOT NULL,
document_set_id VARCHAR NOT NULL,
added_at TIMESTAMP NOT NULL,
PRIMARY KEY (library_id, document_set_id),
FOREIGN KEY (library_id) REFERENCES document_libraries(library_id)
);
Key Features:
Orchestrates business logic and coordinates between domain and repository layers.
Key Responsibilities:
Main Operations:
create_library(): Create new library with validationget_library(): Retrieve library by IDupdate_library(): Update library metadatadelete_library(): Delete library (preserves document sets)add_document_set(): Add single document set referenceremove_document_set(): Remove single document set referenceadd_document_sets_bulk(): Add multiple document sets in bulk (efficient)remove_document_sets_bulk(): Remove multiple document sets in bulk (efficient)list_libraries(): List all libraries with paginationsearch_libraries(): Search by name and tagsget_document_sets(): Get all document sets in libraryupdate_aggregate_metrics(): Update computed statisticsBase URL: /api/v1/document-libraries
Endpoints:
POST / - Create libraryGET /{library_id} - Get libraryPUT /{library_id} - Update libraryDELETE /{library_id} - Delete libraryPOST /{library_id}/document-sets/{set_id} - Add document setDELETE /{library_id}/document-sets/{set_id} - Remove document setGET / - List all librariesGET /search - Search librariesRequest Models:
CreateLibraryRequest: For creating new librariesUpdateLibraryRequest: For updating existing librariesResponse Models:
LibraryResponse: Single library responseLibraryListResponse: List of libraries with paginationMapper:
DocumentLibraryMapper: Converts between domain models and DTOstests/unit/core/assets_management/document_libraries/
├── conftest.py # Test fixtures
├── domain/
│ └── models/
│ └── test_document_library_validation.py # Domain model tests
├── application/
│ └── services/
│ └── test_document_library_service.py # Service tests
└── adapters/
└── repositories/
└── test_duckdb_document_library_repository.py # Repository tests
# Navigate to backend directory
cd src/docpipe_app/backend
# Set PYTHONPATH
export PYTHONPATH="$(cd ../../.. && pwd)/src/docpipe_app/backend:${PYTHONPATH}"
# Run all document library tests
uv run pytest ../../../tests/unit/core/assets_management/document_libraries/ -v
# Run specific test file
uv run pytest ../../../tests/unit/core/assets_management/document_libraries/domain/models/test_document_library_validation.py -v
# Run with coverage
uv run pytest ../../../tests/unit/core/assets_management/document_libraries/ --cov=core.assets_management.document_libraries --cov-report=html
from core.assets_management.document_libraries.application.services.document_library_service import DocumentLibraryService
from core.assets_management.document_libraries.adapters.repositories.duckdb_document_library_metadata_repository import DuckDBDocumentLibraryMetadataRepository
from core.assets_management.document_libraries.adapters.storage.duckdb_storage import DuckDBDocumentLibraryStorage
# Initialize storage and repository
storage = DuckDBDocumentLibraryStorage(db_path="libraries.duckdb")
repository = DuckDBDocumentLibraryMetadataRepository(storage=storage)
# Create service
service = DocumentLibraryService(repository=repository)
# Create a library
library = service.create_library(
name="Financial Documents Q1 2024",
description="Collection of financial documents for Q1 2024",
tags=["finance", "q1-2024", "reports"]
)
print(f"Created library: {library.library_id}")
# Add single document set
service.add_document_set(
library_id=library.library_id,
document_set_id="docset-001"
)
# Add multiple document sets in bulk (more efficient)
service.add_document_sets_bulk(
library_id=library.library_id,
document_set_ids=["docset-002", "docset-003", "docset-004"]
)
# Update aggregate metrics
service.update_aggregate_metrics(
library_id=library.library_id,
total_document_sets=4,
total_documents=200,
total_size_bytes=52428800
)
# Remove single document set
service.remove_document_set(
library_id=library.library_id,
document_set_id="docset-001"
)
# Remove multiple document sets in bulk (more efficient)
service.remove_document_sets_bulk(
library_id=library.library_id,
document_set_ids=["docset-002", "docset-003"]
)
# Search by name
results = service.search_libraries(
name="financial",
skip=0,
limit=10
)
# Filter by tags
results = service.search_libraries(
tags=["finance", "q1-2024"],
skip=0,
limit=10
)
Document Libraries reference Document Sets by ID but do not store document content. This design:
DocumentLibraryNotFoundError: Library does not existDocumentLibraryAlreadyExistsError: Library with same name existsInvalidDocumentLibraryError: Validation failureDocumentLibraryInvalidDataException: Invalid data statelibrary_idname(library_id, document_set_id) in junction tableAll list operations support pagination via skip and limit parameters to handle large result sets efficiently.
Consider implementing caching at the service layer for frequently accessed libraries.
When contributing to this module:
For issues or questions: