Files
wordarr/wordarr/api/libraries.py
T

81 lines
2.5 KiB
Python

from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, ConfigDict
from sqlalchemy import select
from sqlalchemy.orm import Session
from .. import config
from ..db import Library, get_session
router = APIRouter(prefix="/api/libraries", tags=["libraries"])
class LibraryIn(BaseModel):
name: str
media_type: str # ebook | audiobook | comic
root_path: str
folder_template: str | None = None
file_template: str | None = None
language: str = "" # "german" / "english" - filters the Audible search
class LibraryOut(LibraryIn):
id: int
model_config = ConfigDict(from_attributes=True)
def _validate(data: LibraryIn):
if data.media_type not in config.DEFAULT_NAMING:
raise HTTPException(400, f"invalid media_type: {data.media_type}")
@router.get("", response_model=list[LibraryOut])
def list_libraries(session: Session = Depends(get_session)):
return session.scalars(select(Library).order_by(Library.media_type, Library.name)).all()
@router.post("", response_model=LibraryOut)
def create_library(data: LibraryIn, session: Session = Depends(get_session)):
_validate(data)
defaults = config.DEFAULT_NAMING[data.media_type]
lib = Library(
name=data.name,
media_type=data.media_type,
root_path=data.root_path,
folder_template=data.folder_template or defaults["folder"],
file_template=data.file_template or defaults["file"],
language=data.language,
)
session.add(lib)
session.commit()
return lib
@router.put("/{library_id}", response_model=LibraryOut)
def update_library(library_id: int, data: LibraryIn, session: Session = Depends(get_session)):
_validate(data)
lib = session.get(Library, library_id)
if not lib:
raise HTTPException(404, "library not found")
defaults = config.DEFAULT_NAMING[data.media_type]
lib.name = data.name
lib.media_type = data.media_type
lib.root_path = data.root_path
lib.folder_template = data.folder_template or defaults["folder"]
lib.file_template = data.file_template or defaults["file"]
lib.language = data.language
session.commit()
return lib
@router.delete("/{library_id}")
def delete_library(library_id: int, session: Session = Depends(get_session)):
lib = session.get(Library, library_id)
if not lib:
raise HTTPException(404, "library not found")
if lib.requests:
raise HTTPException(409, "library has requests; delete them first")
session.delete(lib)
session.commit()
return {"ok": True}