codey/api
1
0
Fork 0
Code Issues Pull Requests Activity

docstrings / formatting

Browse Source
This commit is contained in:
codey
2025-09-23 13:17:34 -04:00
parent c2044711fb
commit 19afb287cd
16 changed files with 1165 additions and 428 deletions
Download Patch File Download Diff File Expand all files Collapse all files

138
endpoints/constructors.py
View File

@@ -15,7 +15,10 @@ class LastFMException(Exception):
class ValidArtistSearchRequest(BaseModel):
"""
- **a**: artist name
Request model for searching an artist by name.
Attributes:
- **a** (str): Artist name.
"""
a: str
@@ -33,8 +36,11 @@ class ValidArtistSearchRequest(BaseModel):
class ValidAlbumDetailRequest(BaseModel):
"""
- **a**: artist name
- **release**: album/release name (as sourced from here/LastFM)
Request model for album details.
Attributes:
- **a** (str): Artist name.
- **release** (str): Album/release name.
"""
a: str
@@ -54,8 +60,11 @@ class ValidAlbumDetailRequest(BaseModel):
class ValidTrackInfoRequest(BaseModel):
"""
- **a**: artist name
- **t**: track
Request model for track info.
Attributes:
- **a** (str): Artist name.
- **t** (str): Track name.
"""
a: str
@@ -80,7 +89,10 @@ Rand Msg
class RandMsgRequest(BaseModel):
"""
- **short**: Short randmsg?
Request model for random message.
Attributes:
- **short** (Optional[bool]): Short randmsg?
"""
short: Optional[bool] = False
@@ -93,7 +105,10 @@ YT
class ValidYTSearchRequest(BaseModel):
"""
- **t**: title to search
Request model for YouTube search.
Attributes:
- **t** (str): Title to search.
"""
t: str = "rick astley - never gonna give you up"
@@ -106,7 +121,10 @@ Transcriptions
class ValidShowEpisodeListRequest(BaseModel):
"""
- **s**: show id
Request model for show episode list.
Attributes:
- **s** (int): Show ID.
"""
s: int
@@ -114,8 +132,11 @@ class ValidShowEpisodeListRequest(BaseModel):
class ValidShowEpisodeLineRequest(BaseModel):
"""
- **s**: show id
- **e**: episode id
Request model for show episode lines.
Attributes:
- **s** (int): Show ID.
- **e** (int): Episode ID.
"""
s: int
@@ -129,14 +150,17 @@ Lyric Search
class ValidLyricRequest(BaseModel):
"""
- **a**: artist
- **s**: song
- **t**: track (artist and song combined) [used only if a & s are not used]
- **extra**: include extra details in response [optional, default: false]
- **lrc**: Request LRCs?
- **sub**: text to search within lyrics, if found lyrics will begin at found verse [optional]
- **src**: the script/utility which initiated the request
- **excluded_sources**: sources to exclude (new only)
Request model for lyric search.
Attributes:
- **a** (Optional[str]): Artist.
- **s** (Optional[str]): Song.
- **t** (Optional[str]): Track (artist and song combined).
- **extra** (Optional[bool]): Include extra details in response.
- **lrc** (Optional[bool]): Request LRCs?
- **sub** (Optional[str]): Text to search within lyrics.
- **src** (str): The script/utility which initiated the request.
- **excluded_sources** (Optional[list]): Sources to exclude.
"""
a: Optional[str] = None
@@ -166,7 +190,10 @@ class ValidLyricRequest(BaseModel):
class ValidTypeAheadRequest(BaseModel):
"""
- **query**: query string
Request model for typeahead query.
Attributes:
- **query** (str): Query string.
"""
query: str
@@ -183,11 +210,15 @@ class RadioException(Exception):
class ValidRadioSongRequest(BaseModel):
"""
- **key**: API Key
- **artist**: artist to search
- **song**: song to search
- **artistsong**: may be used IN PLACE OF artist/song to perform a combined/string search in the format "artist - song"
- **alsoSkip**: Whether to skip immediately to this track [not implemented]
Request model for radio song request.
Attributes:
- **key** (str): API Key.
- **artist** (Optional[str]): Artist to search.
- **song** (Optional[str]): Song to search.
- **artistsong** (Optional[str]): Combined artist-song string.
- **alsoSkip** (Optional[bool]): Whether to skip immediately to this track.
- **station** (Station): Station name.
"""
key: str
@@ -200,7 +231,10 @@ class ValidRadioSongRequest(BaseModel):
class ValidRadioTypeaheadRequest(BaseModel):
"""
- **query**: Typeahead query
Request model for radio typeahead.
Attributes:
- **query** (str): Typeahead query.
"""
query: str
@@ -208,8 +242,11 @@ class ValidRadioTypeaheadRequest(BaseModel):
class ValidRadioQueueGetRequest(BaseModel):
"""
- **key**: API key (optional, needed if specifying a non-default limit)
- **limit**: optional, default: 15k
Request model for radio queue get.
Attributes:
- **key** (Optional[str]): API key.
- **limit** (Optional[int]): Result limit.
"""
key: Optional[str] = None
@@ -218,9 +255,12 @@ class ValidRadioQueueGetRequest(BaseModel):
class ValidRadioNextRequest(BaseModel):
"""
- **key**: API Key
- **skipTo**: UUID to skip to [optional]
- **station**: Station (default: "main")
Request model for radio next track.
Attributes:
- **key** (str): API Key.
- **skipTo** (Optional[str]): UUID to skip to.
- **station** (Station): Station name.
"""
key: str
@@ -230,17 +270,23 @@ class ValidRadioNextRequest(BaseModel):
class ValidRadioReshuffleRequest(ValidRadioNextRequest):
"""
- **key**: API Key
- **station**: Station (default: "main")
Request model for radio reshuffle.
Attributes:
- **key** (str): API Key.
- **station** (Station): Station name.
"""
class ValidRadioQueueRequest(BaseModel):
"""
- **draw**: DataTables draw count, default 1
- **start**: paging start position, default 0
- **search**: Optional search query
- **station**: Station (default: "main")
Request model for radio queue.
Attributes:
- **draw** (Optional[int]): DataTables draw count.
- **start** (Optional[int]): Paging start position.
- **search** (Optional[str]): Search query.
- **station** (Station): Station name.
"""
draw: Optional[int] = 1
@@ -251,10 +297,13 @@ class ValidRadioQueueRequest(BaseModel):
class ValidRadioQueueShiftRequest(BaseModel):
"""
- **key**: API Key
- **uuid**: UUID to shift
- **next**: Play next if true, immediately if false, default False
- **station**: Station (default: "main")
Request model for radio queue shift.
Attributes:
- **key** (str): API Key.
- **uuid** (str): UUID to shift.
- **next** (Optional[bool]): Play next if true.
- **station** (Station): Station name.
"""
key: str
@@ -265,9 +314,12 @@ class ValidRadioQueueShiftRequest(BaseModel):
class ValidRadioQueueRemovalRequest(BaseModel):
"""
- **key**: API Key
- **uuid**: UUID to remove
- **station**: Station (default: "main")
Request model for radio queue removal.
Attributes:
- **key** (str): API Key.
- **uuid** (str): UUID to remove.
- **station** (Station): Station name.
"""
key: str

49
endpoints/lastfm.py
View File

@@ -17,6 +17,7 @@ class LastFM(FastAPI):
"""Last.FM Endpoints"""
def __init__(self, app: FastAPI, util, constants) -> None:
"""Initialize LastFM endpoints."""
self.app: FastAPI = app
self.util = util
self.constants = constants
@@ -44,8 +45,13 @@ class LastFM(FastAPI):
self, data: ValidArtistSearchRequest
) -> JSONResponse:
"""
Get artist info
- **a**: Artist to search
Get artist information by name.
Parameters:
- **data** (ValidArtistSearchRequest): Request containing artist name.
Returns:
- **JSONResponse**: Contains artist information or an error message.
"""
artist: Optional[str] = data.a.strip()
if not artist:
@@ -81,8 +87,13 @@ class LastFM(FastAPI):
self, data: ValidArtistSearchRequest
) -> JSONResponse:
"""
Get artist's albums/releases
- **a**: Artist to search
Get artist's albums/releases.
Parameters:
- **data** (ValidArtistSearchRequest): Request containing artist name.
Returns:
- **JSONResponse**: Contains a list of albums or an error message.
"""
artist: str = data.a.strip()
if not artist:
@@ -121,9 +132,13 @@ class LastFM(FastAPI):
self, data: ValidAlbumDetailRequest
) -> JSONResponse:
"""
Get details of a particular release by an artist
- **a**: Artist to search
- **release**: Release title to search
Get details of a particular release by an artist.
Parameters:
- **data** (ValidAlbumDetailRequest): Request containing artist and release name.
Returns:
- **JSONResponse**: Release details or error.
"""
artist: str = data.a.strip()
release: str = data.release.strip()
@@ -157,9 +172,13 @@ class LastFM(FastAPI):
self, data: ValidAlbumDetailRequest
) -> JSONResponse:
"""
Get track list for a particular release by an artist
- **a**: Artist to search
- **release**: Release title to search
Get track list for a particular release by an artist.
Parameters:
- **data** (ValidAlbumDetailRequest): Request containing artist and release name.
Returns:
- **JSONResponse**: Track list or error.
"""
artist: str = data.a.strip()
release: str = data.release.strip()
@@ -189,9 +208,13 @@ class LastFM(FastAPI):
async def track_info_handler(self, data: ValidTrackInfoRequest) -> JSONResponse:
"""
Get track info from Last.FM given an artist/track
- **a**: Artist to search
- **t**: Track title to search
Get track info from Last.FM given an artist/track.
Parameters:
- **data** (ValidTrackInfoRequest): Request containing artist and track name.
Returns:
- **JSONResponse**: Track info or error.
"""
try:
artist: str = data.a

37
endpoints/lyric_search.py
View File

@@ -20,12 +20,21 @@ class CacheUtils:
"""
def __init__(self) -> None:
"""Initialize CacheUtils."""
self.lyrics_db_path: LiteralString = os.path.join(
"/usr/local/share", "sqlite_dbs", "cached_lyrics.db"
)
async def check_typeahead(self, query: str) -> Optional[list[str]]:
"""Lyric Search Typeahead DB Handler"""
"""
Check typeahead suggestions for lyric search.
Args:
query: The search query.
Returns:
List of matching artist-song strings, or None if query is empty.
"""
if not query:
return None
async with sqlite3.connect(self.lyrics_db_path, timeout=1) as _db:
@@ -46,6 +55,7 @@ class LyricSearch(FastAPI):
"""
def __init__(self, app: FastAPI, util, constants) -> None:
"""Initialize LyricSearch endpoints."""
self.app: FastAPI = app
self.util = util
self.constants = constants
@@ -92,8 +102,13 @@ class LyricSearch(FastAPI):
async def typeahead_handler(self, data: ValidTypeAheadRequest) -> JSONResponse:
"""
Lyric search typeahead handler
- **query**: Typeahead query
Handle lyric search typeahead requests.
Parameters:
- **data** (ValidTypeAheadRequest): Request containing the query.
Returns:
- **JSONResponse**: Typeahead suggestions or error.
"""
if not isinstance(data.query, str):
return JSONResponse(
@@ -112,15 +127,13 @@ class LyricSearch(FastAPI):
async def lyric_search_handler(self, data: ValidLyricRequest) -> JSONResponse:
"""
Search for lyrics
- **a**: artist
- **s**: song
- **t**: track (artist and song combined) [used only if a & s are not used]
- **extra**: include extra details in response [optional, default: false]
- **lrc**: Request LRCs?
- **sub**: text to search within lyrics, if found lyrics will begin at found verse [optional, default: none]
- **src**: the script/utility which initiated the request
- **excluded_sources**: sources to exclude [optional, default: none]
Search for lyrics.
Parameters:
- **data** (ValidLyricRequest): Request containing artist, song, and other parameters.
Returns:
- **JSONResponse**: Lyrics data or error.
"""
if (not data.a or not data.s) and not data.t or not data.src:
raise HTTPException(detail="Invalid request", status_code=500)

35
endpoints/meme.py
View File

@@ -6,10 +6,11 @@ from utils.meme_util import MemeUtil
class Meme(FastAPI):
"""
Misc Endpoints
Meme Endpoints
"""
def __init__(self, app: FastAPI, my_util, constants) -> None:
"""Initialize Meme endpoints."""
self.app: FastAPI = app
self.util = my_util
self.meme_util = MemeUtil(constants)
@@ -35,21 +36,47 @@ class Meme(FastAPI):
)
async def get_meme_by_id(self, id: int, request: Request) -> Response:
"""Get meme (image) by id"""
"""
Get meme image by ID.
Parameters:
- **id** (int): Meme ID.
- **request** (Request): The request object.
Returns:
- **Response**: Image response or 404.
"""
meme_image = await self.meme_util.get_meme_by_id(id)
if not meme_image:
return Response(status_code=404, content="Not found")
return Response(content=meme_image, media_type="image/png")
async def random_meme(self, request: Request) -> Response:
"""Get random meme (image)"""
"""
Get a random meme image.
Parameters:
- **request** (Request): The request object.
Returns:
- **Response**: Image response or 404.
"""
meme_image = await self.meme_util.get_random_meme()
if not meme_image:
return Response(status_code=404, content="Not found")
return Response(content=meme_image, media_type="image/png")
async def list_memes(self, page: int, request: Request) -> Response:
"""Get meme (image) by id"""
"""
List memes with pagination.
Parameters:
- **page** (int): Page number.
- **request** (Request): The request object.
Returns:
- **JSONResponse**: List of memes with paging info.
"""
meme_list = await self.meme_util.list_memes(page)
page_count = await self.meme_util.get_page_count()
return JSONResponse(

65
endpoints/misc.py
View File

@@ -24,6 +24,7 @@ class Misc(FastAPI):
"""
def __init__(self, app: FastAPI, my_util, constants, radio) -> None:
"""Initialize Misc endpoints."""
self.app: FastAPI = app
self.util = my_util
self.constants = constants
@@ -84,12 +85,29 @@ class Misc(FastAPI):
return "No."
async def no(self) -> JSONResponse:
"""NaaS"""
"""
Get a random 'no' reason.
Returns:
- **JSONResponse**: Contains a random 'no' reason.
"""
return JSONResponse(content={"no": self.get_no()})
async def upload_activity_image(
self, image: UploadFile, key: Annotated[str, Form()], request: Request
) -> Response:
"""
Upload activity image.
Parameters:
- **image** (UploadFile): The uploaded image file.
- **key** (str): The API key for authentication.
- **request** (Request): The HTTP request object.
Returns:
- **Response**: Indicates success or failure of the upload.
"""
if (
not key
or not isinstance(key, str)
@@ -112,6 +130,15 @@ class Misc(FastAPI):
)
async def get_activity_image(self, request: Request) -> Response:
"""
Get the activity image.
Parameters:
- **request** (Request): The HTTP request object.
Returns:
- **Response**: The activity image or a fallback image.
"""
if isinstance(self.activity_image, bytes):
return Response(content=self.activity_image, media_type="image/png")
@@ -125,11 +152,13 @@ class Misc(FastAPI):
async def get_radio_np(self, station: str = "main") -> tuple[str, str, str]:
"""
Get radio now playing
Get radio now playing info.
Args:
None
station: Station name.
Returns:
str: Radio now playing in artist - song format
Tuple of (artistsong, album, genre).
"""
np: dict = self.radio.radio_util.now_playing[station]
@@ -145,7 +174,10 @@ class Misc(FastAPI):
async def homepage_redis_widget(self) -> JSONResponse:
"""
Homepage Redis Widget Handler
Get Redis stats for homepage widget.
Returns:
- **JSONResponse**: Contains Redis stats.
"""
# Measure response time w/ test lyric search
time_start: float = time.time() # Start time for response_time
@@ -169,7 +201,10 @@ class Misc(FastAPI):
async def homepage_rq_widget(self) -> JSONResponse:
"""
Homepage RQ Widget Handler
Get RQ job stats for homepage widget.
Returns:
- **JSONResponse**: Contains RQ job stats.
"""
queue_name = "dls"
queue = Queue(queue_name, self.redis_client)
@@ -193,7 +228,10 @@ class Misc(FastAPI):
async def homepage_sqlite_widget(self) -> JSONResponse:
"""
Homepage SQLite Widget Handler
Get SQLite stats for homepage widget.
Returns:
- **JSONResponse**: Contains SQLite stats.
"""
row_count: int = await self.lyr_cache.sqlite_rowcount()
distinct_artists: int = await self.lyr_cache.sqlite_distinct("artist")
@@ -208,7 +246,10 @@ class Misc(FastAPI):
async def homepage_lyrics_widget(self) -> JSONResponse:
"""
Homepage Lyrics Widget Handler
Get lyrics stats for homepage widget.
Returns:
- **JSONResponse**: Contains lyrics stats.
"""
found_counts: Optional[dict] = await self.redis_cache.get_found_counts()
if not isinstance(found_counts, dict):
@@ -228,7 +269,13 @@ class Misc(FastAPI):
async def homepage_radio_widget(self, station: str = "main") -> JSONResponse:
"""
Homepage Radio Widget Handler
Get radio now playing for homepage widget.
Parameters:
- **station** (str): Station name. Defaults to "main".
Returns:
- **JSONResponse**: Contains now playing info.
"""
radio_np: tuple = await self.get_radio_np(station)
if not radio_np:

126
endpoints/radio.py
View File

@@ -68,10 +68,14 @@ class Radio(FastAPI):
self, data: ValidRadioNextRequest, request: Request
) -> JSONResponse:
"""
Skip to the next track in the queue, or to uuid specified in skipTo if provided
- **key**: API key
- **skipTo**: Optional UUID to skip to
- **station**: default "main"
Skip to the next track in the queue, or to the UUID specified in `skipTo` if provided.
Parameters:
- **data** (ValidRadioNextRequest): Contains the API key, optional UUID to skip to, and station name.
- **request** (Request): The HTTP request object.
Returns:
- **JSONResponse**: Indicates success or failure of the skip operation.
"""
try:
if not self.util.check_key(path=request.url.path, req_type=4, key=data.key):
@@ -114,10 +118,16 @@ class Radio(FastAPI):
self, data: ValidRadioReshuffleRequest, request: Request
) -> JSONResponse:
"""
Reshuffle the play queue
- **key**: API key
- **station**: default "main"
Reshuffle the play queue.
Parameters:
- **data** (ValidRadioReshuffleRequest): Contains the API key and station name.
- **request** (Request): The HTTP request object.
Returns:
- **JSONResponse**: Indicates success of the reshuffle operation.
"""
if not self.util.check_key(path=request.url.path, req_type=4, key=data.key):
raise HTTPException(status_code=403, detail="Unauthorized")
@@ -130,7 +140,14 @@ class Radio(FastAPI):
data: Optional[ValidRadioQueueRequest] = None,
) -> JSONResponse:
"""
Get current play queue (paged, 20 results per page)
Get the current play queue (paged, 20 results per page).
Parameters:
- **request** (Request): The HTTP request object.
- **data** (Optional[ValidRadioQueueRequest]): Contains the station name and optional search query.
Returns:
- **JSONResponse**: Contains the paged queue data.
"""
if not (data and data.station):
return JSONResponse(status_code=500,
@@ -191,13 +208,16 @@ class Radio(FastAPI):
self, data: ValidRadioQueueShiftRequest, request: Request
) -> JSONResponse:
"""
Shift position of a UUID within the queue
[currently limited to playing next or immediately]
- **key**: API key
- **uuid**: UUID to shift
- **next**: Play track next? If False, skips to the track
- **station**: default "main"
Shift the position of a UUID within the queue.
Parameters:
- **data** (ValidRadioQueueShiftRequest): Contains the API key, UUID to shift, and station name.
- **request** (Request): The HTTP request object.
Returns:
- **JSONResponse**: Indicates success of the shift operation.
"""
if not self.util.check_key(path=request.url.path, req_type=4, key=data.key):
raise HTTPException(status_code=403, detail="Unauthorized")
@@ -225,11 +245,16 @@ class Radio(FastAPI):
self, data: ValidRadioQueueRemovalRequest, request: Request
) -> JSONResponse:
"""
Remove an item from the current play queue
- **key**: API key
- **uuid**: UUID of queue item to remove
- **station**: default "main"
Remove an item from the current play queue.
Parameters:
- **data** (ValidRadioQueueRemovalRequest): Contains the API key, UUID of the item to remove, and station name.
- **request** (Request): The HTTP request object.
Returns:
- **JSONResponse**: Indicates success of the removal operation.
"""
if not self.util.check_key(path=request.url.path, req_type=4, key=data.key):
raise HTTPException(status_code=403, detail="Unauthorized")
@@ -254,10 +279,15 @@ class Radio(FastAPI):
station: Station = "main"
) -> Response:
"""
Get album art, optional parameter track_id may be specified.
Otherwise, current track album art will be pulled.
- **track_id**: Optional, if provided, will attempt to retrieve the album art of this track_id. Current track used otherwise.
- **station**: default "main"
Get album art for the current or specified track.
Parameters:
- **request** (Request): The HTTP request object.
- **track_id** (Optional[int]): ID of the track to retrieve album art for. Defaults to the current track.
- **station** (Station): Name of the station. Defaults to "main".
Returns:
- **Response**: Contains the album art image or a default image.
"""
try:
if not track_id:
@@ -288,9 +318,16 @@ class Radio(FastAPI):
async def radio_now_playing(self, request: Request,
station: Station = "main") -> JSONResponse:
"""
Get currently playing track info
- **station**: default "main"
Get information about the currently playing track.
Parameters:
- **request** (Request): The HTTP request object.
- **station** (Station): Name of the station. Defaults to "main".
Returns:
- **JSONResponse**: Contains the track information.
"""
ret_obj: dict = {**self.radio_util.now_playing[station]}
ret_obj["station"] = station
try:
@@ -308,12 +345,17 @@ class Radio(FastAPI):
background_tasks: BackgroundTasks,
) -> JSONResponse:
"""
Get next track
(Track will be removed from the queue in the process.)
- **key**: API key
- **skipTo**: Optional UUID to skip to
- **station**: default: "main"
Get the next track in the queue. The track will be removed from the queue in the process.
Parameters:
- **data** (ValidRadioNextRequest): Contains the API key, optional UUID to skip to, and station name.
- **request** (Request): The HTTP request object.
- **background_tasks** (BackgroundTasks): Background tasks for webhook execution.
Returns:
- **JSONResponse**: Contains the next track information.
"""
logging.info("Radio get next")
if data.station not in self.radio_util.active_playlist.keys():
raise HTTPException(status_code=500, detail="No such station/not ready")
@@ -372,14 +414,16 @@ class Radio(FastAPI):
self, data: ValidRadioSongRequest, request: Request
) -> JSONResponse:
"""
Song request handler
- **key**: API key
- **artist**: Artist to search
- **song**: Song to search
- **artistsong**: Optional "Artist - Song" pair to search, in place of artist/song
- **alsoSkip**: If True, skips to the track; otherwise, track will be placed next up in queue
- **station**: default "main"
Handle song requests.
Parameters:
- **data** (ValidRadioSongRequest): Contains the API key, artist, song, and station name.
- **request** (Request): The HTTP request object.
Returns:
- **JSONResponse**: Indicates success or failure of the request.
"""
if not self.util.check_key(path=request.url.path, req_type=4, key=data.key):
raise HTTPException(status_code=403, detail="Unauthorized")
artistsong: Optional[str] = data.artistsong
@@ -413,8 +457,14 @@ class Radio(FastAPI):
self, data: ValidRadioTypeaheadRequest, request: Request
) -> JSONResponse:
"""
Radio typeahead handler
- **query**: Typeahead query
Handle typeahead queries for the radio.
Parameters:
- **data** (ValidRadioTypeaheadRequest): Contains the typeahead query.
- **request** (Request): The HTTP request object.
Returns:
- **JSONResponse**: Contains the typeahead results.
"""
if not isinstance(data.query, str):
return JSONResponse(

10
endpoints/rand_msg.py
View File

@@ -14,6 +14,7 @@ class RandMsg(FastAPI):
"""
def __init__(self, app: FastAPI, util, constants) -> None:
"""Initialize RandMsg endpoint."""
self.app: FastAPI = app
self.util = util
self.constants = constants
@@ -30,8 +31,13 @@ class RandMsg(FastAPI):
self, data: Optional[RandMsgRequest] = None
) -> JSONResponse:
"""
Get a randomly generated message
- **short**: Optional, if True, will limit length of returned random messages to <=126 characters (Discord restriction related)
Get a randomly generated message.
Parameters:
- **data** (Optional[RandMsgRequest]): Optional request data for short messages.
Returns:
- **JSONResponse**: Contains a random message.
"""
random.seed()
short: Optional[bool] = False

119
endpoints/rip.py
View File

@@ -33,6 +33,7 @@ class RIP(FastAPI):
"""
def __init__(self, app: FastAPI, my_util, constants) -> None:
"""Initialize RIP endpoints."""
self.app: FastAPI = app
self.util = my_util
self.trip_util = SRUtil()
@@ -72,7 +73,15 @@ class RIP(FastAPI):
)
def _format_job(self, job: Job):
"""Helper to normalize job data into JSON."""
"""
Helper to normalize job data into JSON.
Parameters:
- job (Job): The job object to format.
Returns:
- dict: Contains normalized job data.
"""
job_status: str | JobStatus = job.get_status()
progress = job.meta.get("progress", 0)
if progress == 100 and not job.meta.get("tarball"):
@@ -82,7 +91,13 @@ class RIP(FastAPI):
tracks_out = len(job.meta.get("tracks", []))
# `utils/rip_background.py` sets per-track status to 'Success' or 'Failed'
# so check for 'success' case-insensitively and count matches.
succeeded_tracks = len([t for t in job.meta.get("tracks", []) if str(t.get("status", "")).lower() == "success"])
succeeded_tracks = len(
[
t
for t in job.meta.get("tracks", [])
if str(t.get("status", "")).lower() == "success"
]
)
return {
"id": job.id,
@@ -103,10 +118,20 @@ class RIP(FastAPI):
async def artists_by_name_handler(
self, artist: str, request: Request, user=Depends(get_current_user)
) -> Response:
"""Get artists by name"""
"""
Get artists by name.
Parameters:
- **artist** (str): Artist name.
- **request** (Request): The request object.
- **user**: Current user (dependency).
Returns:
- **Response**: JSON response with artists or 404.
"""
# support optional grouping to return one primary per display name
# with `alternatives` for disambiguation (use ?group=true)
group = bool(request.query_params.get("group", False))
group = bool(request.query_params.get("group", False))
artists = await self.trip_util.get_artists_by_name(artist, group=group)
if not artists:
return Response(status_code=404, content="Not found")
@@ -115,7 +140,17 @@ class RIP(FastAPI):
async def albums_by_artist_id_handler(
self, artist_id: int, request: Request, user=Depends(get_current_user)
) -> Response:
"""Get albums by artist ID"""
"""
Get albums by artist ID.
Parameters:
- **artist_id** (int): Artist ID.
- **request** (Request): The request object.
- **user**: Current user (dependency).
Returns:
- **Response**: JSON response with albums or 404.
"""
albums = await self.trip_util.get_albums_by_artist_id(artist_id)
if not albums:
return Response(status_code=404, content="Not found")
@@ -128,7 +163,18 @@ class RIP(FastAPI):
user=Depends(get_current_user),
quality: Literal["FLAC", "Lossy"] = "FLAC",
) -> Response:
"""Get tracks by album id"""
"""
Get tracks by album ID.
Parameters:
- **album_id** (int): Album ID.
- **request** (Request): The request object.
- **quality** (Literal): Track quality ("FLAC" or "Lossy").
- **user**: Current user (dependency).
Returns:
- **Response**: JSON response with tracks or 404.
"""
tracks = await self.trip_util.get_tracks_by_album_id(album_id, quality)
if not tracks:
return Response(status_code=404, content="Not Found")
@@ -137,7 +183,18 @@ class RIP(FastAPI):
async def tracks_by_artist_song_handler(
self, artist: str, song: str, request: Request, user=Depends(get_current_user)
) -> Response:
"""Get tracks by artist and song name"""
"""
Get tracks by artist and song name.
Parameters:
- **artist** (str): Artist name.
- **song** (str): Song name.
- **request** (Request): The request object.
- **user**: Current user (dependency).
Returns:
- **Response**: JSON response with tracks or 404.
"""
logging.critical("Searching for tracks by artist: %s, song: %s", artist, song)
tracks = await self.trip_util.get_tracks_by_artist_song(artist, song)
if not tracks:
@@ -151,7 +208,18 @@ class RIP(FastAPI):
quality: Literal["FLAC", "Lossy"] = "FLAC",
user=Depends(get_current_user),
) -> Response:
"""Get track by ID"""
"""
Get track by ID.
Parameters:
- **track_id** (int): Track ID.
- **request** (Request): The request object.
- **quality** (Literal): Track quality ("FLAC" or "Lossy").
- **user**: Current user (dependency).
Returns:
- **Response**: JSON response with stream URL or 404.
"""
track = await self.trip_util.get_stream_url_by_track_id(track_id, quality)
if not track:
return Response(status_code=404, content="Not found")
@@ -163,7 +231,17 @@ class RIP(FastAPI):
request: Request,
user=Depends(get_current_user),
) -> Response:
"""Bulk fetch a list of track IDs"""
"""
Bulk fetch a list of track IDs.
Parameters:
- **data** (ValidBulkFetchRequest): Bulk fetch request data.
- **request** (Request): The request object.
- **user**: Current user (dependency).
Returns:
- **Response**: JSON response with job info or error.
"""
if not data or not data.track_ids or not data.target:
return JSONResponse(
content={
@@ -204,7 +282,17 @@ class RIP(FastAPI):
async def job_status_handler(
self, job_id: str, request: Request, user=Depends(get_current_user)
):
"""Get status and result of a single job"""
"""
Get status and result of a single job.
Parameters:
- **job_id** (str): Job ID.
- **request** (Request): The request object.
- **user**: Current user (dependency).
Returns:
- **JSONResponse**: Job status and result or error.
"""
job = None
try:
@@ -233,7 +321,16 @@ class RIP(FastAPI):
return self._format_job(job)
async def job_list_handler(self, request: Request, user=Depends(get_current_user)):
"""List all jobs across all registries (queued, started, finished, failed, etc)."""
"""
List all jobs across all registries.
Parameters:
- **request** (Request): The request object.
- **user**: Current user (dependency).
Returns:
- **JSONResponse**: List of jobs.
"""
jobs_info = []
seen = set()

20
endpoints/transcriptions.py
View File

@@ -13,6 +13,7 @@ class Transcriptions(FastAPI):
"""
def __init__(self, app: FastAPI, util, constants) -> None:
"""Initialize Transcriptions endpoints."""
self.app: FastAPI = app
self.util = util
self.constants = constants
@@ -36,8 +37,13 @@ class Transcriptions(FastAPI):
self, data: ValidShowEpisodeListRequest
) -> JSONResponse:
"""
Get list of episodes by show id
- **s**: Show ID to query
Get list of episodes by show ID.
Parameters:
- **data** (ValidShowEpisodeListRequest): Request containing show ID.
Returns:
- **JSONResponse**: Contains a list of episodes.
"""
show_id: int = data.s
db_path: Optional[Union[str, LiteralString]] = None
@@ -106,9 +112,13 @@ class Transcriptions(FastAPI):
self, data: ValidShowEpisodeLineRequest
) -> Response:
"""
Get lines for a particular episode
- **s**: Show ID to query
- **e**: Episode ID to query
Get lines for a particular episode.
Parameters:
- **data** (ValidShowEpisodeLineRequest): Request containing show and episode ID.
Returns:
- **Response**: Episode lines.
"""
show_id: int = int(data.s)
episode_id: int = int(data.e)

11
endpoints/yt.py
View File

@@ -12,6 +12,7 @@ class YT(FastAPI):
"""
def __init__(self, app: FastAPI, util, constants) -> None:
"""Initialize YT endpoints."""
self.app: FastAPI = app
self.util = util
self.constants = constants
@@ -31,7 +32,15 @@ class YT(FastAPI):
)
async def yt_video_search_handler(self, data: ValidYTSearchRequest) -> JSONResponse:
"""Search for YT Video by Title (closest match returned)"""
"""
Search for YouTube video by title.
Parameters:
- **data** (ValidYTSearchRequest): Request containing title.
Returns:
- **JSONResponse**: Contains video ID or an error message.
"""
title: str = data.t
yts_res: Optional[list[dict]] = await self.ytsearch.search(title)