refactor: streamline queries

TikTok/Queries/Common.py

@@ -10,13 +10,25 @@
     consistent handling of API requests and responses.
 """
 
-from typing import TYPE_CHECKING
+import httpx
+import orjson
+import structlog
+from pydantic import ValidationError, BaseModel
+from cytoolz import curry
+from typing import TYPE_CHECKING, TypeVar, Generic, Type, Any
+
+from TikTok.Exceptions.Query import QueryException
+
+logger = structlog.get_logger()
 
 if TYPE_CHECKING:
     from TikTok.Query import Query
 
+RequestModel = TypeVar("RequestModel", bound=BaseModel)
+ResponseModel = TypeVar("ResponseModel", bound=BaseModel)
+
 
-class QueryClass:
+class QueryClass(Generic[RequestModel, ResponseModel]):
     """
     A subclass to handle common API queries.
 
@@ -32,3 +44,64 @@ def __init__(self, query: "Query") -> None:
             query (Query): The parent Query instance.
         """
         self.query = query
+
+    @curry
+    async def _fetch_data(
+        self,
+        url: str,
+        request_model_class: Type[RequestModel],
+        response_model_class: Type[ResponseModel],
+        params: dict[str, Any] | None = None,
+        json_data: dict[str, Any] | None = None,
+    ) -> ResponseModel:
+        """
+        Generalized method to fetch data from the TikTok API.
+
+        This method handles the HTTP POST request, response validation, and error handling.
+
+        Parameters:
+            url (str): The API endpoint URL.
+            request_model_class (Type[RequestModel]): The Pydantic model class for the request payload.
+            response_model_class (Type[ResponseModel]): The Pydantic model class for the response payload.
+            params (dict[str, Any] | None): Query parameters for the request. Defaults to None.
+            json_data (dict[str, Any] | None): JSON payload for the request. Defaults to None.
+
+        Returns:
+            ResponseModel: An instance of the response_model_class containing the API response data.
+
+        Raises:
+            QueryException: If the API query fails or returns an error.
+            ValidationError: If the response body is invalid according to the expected model.
+            Exception: For any other unexpected errors that may occur during the API request.
+        """
+        headers = request_model_class.HeadersModel(
+            authorization=await self.query.auth.get_access_token()
+        ).model_dump(by_alias=True)
+
+        try:
+            response: httpx.Response = await self.query.client.post(
+                url=url,
+                headers=headers,
+                params=params or {},
+                json=request_model_class(**(json_data or {})).model_dump(
+                    exclude_none=True
+                ),
+            )
+            try:
+                return response_model_class(**orjson.loads(response.content))
+            except ValidationError as _:
+                error_message: dict[str, Any] = orjson.loads(response.text)
+                logger.error(
+                    f"API query failed with status {response.status_code}: {error_message['error']['message']}"
+                )
+                raise QueryException(
+                    f"TikTok API query failed because {error_message['error']['message']}"
+                )
+        except QueryException as e:
+            raise e
+        except ValidationError as e:
+            logger.error(f"Invalid response body: {e}")
+            raise e
+        except Exception as e:
+            logger.error(f"Unknown exception during API query: {e}")
+            raise e

TikTok/Queries/Playlist.py

@@ -2,18 +2,9 @@
 This module contains the PlaylistQueries class for handling playlist-related API queries to the TikTok platform.
 """
 
-import httpx
-import orjson
-import structlog
-from pydantic import ValidationError
-
 from TikTok.ValidationModels import Playlist
-from TikTok.Exceptions.Query import QueryException
-
 from TikTok.Queries.Common import QueryClass
 
-logger = structlog.get_logger()
-
 
 class PlaylistQueries(QueryClass):
     async def info(
@@ -34,41 +25,9 @@ async def info(
             ValidationError: If the response body is invalid according to the expected model.
             Exception: For any other unexpected errors that may occur during the API request.
         """
-        headers = Playlist.RequestHeadersModel(
-            authorization=await self.query.auth.get_access_token()
+        return await self._fetch_data(
+            url=self.query.endpoints.PlaylistInfoURL,
+            request_model_class=Playlist.InfoRequestModel,
+            response_model_class=Playlist.InfoResponseModel,
+            json_data={"playlist_id": playlist_id, "cursor": cursor},
         )
-        try:
-            response: httpx.Response = await self.query.client.post(
-                url=self.query.endpoints.PlaylistInfoURL,
-                headers=headers.model_dump(by_alias=True),
-                json=Playlist.InfoRequestModel(
-                    playlist_id=playlist_id,
-                    cursor=cursor,
-                ).model_dump(exclude_none=True),
-            )
-            if response.status_code != 200:
-                try:
-                    return Playlist.InfoResponseModel(**orjson.loads(response.text))
-                except ValidationError as e:
-                    logger.error(
-                        f"The attempted query failed because the response body was invalid: {e}"
-                    )
-                    raise e
-                except Exception as e:
-                    logger.error(
-                        f"The attempted query failed with the status code {response.status_code}. Details: {e}"
-                    )
-                    raise QueryException(f"TikTok API query failed. Details: {e}")
-            return Playlist.InfoResponseModel(**orjson.loads(response.content))
-        except QueryException as e:
-            raise e
-        except ValidationError as e:
-            logger.error(
-                f"The attempted query failed because the response body was invalid: {e}"
-            )
-            raise e
-        except Exception as e:
-            logger.error(
-                f"An unknown exception occurred while querying the TikTok API: {e}"
-            )
-            raise e