Implement Caching for Path of Exile Items API (#35)

* chore: add function to check whether api params are defaults

* chore: handle serialization errors during caching flow

* feat: cache poe items responses

src/config/constants/app.py

@@ -22,10 +22,14 @@
 ACCESS_TOKEN_DURATION = dt.timedelta(minutes=60)
 REFRESH_TOKEN_DURATION = dt.timedelta(days=15)
 
+# * cache duration in seconds
 USER_CACHE_KEY = "users"
 SINGLE_USER_CACHE_DURATION = 60 * 60
 USERS_CACHE_DURATION = 5 * 60
 
+ITEMS_CACHE_KEY = "items"
+ITEMS_CACHE_DURATION = 6 * 60 * 60
+
 ITEMS_PER_PAGE = 100
 MAXIMUM_ITEMS_PER_PAGE = 500
 

src/routers/poe.py

@@ -1,11 +1,13 @@
-from fastapi import APIRouter, Depends, Query
+from fastapi import APIRouter, Depends, Query, Request
+from redis.asyncio.client import Redis
 
 from src import dependencies as deps
+from src.config.constants import app as consts
 from src.schemas.requests import FilterSortInput, PaginationInput
 from src.schemas.web_responses import poe as resp
 from src.schemas.responses import AppResponse, BaseResponse
 from src.services import poe as service
-from src.utils.routers import create_pagination_response
+import src.utils.routers as router_utils
 
 
 dependencies = [Depends(deps.check_access_token)]
@@ -24,14 +26,36 @@ async def get_all_categories():
 
 @router.get("/items", responses=resp.GET_ITEMS_RESPONSES)
 async def get_items(
+    request: Request,
     pagination: PaginationInput = Depends(),  # using Depends allows us to encapsulate Query params within Pydantic models
     filter_: list[str] | None = Query(None, alias="filter"),
     sort: list[str] | None = Query(None),
 ):
-    """Gets a list of all items, modified by any given parameters."""
+    """Gets a list of all items, filtered and modified by any given request parameters. Uses cached or caches result
+    dataset if using default parameters."""
+
+    async def get_items_response():
+        items, total_items = await service.get_items(pagination, filter_sort_input)
+        response = router_utils.create_pagination_response(items, total_items, pagination, "items")
+
+        return response
+
+    redis_client: Redis = request.app.state.redis
+    redis_key = f"{consts.ITEMS_CACHE_KEY}"
 
     filter_sort_input = FilterSortInput(sort=sort, filter=filter_)
-    items, total_items = await service.get_items(pagination, filter_sort_input)
+    is_default_request_input = service.check_default_request_input(pagination, filter_sort_input)
+
+    if not is_default_request_input:
+        response = await get_items_response()
+    else:
+        filter_key = filter_sort_input.filter_[0].value  # type: ignore
+        page = pagination.page
+
+        # * get available cached data or cache the data and prepare api resposne
+        redis_key = f"{redis_key}:f_{filter_key}:p_{page}"
+        response = await router_utils.get_or_cache_serialized_entity(
+            redis_key, get_items_response(), None, consts.ITEMS_CACHE_DURATION, redis_client
+        )
 
-    response = create_pagination_response(items, total_items, pagination, "items")
     return AppResponse(response)

src/services/poe.py

@@ -1,10 +1,12 @@
 from collections import defaultdict
+from typing import cast
 
 from loguru import logger
 
+import src.config.constants.app as consts
 from src.models.poe import Item, ItemCategory
 from src.schemas.poe import ItemBase, ItemGroupMapping, ItemCategoryResponse
-from src.schemas.requests import FilterSortInput, PaginationInput
+from src.schemas.requests import FilterSchema, FilterSortInput, PaginationInput, SortSchema
 from src.utils.services import QueryChainer
 
 
@@ -71,3 +73,31 @@ async def get_items(
             item.price_info.price_history = item.price_info.price_history[-7:]
 
     return items, items_count
+
+
+def check_default_request_input(pagination: PaginationInput, filter_sort_input: FilterSortInput | None) -> bool:
+    """Checks whether an Items API's request has the default/expected inputs."""
+
+    if not filter_sort_input or pagination.per_page == consts.MAXIMUM_ITEMS_PER_PAGE:
+        return False
+
+    if not filter_sort_input.filter_ or not filter_sort_input.sort:
+        return False
+
+    filter_ = cast(list[FilterSchema], filter_sort_input.filter_)
+    sort = cast(list[SortSchema], filter_sort_input.sort)
+
+    is_default_filter_input = len(filter_) == 1
+    is_default_sort_input = len(sort) == 1
+
+    for entry in filter_:
+        if entry.field != "category":
+            is_default_filter_input = False
+            break
+
+    for entry in sort:
+        if entry.field != "price_info.chaos_price" and entry.operation != "desc":
+            is_default_sort_input = False
+            break
+
+    return is_default_filter_input and is_default_sort_input

src/utils/routers.py

@@ -37,7 +37,10 @@ async def get_or_cache_serialized_entity(
         assert get_entity_function is not None
 
         data = await get_entity_function
-        serialized_entity = serialize_response(BaseResponse(data=data, key=response_key))
+        if isinstance(data, BaseResponse):
+            serialized_entity = serialize_response(data)
+        else:
+            serialized_entity = serialize_response(BaseResponse(data=data, key=response_key))
 
     await cache_data(redis_key, serialized_entity, expire_in, redis_client)
     logger.debug(f"cached '{redis_key}' data")

src/utils/services.py

@@ -18,7 +18,7 @@ def serialize_response(response: BaseResponse[T, E] | bytes) -> bytes:
     """Convenience function that serializes responses if they are `BaseResponse`s, else returns them as-is."""
 
     if isinstance(response, BaseResponse):
-        serialized_response = orjson.dumps(response.model_dump())
+        serialized_response = orjson.dumps(response.model_dump(mode="json"))
     else:
         serialized_response = response