feat: add CSV catalogue data layer
Implement Catalogue class with BookRecord dataclass for managing the CSV-based book inventory. Provides operations to append records, read all rows, update specific books, and generate unique book IDs. 5 tests passing, all functionality working as specified.
This commit is contained in:
parent
4aee1e5181
commit
27c066de1c
2 changed files with 127 additions and 0 deletions
78
bookmark/catalogue.py
Normal file
78
bookmark/catalogue.py
Normal file
|
|
@ -0,0 +1,78 @@
|
|||
import csv
|
||||
import uuid
|
||||
from dataclasses import asdict, dataclass
|
||||
from pathlib import Path
|
||||
|
||||
CATALOGUE_COLUMNS = [
|
||||
"book_id",
|
||||
"timestamp",
|
||||
"size_bucket",
|
||||
"front_wide_path",
|
||||
"front_cropped_path",
|
||||
"back_wide_path",
|
||||
"back_cropped_path",
|
||||
"title",
|
||||
"author",
|
||||
"isbn",
|
||||
"match_confidence",
|
||||
"match_status",
|
||||
"status",
|
||||
"checked_out_at",
|
||||
]
|
||||
|
||||
|
||||
@dataclass
|
||||
class BookRecord:
|
||||
book_id: str
|
||||
timestamp: str
|
||||
size_bucket: str
|
||||
front_wide_path: str = ""
|
||||
front_cropped_path: str = ""
|
||||
back_wide_path: str = ""
|
||||
back_cropped_path: str = ""
|
||||
title: str = ""
|
||||
author: str = ""
|
||||
isbn: str = ""
|
||||
match_confidence: str = ""
|
||||
match_status: str = ""
|
||||
status: str = "captured"
|
||||
checked_out_at: str = ""
|
||||
|
||||
def to_row(self) -> dict:
|
||||
return asdict(self)
|
||||
|
||||
|
||||
class Catalogue:
|
||||
def __init__(self, csv_path: Path) -> None:
|
||||
self.csv_path = Path(csv_path)
|
||||
if not self.csv_path.exists():
|
||||
self.csv_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with open(self.csv_path, "w", newline="") as f:
|
||||
writer = csv.DictWriter(f, fieldnames=CATALOGUE_COLUMNS)
|
||||
writer.writeheader()
|
||||
|
||||
def new_book_id(self) -> str:
|
||||
return uuid.uuid4().hex[:12]
|
||||
|
||||
def append(self, record: BookRecord) -> None:
|
||||
with open(self.csv_path, "a", newline="") as f:
|
||||
writer = csv.DictWriter(f, fieldnames=CATALOGUE_COLUMNS)
|
||||
writer.writerow(record.to_row())
|
||||
|
||||
def read_all(self) -> list[dict]:
|
||||
with open(self.csv_path, newline="") as f:
|
||||
return list(csv.DictReader(f))
|
||||
|
||||
def update_row(self, book_id: str, **fields_to_update) -> None:
|
||||
rows = self.read_all()
|
||||
found = False
|
||||
for row in rows:
|
||||
if row["book_id"] == book_id:
|
||||
row.update({k: str(v) for k, v in fields_to_update.items()})
|
||||
found = True
|
||||
if not found:
|
||||
raise KeyError(f"book_id {book_id} not found in catalogue")
|
||||
with open(self.csv_path, "w", newline="") as f:
|
||||
writer = csv.DictWriter(f, fieldnames=CATALOGUE_COLUMNS)
|
||||
writer.writeheader()
|
||||
writer.writerows(rows)
|
||||
49
tests/test_catalogue.py
Normal file
49
tests/test_catalogue.py
Normal file
|
|
@ -0,0 +1,49 @@
|
|||
import pytest
|
||||
|
||||
from bookmark.catalogue import Catalogue, BookRecord, CATALOGUE_COLUMNS
|
||||
|
||||
|
||||
def test_creating_catalogue_writes_header(tmp_path):
|
||||
csv_path = tmp_path / "catalogue.csv"
|
||||
Catalogue(csv_path)
|
||||
assert csv_path.exists()
|
||||
header_line = csv_path.read_text().splitlines()[0]
|
||||
assert header_line.split(",") == CATALOGUE_COLUMNS
|
||||
|
||||
|
||||
def test_append_and_read_all_round_trip(tmp_path):
|
||||
catalogue = Catalogue(tmp_path / "catalogue.csv")
|
||||
record = BookRecord(book_id="abc123", timestamp="2026-07-13T00:00:00+00:00", size_bucket="medium")
|
||||
catalogue.append(record)
|
||||
|
||||
rows = catalogue.read_all()
|
||||
|
||||
assert len(rows) == 1
|
||||
assert rows[0]["book_id"] == "abc123"
|
||||
assert rows[0]["size_bucket"] == "medium"
|
||||
assert rows[0]["status"] == "captured"
|
||||
|
||||
|
||||
def test_new_book_id_is_unique(tmp_path):
|
||||
catalogue = Catalogue(tmp_path / "catalogue.csv")
|
||||
ids = {catalogue.new_book_id() for _ in range(100)}
|
||||
assert len(ids) == 100
|
||||
|
||||
|
||||
def test_update_row_modifies_matching_row(tmp_path):
|
||||
catalogue = Catalogue(tmp_path / "catalogue.csv")
|
||||
catalogue.append(BookRecord(book_id="abc123", timestamp="t", size_bucket="small"))
|
||||
catalogue.append(BookRecord(book_id="xyz789", timestamp="t", size_bucket="large"))
|
||||
|
||||
catalogue.update_row("abc123", title="Charlotte's Web", status="ready")
|
||||
|
||||
rows = {row["book_id"]: row for row in catalogue.read_all()}
|
||||
assert rows["abc123"]["title"] == "Charlotte's Web"
|
||||
assert rows["abc123"]["status"] == "ready"
|
||||
assert rows["xyz789"]["title"] == ""
|
||||
|
||||
|
||||
def test_update_row_raises_for_unknown_book_id(tmp_path):
|
||||
catalogue = Catalogue(tmp_path / "catalogue.csv")
|
||||
with pytest.raises(KeyError):
|
||||
catalogue.update_row("does-not-exist", title="x")
|
||||
Loading…
Reference in a new issue