response_models.py•12.4 kB
"""Consistent Response Models for Calibre Library API
Provides standardized response formats with proper typing,
validation, and extensibility for all API endpoints.
"""
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Union
from enum import Enum
import time
class ResponseStatus(Enum):
"""Standard response statuses."""
SUCCESS = "success"
ERROR = "error"
WARNING = "warning"
PARTIAL = "partial"
@dataclass
class PaginationInfo:
"""Standardized pagination metadata."""
offset: int
limit: int
total_results: int
has_more: bool
@property
def current_page(self) -> int:
"""Calculate current page number (1-indexed)."""
return (self.offset // self.limit) + 1
@property
def total_pages(self) -> int:
"""Calculate total number of pages."""
if self.limit == 0:
return 0
return (self.total_results + self.limit - 1) // self.limit
@property
def results_range(self) -> str:
"""Human readable results range."""
start = self.offset + 1
end = min(self.offset + self.limit, self.total_results)
return f"{start}-{end} of {self.total_results}"
@dataclass
class QueryMetadata:
"""Metadata about the query execution."""
query_id: str
execution_time_ms: float
complexity: str
cache_hit: bool = False
warnings: List[str] = field(default_factory=list)
# Query optimization info
database_queries: int = 0
rows_examined: int = 0
optimizations_applied: List[str] = field(default_factory=list)
@dataclass
class SeriesInfo:
"""Standardized series information."""
name: str
index: Optional[float] = None
total_books: Optional[int] = None
@property
def display_index(self) -> str:
"""Format series index for display."""
if self.index is None:
return ""
if self.index == int(self.index):
return str(int(self.index))
return f"{self.index:.1f}"
@dataclass
class BookSummary:
"""Lightweight book representation for search results."""
book_id: int
title: str
authors: List[str]
series: Optional[SeriesInfo] = None
published_date: Optional[str] = None
formats: List[str] = field(default_factory=list)
tags: List[str] = field(default_factory=list)
# Optional enrichment data
rating: Optional[float] = None
language: Optional[str] = None
relevance_score: Optional[float] = None
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary for JSON serialization."""
result = {
"book_id": self.book_id,
"title": self.title,
"authors": self.authors,
"published_date": self.published_date,
"formats": self.formats,
"tags": self.tags[:10] # Limit tags for performance
}
if self.series:
result["series"] = {
"name": self.series.name,
"index": self.series.index,
"display_index": self.series.display_index
}
if self.rating is not None:
result["rating"] = self.rating
if self.language:
result["language"] = self.language
if self.relevance_score is not None:
result["relevance_score"] = round(self.relevance_score, 3)
return result
@dataclass
class BookDetails:
"""Complete book information for detail views."""
book_id: int
title: str
authors: List[str]
series: Optional[SeriesInfo] = None
# Publication info
published_date: Optional[str] = None
publisher: Optional[str] = None
isbn: Optional[str] = None
language: Optional[str] = None
# Library metadata
added_date: Optional[str] = None
modified_date: Optional[str] = None
rating: Optional[float] = None
tags: List[str] = field(default_factory=list)
# Content info
description: Optional[str] = None
formats: Dict[str, str] = field(default_factory=dict) # format -> file_path
file_size_mb: Optional[float] = None
# Optional content preview
content_preview: Optional[str] = None
word_count: Optional[int] = None
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary with optional fields omitted if empty."""
result = {
"book_id": self.book_id,
"title": self.title,
"authors": self.authors
}
# Add series info
if self.series:
result["series"] = {
"name": self.series.name,
"index": self.series.index,
"display_index": self.series.display_index,
"total_books": self.series.total_books
}
# Add publication info
for field_name, value in [
("published_date", self.published_date),
("publisher", self.publisher),
("isbn", self.isbn),
("language", self.language)
]:
if value:
result[field_name] = value
# Add library metadata
for field_name, value in [
("added_date", self.added_date),
("modified_date", self.modified_date),
("rating", self.rating)
]:
if value is not None:
result[field_name] = value
if self.tags:
result["tags"] = self.tags
if self.description:
result["description"] = self.description
if self.formats:
result["formats"] = self.formats
if self.file_size_mb is not None:
result["file_size_mb"] = round(self.file_size_mb, 2)
# Optional content fields
if self.content_preview:
result["content_preview"] = self.content_preview
if self.word_count is not None:
result["word_count"] = self.word_count
return result
@dataclass
class ContentMatch:
"""Represents a content search match."""
book_id: int
book_title: str
authors: List[str]
match_text: str
context_before: str
context_after: str
relevance_score: float
page_number: Optional[int] = None
chapter_title: Optional[str] = None
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary for JSON serialization."""
result = {
"book_id": self.book_id,
"book_title": self.book_title,
"authors": self.authors,
"match_text": self.match_text,
"context_before": self.context_before,
"context_after": self.context_after,
"relevance_score": round(self.relevance_score, 3)
}
if self.page_number is not None:
result["page_number"] = self.page_number
if self.chapter_title:
result["chapter_title"] = self.chapter_title
return result
@dataclass
class LibraryResponse:
"""Base response container for all API operations."""
status: ResponseStatus
data: Optional[Union[List[Dict], Dict]] = None
pagination: Optional[PaginationInfo] = None
query_metadata: Optional[QueryMetadata] = None
# Error information
error_message: Optional[str] = None
error_code: Optional[str] = None
# Additional context
warnings: List[str] = field(default_factory=list)
suggestions: List[str] = field(default_factory=list)
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary for JSON response."""
result = {
"status": self.status.value,
"timestamp": time.time()
}
if self.data is not None:
result["data"] = self.data
if self.pagination:
result["pagination"] = {
"offset": self.pagination.offset,
"limit": self.pagination.limit,
"total_results": self.pagination.total_results,
"has_more": self.pagination.has_more,
"current_page": self.pagination.current_page,
"total_pages": self.pagination.total_pages,
"results_range": self.pagination.results_range
}
if self.query_metadata:
result["query_metadata"] = {
"query_id": self.query_metadata.query_id,
"execution_time_ms": round(self.query_metadata.execution_time_ms, 2),
"complexity": self.query_metadata.complexity,
"cache_hit": self.query_metadata.cache_hit,
"database_queries": self.query_metadata.database_queries,
"rows_examined": self.query_metadata.rows_examined
}
if self.query_metadata.warnings:
result["query_metadata"]["warnings"] = self.query_metadata.warnings
if self.query_metadata.optimizations_applied:
result["query_metadata"]["optimizations_applied"] = self.query_metadata.optimizations_applied
if self.status == ResponseStatus.ERROR:
result["error"] = {
"message": self.error_message,
"code": self.error_code
}
if self.warnings:
result["warnings"] = self.warnings
if self.suggestions:
result["suggestions"] = self.suggestions
return result
@classmethod
def success(cls, data: Union[List, Dict], pagination: Optional[PaginationInfo] = None,
query_metadata: Optional[QueryMetadata] = None) -> 'LibraryResponse':
"""Create successful response."""
return cls(
status=ResponseStatus.SUCCESS,
data=data,
pagination=pagination,
query_metadata=query_metadata
)
@classmethod
def error(cls, message: str, code: Optional[str] = None) -> 'LibraryResponse':
"""Create error response."""
return cls(
status=ResponseStatus.ERROR,
error_message=message,
error_code=code
)
@classmethod
def partial(cls, data: Union[List, Dict], warnings: List[str],
pagination: Optional[PaginationInfo] = None) -> 'LibraryResponse':
"""Create partial success response."""
return cls(
status=ResponseStatus.PARTIAL,
data=data,
pagination=pagination,
warnings=warnings
)
# Response builders for common patterns
class ResponseBuilder:
"""Helper class for building standardized responses."""
@staticmethod
def book_search_response(books: List[BookSummary], pagination: PaginationInfo,
query_metadata: QueryMetadata) -> LibraryResponse:
"""Build response for book search operations."""
data = [book.to_dict() for book in books]
return LibraryResponse.success(
data={"books": data},
pagination=pagination,
query_metadata=query_metadata
)
@staticmethod
def book_details_response(books: List[BookDetails],
query_metadata: QueryMetadata) -> LibraryResponse:
"""Build response for book detail operations."""
data = [book.to_dict() for book in books]
return LibraryResponse.success(
data={"books": data},
query_metadata=query_metadata
)
@staticmethod
def content_search_response(matches: List[ContentMatch], pagination: PaginationInfo,
query_metadata: QueryMetadata) -> LibraryResponse:
"""Build response for content search operations."""
data = [match.to_dict() for match in matches]
return LibraryResponse.success(
data={"matches": data},
pagination=pagination,
query_metadata=query_metadata
)
@staticmethod
def library_stats_response(stats: Dict[str, Any]) -> LibraryResponse:
"""Build response for library statistics."""
return LibraryResponse.success(data={"statistics": stats})