feat: add query for pinned user videos

TikTok/Queries/User.py

@@ -79,7 +79,7 @@ async def info(
     async def liked_videos(
         self,
         username: str,
-        fields: list[User.UserLikedVideosQueryFields],
+        fields: list[User.UserVideosQueryFields],
         max_count: int | None = None,
         cursor: int | None = None,
     ) -> User.UserLikedVideosResponseModel:
@@ -100,7 +100,7 @@ async def liked_videos(
             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 = User.UserLikedVideosRequestHeadersModel(
+        headers = User.UserDataRequestHeadersModel(
             authorization=await self.query.auth.get_access_token()
         )
         try:
@@ -138,3 +138,56 @@ async def liked_videos(
                 f"An unknown exception occurred while querying the TikTok API: {e}"
             )
             raise e
+
+    async def pinned_videos(
+        self, username: str, fields: list[User.UserVideosQueryFields]
+    ) -> User.UserPinnedVideosResponseModel:
+        """
+        Retrieves a list of videos pinned by the specified TikTok user.
+
+        Parameters:
+            username (str): The username of the TikTok user whose pinned videos are to be retrieved.
+            fields (list[User.UserVideosQueryFields]): A list of fields to retrieve from the API.
+
+        Returns:
+            User.UserPinnedVideosResponseModel: The response data model containing the user's pinned videos.
+
+        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 = User.UserDataRequestHeadersModel(
+            authorization=await self.query.auth.get_access_token()
+        )
+        try:
+            response: httpx.Response = await self.query.client.post(
+                url=self.query.endpoints.UserPinnedVideosURL,
+                headers=headers.model_dump(by_alias=True),
+                params={"fields": fields},
+                json=User.UserPinnedVideosRequestModel(
+                    username=username,
+                ).model_dump(),
+            )
+            if response.status_code != 200:
+                error_message: dict[str, str] = orjson.loads(response.text)
+
+                logger.error(
+                    f"The attempted query failed with the status code: {response.status_code} because {error_message['error']['message']}"
+                )
+                raise QueryException(
+                    f"TikTok API query failed because {error_message['error']['message']}"
+                )
+            return User.UserPinnedVideosResponseModel(**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

TikTok/ValidationModels/RestAPI.py

@@ -51,6 +51,9 @@ class APIEndpoints(NoExtraFieldsBaseModel):
     UserLikedVideosURL: HttpUrl = (
         f"{BaseAPI.base_url}/{BaseAPI.api_version}/research/user/liked_videos/"
     )
+    UserPinnedVideosURL: HttpUrl = (
+        f"{BaseAPI.base_url}/{BaseAPI.api_version}/research/user/pinned_videos/"
+    )
     PlaylistInfoURL: HttpUrl = (
         f"{BaseAPI.base_url}/{BaseAPI.api_version}/research/playlist/info/"
     )

TikTok/ValidationModels/User.py

@@ -37,6 +37,17 @@ class UserInfoQueryFields(StrEnum):
 UserInfoRequestHeadersModel = AuthorizationHeaderModel
 
 
+class UserDataRequestHeadersModel(AuthorizationHeaderModel):
+    """
+    Model for request headers specific to user data requests.
+
+    Attributes:
+        content_type (str): The content type of the request, defaulting to "application/json".
+    """
+
+    content_type: str = Field(default="application/json", alias="Content-Type")
+
+
 class UserInfoResponseDataModel(NoExtraFieldsBaseModel):
     """
     Model for user data in the API response.
@@ -103,7 +114,7 @@ class UserInfoResponseModel(BaseModel):
     error: ResponseErrorModel
 
 
-class UserLikedVideosQueryFields(StrEnum):
+class UserVideosQueryFields(StrEnum):
     """
     Enumeration of query fields for liked videos.
 
@@ -140,17 +151,6 @@ class UserLikedVideosQueryFields(StrEnum):
     favorites_count = "favorites_count"
 
 
-class UserLikedVideosRequestHeadersModel(AuthorizationHeaderModel):
-    """
-    Model for request headers specific to liked videos requests.
-
-    Attributes:
-        content_type (str): The content type of the request, defaulting to "application/json".
-    """
-
-    content_type: str = Field(default="application/json", alias="Content-Type")
-
-
 class UserLikedVideosRequestModel(NoExtraFieldsBaseModel):
     """
     Model for the request to retrieve liked videos of a user.
@@ -177,7 +177,18 @@ class UserLikedVideosRequestModel(NoExtraFieldsBaseModel):
     )
 
 
-class UserLikedVideosDataModel(BaseModel):
+class UserPinnedVideosRequestModel(NoExtraFieldsBaseModel):
+    """
+    Model for the request to retrieve pinned videos of a user.
+
+    Attributes:
+        username (str): The unique identifier of the user whose pinned videos are to be fetched.
+    """
+
+    username: str = Field(description="Username as the unique identifier")
+
+
+class UserVideosDataModel(BaseModel):
     """
     Model representing liked video data response data in the API response.
 
@@ -260,11 +271,24 @@ class UserLikedVideosResponseDataModel(BaseModel):
         description="Retrieve liked videos starting from the specified Unix timestamp in UTC seconds"
     )
     has_more: bool = Field(description="Whether there are more liked videos or not")
-    user_liked_videos: list[UserLikedVideosDataModel] = Field(
+    user_liked_videos: list[UserVideosDataModel] = Field(
         description="The list of liked videos"
     )
 
 
+class UserPinnedVideosResponseDataModel(BaseModel):
+    """
+    Model for the response data of pinned videos.
+
+    Attributes:
+        pinned_videos_list (list[UserVideosDataModel]): A list of data models containing information about the user's pinned videos.
+    """
+
+    pinned_videos_list: list[UserVideosDataModel] = Field(
+        description="A list of video objects that match the query"
+    )
+
+
 class UserLikedVideosResponseModel(BaseModel):
     """
     Model for the complete API response for liked videos.
@@ -274,5 +298,18 @@ class UserLikedVideosResponseModel(BaseModel):
         error (ResponseErrorModel): Error information, if any.
     """
 
-    data: UserLikedVideosDataModel
+    data: UserVideosDataModel
+    error: ResponseErrorModel
+
+
+class UserPinnedVideosResponseModel(BaseModel):
+    """
+    Model for the complete API response for pinned videos.
+
+    Attributes:
+        data (UserPinnedVideosDataModel): The returned list of pinned video objects.
+        error (ResponseErrorModel): Error information, if any.
+    """
+
+    data: UserPinnedVideosResponseDataModel
     error: ResponseErrorModel