Handle optional book.path to support books without files

Books may not have a path (e.g., physical books, metadata-only entries).
Updated path-dependent operations to handle None gracefully:

- get_file: raise ValueError if book has no path
- update_book: skip path relocation if no path exists
- remove_files: skip filesystem cleanup if no path exists

Also fixed _save_book_files return type and removed unused imports.
This commit is contained in:
2026-03-07 12:38:04 -05:00
parent 3a5ea1d158
commit c67ca0e1df

View File

@@ -38,18 +38,14 @@ from chitai.database.models import (
BookSeries, BookSeries,
FileMetadata, FileMetadata,
Identifier, Identifier,
BookList,
Library, Library,
) )
from chitai.database.models.book_progress import BookProgress
from chitai.schemas.book import BooksCreateFromFiles from chitai.schemas.book import BooksCreateFromFiles
from chitai.services.filesystem_library import BookPathGenerator from chitai.services.filesystem_library import BookPathGenerator
from chitai.services.metadata_extractor import Extractor as MetadataExtractor from chitai.services.metadata_extractor import Extractor as MetadataExtractor
from chitai.services.utils import ( from chitai.services.utils import (
cleanup_empty_parent_directories, cleanup_empty_parent_directories,
delete_directory,
delete_file, delete_file,
is_empty,
move_dir_contents, move_dir_contents,
move_file, move_file,
save_image, save_image,
@@ -262,6 +258,9 @@ class BookService(SQLAlchemyAsyncRepositoryService[Book]):
ValueError: If the file is missing or not found for the given book. ValueError: If the file is missing or not found for the given book.
""" """
book = await self.get(book_id) book = await self.get(book_id)
if book.path is None:
raise ValueError("Cannot download file: book has no path")
for file in book.files: for file in book.files:
if file.id != file_id: if file.id != file_id:
continue continue
@@ -342,14 +341,15 @@ class BookService(SQLAlchemyAsyncRepositoryService[Book]):
await self._save_cover_image(data) await self._save_cover_image(data)
# TODO: extract out into its own function _update_book_path # TODO: extract out into its own function _update_book_path
# Check if file path must be updated # Check if file path must be updated (only for books with files)
path_gen = BookPathGenerator(library.root_path) if book.path is not None:
updated_path = path_gen.generate_path(book.to_dict() | data) path_gen = BookPathGenerator(library.root_path)
if str(updated_path) != book.path: updated_path = path_gen.generate_path(book.to_dict() | data)
# TODO: Move only the files associated with the book instead of the whole directory if str(updated_path) != book.path:
await move_dir_contents(book.path, updated_path) # TODO: Move only the files associated with the book instead of the whole directory
data["path"] = str(updated_path) await move_dir_contents(book.path, updated_path)
cleanup_empty_parent_directories(Path(book.path), Path(library.root_path)) data["path"] = str(updated_path)
cleanup_empty_parent_directories(Path(book.path), Path(library.root_path))
return await super().update(data, item_id=book_id, execution_options={"populate_existing": True}) return await super().update(data, item_id=book_id, execution_options={"populate_existing": True})
@@ -367,8 +367,8 @@ class BookService(SQLAlchemyAsyncRepositoryService[Book]):
book = await self.get(book_id) book = await self.get(book_id)
data = book.to_dict() data = book.to_dict()
data["files"] = files data["files"] = files
await self._save_book_files(library, data) new_files = await self._save_book_files(library, data)
book.files.extend(data["files"]) book.files.extend(new_files)
await self.update_book(book.id, {"files": [file for file in book.files]}, library) await self.update_book(book.id, {"files": [file for file in book.files]}, library)
async def remove_files( async def remove_files(
@@ -387,7 +387,7 @@ class BookService(SQLAlchemyAsyncRepositoryService[Book]):
""" """
book = await self.get_one(Book.id == book_id, Book.library_id == library.id) book = await self.get_one(Book.id == book_id, Book.library_id == library.id)
if delete_files: if delete_files and book.path is not None:
# TODO: Extract this out into its own function # TODO: Extract this out into its own function
for file in (file for file in book.files if file.id in file_ids): for file in (file for file in book.files if file.id in file_ids):
full_path = Path(book.path) / Path(file.path) full_path = Path(book.path) / Path(file.path)
@@ -446,7 +446,7 @@ class BookService(SQLAlchemyAsyncRepositoryService[Book]):
return data return data
async def _populate_with_unique_relationships(self, data: ModelDictT[Book]): async def _populate_with_unique_relationships(self, data: ModelDictT[Book]) -> ModelDictT[Book]:
""" """
Ensure relationship entities (authors, series, tags, etc.) are unique in the database. Ensure relationship entities (authors, series, tags, etc.) are unique in the database.
@@ -508,7 +508,7 @@ class BookService(SQLAlchemyAsyncRepositoryService[Book]):
return model_data return model_data
async def _save_book_files(self, library: Library, data: dict) -> dict: async def _save_book_files(self, library: Library, data: dict) -> list[FileMetadata]:
""" """
Save uploaded book files to the filesystem. Save uploaded book files to the filesystem.
@@ -558,7 +558,7 @@ class BookService(SQLAlchemyAsyncRepositoryService[Book]):
) )
data["files"] = file_metadata data["files"] = file_metadata
return data return data["files"]
async def _parse_metadata_from_files(self, data: dict, root_path: Path | None = None) -> dict: async def _parse_metadata_from_files(self, data: dict, root_path: Path | None = None) -> dict:
""" """