Refactor IMVDBApi and types for improved type safety and clarity

- Updated IMVDBApi to use Annotated types for query parameters and response models.
- Removed commented-out sync_video_search method to clean up the code.
- Refactored types.py to replace TypedDict with Pydantic's ModelBase for better validation and documentation.
- Enhanced field descriptions in VideoType and SearchResponseType for clearer API documentation.
This commit is contained in:
2026-03-19 12:30:41 -04:00
parent ce61346660
commit 6003280c79
2 changed files with 41 additions and 68 deletions

View File

@@ -1,18 +1,17 @@
from typing import Any from typing import Annotated, Any, Self
from beets import logging
from httpx_auth import HeaderApiKey
from lapidary.runtime import ( from lapidary.runtime import (
Body, Body,
ClientBase, ClientBase,
Query, Query,
Response, Response,
Responses, Responses,
UnexpectedResponse,
get, get,
) )
from typing import Annotated, Any, Literal, Sequence, cast, Self
from httpx_auth import HeaderApiKey
from beetsplug.beets_music_videos.types import SearchResponseType, VideoType from beetsplug.beets_music_videos.types import SearchResponseType
from beets import logging
log = logging.getLogger("beets") log = logging.getLogger("beets")
@@ -46,12 +45,13 @@ class IMVDBApi(ClientBase): # type: ignore
@get("/search/videos") # type: ignore[misc] @get("/search/videos") # type: ignore[misc]
def video_search( def video_search(
self: Self, self: Self,
*,
q: Annotated[str, Query()], q: Annotated[str, Query()],
) -> Annotated[ ) -> Annotated[
SearchResponseType, SearchResponseType,
Responses( Responses(
responses={ responses={
"2xx": Response(body=Body({"application/json": SearchResponseType})), "2xx": Response(Body({"application/json": SearchResponseType})),
} }
), ),
]: ]:
@@ -60,35 +60,3 @@ class IMVDBApi(ClientBase): # type: ignore
Implemented at runtime by Lapidary; this stub is never executed. Implemented at runtime by Lapidary; this stub is never executed.
""" """
raise NotImplementedError raise NotImplementedError
# def sync_video_search(self, query: str) -> list[VideoType]:
# """Sync wrapper used by the plugin. Runs video_search and returns .results."""
# import asyncio
# # Reuse a single event loop to avoid "Event loop is closed" on subsequent calls.
# try:
# loop = asyncio.get_event_loop()
# if loop.is_closed():
# loop = asyncio.new_event_loop()
# asyncio.set_event_loop(loop)
# except RuntimeError:
# loop = asyncio.new_event_loop()
# asyncio.set_event_loop(loop)
# try:
# resp = loop.run_until_complete(self.video_search(q=query))
# return resp["results"]
# except UnexpectedResponse as e:
# # Lapidary raises UnexpectedResponse when body validation fails
# # (e.g. API shape differs from SearchResponseType). For 200, parse
# # body and return results.
# if e.response.status_code == 200:
# try:
# body = e.response.json()
# if isinstance(body, dict):
# return cast(list[VideoType], body["results"])
# except Exception:
# pass
# log.debug("imvdb search failed for %r: %s", query[:50], e)
# return []
# except Exception as e:
# log.debug("imvdb search failed for %r: %s", query[:50], e)
# return []

View File

@@ -1,43 +1,48 @@
from __future__ import annotations from __future__ import annotations
from collections.abc import Sequence
from typing import Annotated
from lapidary.runtime import ModelBase
from pydantic import Field
from typing_extensions import TypedDict from typing_extensions import TypedDict
class ImageSizes(TypedDict, total=False): class ImageSizes(TypedDict, total=False):
"""Image URL set; keys are size codes (o=original, l=large, b=big, t=thumb).""" """Image URL set; keys are size codes (o=original, l=large, b=big, t=thumb)."""
o: str o: Annotated[str, Field(description="Original image URL")]
l: str l: Annotated[str, Field(description="Large image URL")]
b: str b: Annotated[str, Field(description="Big image URL")]
t: str t: Annotated[str, Field(description="Thumbnail image URL")]
class ArtistType(TypedDict): class ArtistType(ModelBase):
name: str name: Annotated[str, Field(description="Name")]
slug: str slug: Annotated[str, Field(description="Slug")]
url: str url: Annotated[str, Field(description="URL")]
discogs_id: int | None discogs_id: Annotated[int | None, Field(description="Discogs ID")]
class VideoType(TypedDict): class VideoType(ModelBase):
id: str id: Annotated[str, Field(description="Video ID")]
production_status: str production_status: Annotated[str, Field(description="Production status")]
song_title: str song_title: Annotated[str, Field(description="Song title")]
url: str url: Annotated[str, Field(description="Video URL")]
multiple_versions: bool multiple_versions: Annotated[bool, Field(description="Whether the video has multiple versions")]
version_name: str | None version_name: Annotated[str | None, Field(description="Version name")]
version_number: int version_number: Annotated[int, Field(description="Version number")]
is_imvdb_pick: bool is_imvdb_pick: Annotated[bool, Field(description="Whether the video is an IMVDB pick")]
aspect_ratio: str | None aspect_ratio: Annotated[str | None, Field(description="Aspect ratio")]
year: int year: Annotated[int, Field(description="Year")]
verified_credits: bool verified_credits: Annotated[bool, Field(description="Whether the video has verified credits")]
image: ImageSizes image: Annotated[ImageSizes, Field(description="Image sizes")]
artists: list[ArtistType] # API returns "artists", not "artist" artists: Annotated[Sequence[ArtistType], Field(description="List of artists")] # API returns "artists", not "artist"
class SearchResponseType(TypedDict): class SearchResponseType(ModelBase):
total: int total: Annotated[int, Field(description="Total number of results")]
current_page: int current_page: Annotated[int, Field(description="Current page number")]
total_pages: int total_pages: Annotated[int, Field(description="Total number of pages")]
per_page: int per_page: Annotated[int, Field(description="Number of results per page")]
results: list[VideoType] results: Annotated[Sequence[VideoType], Field(description="List of video results")]