from aiohttp import web from aiohttp_apispec import docs, querystring_schema from typing import List from server.constants import BASE_API_PREFIX, cache_keys from server.scenarios.charts import summary as scenarios from server.schemas.charts.summary import ChartStats, ChartSummary, ChartTrackSummary from server.utils.pagination import paginate from server.utils.response import dump_response_schema router = web.RouteTableDef() @router.get(BASE_API_PREFIX + "/charts/stats/") @docs(tags=["charts", "spotify", "apple"], summary="Get tracks chart entry date, max and current position.") @querystring_schema(ChartStats.Request) @dump_response_schema(ChartStats.Response(many=True), apply=False) async def get_charts_stats(request: web.Request) -> List[dict]: return await scenarios.get_stats(schema=ChartStats.Response, **request["querystring"]) @router.get(BASE_API_PREFIX + "/charts/summary/") @docs(tags=["charts", "spotify", "apple"], summary="Get track chart entry date, max and current position.") @querystring_schema(ChartSummary.Request) @dump_response_schema(ChartSummary.Response) @paginate(cache_keys.CHART_SUMMARY, extend_original_dict=True) async def get_charts_summary(request: web.Request) -> List[dict]: return await scenarios.get_summary_sorted(**request["querystring"]) @router.get(BASE_API_PREFIX + "/charts/tracks/{isrc}/summary/") @docs(tags=["charts", "spotify", "apple"], summary="Get track chart summary.") @querystring_schema(ChartTrackSummary.Request) @dump_response_schema(ChartTrackSummary.Response, apply=False) async def get_charts_tracks_summary(request: web.Request) -> dict: return await scenarios.get_track_summary(isrc=request.match_info["isrc"], **request["querystring"])