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

minor

Browse Source
This commit is contained in:
codey
2025-08-21 15:35:10 -04:00
parent dd8d07b2f0
commit a8d089c0fe
5 changed files with 57 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
endpoints/constructors.py
View File

@@ -1,6 +1,9 @@
from typing import Optional
from typing import Literal
from pydantic import BaseModel
Station = Literal["main", "rock", "rap", "electronic", "pop"]
"""
LastFM
"""
@@ -96,25 +99,6 @@ class ValidYTSearchRequest(BaseModel):
t: str = "rick astley - never gonna give you up"
"""
XC
"""
class ValidXCRequest(BaseModel):
"""
- **key**: valid XC API key
- **bid**: bot id
- **cmd**: bot command
- **data**: command data
"""
key: str
bid: int
cmd: str
data: Optional[dict]
"""
Transcriptions
"""
@@ -211,7 +195,7 @@ class ValidRadioSongRequest(BaseModel):
song: Optional[str] = None
artistsong: Optional[str] = None
alsoSkip: Optional[bool] = False
station: str = "main"
station: Station = "main"
class ValidRadioTypeaheadRequest(BaseModel):
@@ -241,7 +225,7 @@ class ValidRadioNextRequest(BaseModel):
key: str
skipTo: Optional[str] = None
station: str = "main"
station: Station = "main"
class ValidRadioReshuffleRequest(ValidRadioNextRequest):
@@ -262,7 +246,7 @@ class ValidRadioQueueRequest(BaseModel):
draw: Optional[int] = 1
start: Optional[int] = 0
search: Optional[str] = None
station: str = "main"
station: Station = "main"
class ValidRadioQueueShiftRequest(BaseModel):
@@ -276,7 +260,7 @@ class ValidRadioQueueShiftRequest(BaseModel):
key: str
uuid: str
next: Optional[bool] = False
station: str = "main"
station: Station = "main"
class ValidRadioQueueRemovalRequest(BaseModel):
@@ -288,4 +272,4 @@ class ValidRadioQueueRemovalRequest(BaseModel):
key: str
uuid: str
station: str = "main"
station: Station = "main"

2
endpoints/meme.py
View File

@@ -30,7 +30,7 @@ class Meme(FastAPI):
f"/{endpoint}",
handler,
methods=["GET"],
include_in_schema=True,
include_in_schema=False,
dependencies=dependencies,
)

48
endpoints/radio.py
View File

@@ -10,6 +10,7 @@ from .constructors import (
ValidRadioSongRequest,
ValidRadioTypeaheadRequest,
ValidRadioQueueRequest,
Station
)
from utils import radio_util
from typing import Optional
@@ -43,23 +44,19 @@ class Radio(FastAPI):
"radio/reshuffle": self.radio_reshuffle,
"radio/queue_remove": self.radio_queue_remove,
"radio/ls._next_": self.radio_get_next,
"radio/album_art": self.album_art_handler,
}
for endpoint, handler in self.endpoints.items():
methods: list[str] = ["POST"]
if endpoint == "radio/album_art":
methods = ["GET"]
app.add_api_route(
f"/{endpoint}", handler, methods=["POST"], include_in_schema=True,
f"/{endpoint}", handler, methods=methods, include_in_schema=False,
dependencies=[Depends(
RateLimiter(times=10, seconds=2))] if not endpoint == "radio/np" else None,
)
# NOTE: Not in loop because method is GET for this endpoint
app.add_api_route(
"/radio/album_art",
self.album_art_handler,
methods=["GET"],
include_in_schema=True,
)
app.add_event_handler("startup", self.on_start)
async def on_start(self) -> None:
@@ -135,21 +132,35 @@ class Radio(FastAPI):
"""
Get current play queue (paged, 20 results per page)
"""
if not (data and data.station):
return JSONResponse(status_code=500,
content={
"err": True,
"errorText": "Invalid request.",
})
search: Optional[str] = None
draw: int = 0
if isinstance(data, ValidRadioQueueRequest):
search = data.search
draw = data.draw
start: int = int(data.start)
draw = data.draw or 0
start: int = int(data.start or 0)
end: int = start + 20
else:
start: int = 0
end: int = 20
orig_queue: list[dict] = self.radio_util.active_playlist[data.station]
if not search:
queue_full: list = orig_queue
queue_full: Optional[list] = orig_queue
else:
queue_full: list = self.radio_util.datatables_search(data.search, data.station)
queue_full = self.radio_util.datatables_search(search, data.station)
if not queue_full:
return JSONResponse(
status_code=500,
content={
"err": True,
"errorText": "No queue found.",
}
)
queue: list = queue_full[start:end]
queue_out: list[dict] = []
for x, item in enumerate(queue):
@@ -240,7 +251,7 @@ class Radio(FastAPI):
async def album_art_handler(
self, request: Request, track_id: Optional[int] = None,
station: Optional[str] = "main"
station: Station = "main"
) -> Response:
"""
Get album art, optional parameter track_id may be specified.
@@ -251,6 +262,13 @@ class Radio(FastAPI):
try:
if not track_id:
track_id = self.radio_util.now_playing[station].get("id")
if not track_id:
# Still no track ID
return JSONResponse(status_code=500,
content={
"err": True,
"errorText": "Invalid request",
})
logging.debug("Seeking album art with trackId: %s", track_id)
album_art: Optional[bytes] = self.radio_util.get_album_art(
track_id=track_id
@@ -269,7 +287,7 @@ class Radio(FastAPI):
)
async def radio_now_playing(self, request: Request,
station: Optional[str] = "main") -> JSONResponse:
station: Station = "main") -> JSONResponse:
"""
Get currently playing track info
- **station**: default "main"

6
endpoints/rip.py
View File

@@ -24,7 +24,7 @@ from pydantic import BaseModel
class ValidBulkFetchRequest(BaseModel):
track_ids: list[int]
target: str
quality: Literal["FLAC", "Lossy"]
quality: Literal["FLAC", "Lossy"] = "FLAC"
class RIP(FastAPI):
@@ -120,7 +120,7 @@ class RIP(FastAPI):
album_id: int,
request: Request,
user=Depends(get_current_user),
quality: str = "FLAC",
quality: Literal["FLAC", "Lossy"] = "FLAC"
) -> Response:
"""Get tracks by album id"""
tracks = await self.trip_util.get_tracks_by_album_id(album_id, quality)
@@ -141,8 +141,8 @@ class RIP(FastAPI):
async def track_by_id_handler(
self,
track_id: int,
quality: str,
request: Request,
quality: Literal["FLAC", "Lossy"] = "FLAC",
user=Depends(get_current_user),
) -> Response:
"""Get track by ID"""

18
endpoints/transcriptions.py
View File

@@ -1,9 +1,9 @@
import os
import aiosqlite as sqlite3
from fastapi import FastAPI, Depends
from fastapi import FastAPI, Depends, Response
from fastapi_throttle import RateLimiter
from fastapi.responses import JSONResponse
from typing import Optional, LiteralString, Union
from typing import Optional, LiteralString, Union, Iterable, cast
from .constructors import ValidShowEpisodeLineRequest, ValidShowEpisodeListRequest
@@ -88,7 +88,7 @@ class Transcriptions(FastAPI):
async with sqlite3.connect(database=db_path, timeout=1) as _db:
async with await _db.execute(db_query) as _cursor:
result: list[tuple] = await _cursor.fetchall()
result: Iterable[sqlite3.Row] = await _cursor.fetchall()
return JSONResponse(
content={
"show_title": show_title,
@@ -104,7 +104,7 @@ class Transcriptions(FastAPI):
async def get_episode_lines_handler(
self, data: ValidShowEpisodeLineRequest
) -> JSONResponse:
) -> Response:
"""
Get lines for a particular episode
- **s**: Show ID to query
@@ -138,8 +138,14 @@ class Transcriptions(FastAPI):
async with sqlite3.connect(database=db_path, timeout=1) as _db:
params: tuple = (episode_id,)
async with await _db.execute(db_query, params) as _cursor:
result: list[tuple] = await _cursor.fetchall()
first_result: tuple = result[0]
result: Iterable[sqlite3.Row] = await _cursor.fetchall()
result_list = cast(list[sqlite3.Row], result)
if not result_list:
return Response(
status_code=404,
content="Not found",
)
first_result: sqlite3.Row = result_list[0]
return JSONResponse(
content={
"episode_id": episode_id,