diff --git a/TikTok/Queries/User.py b/TikTok/Queries/User.py index bb47fad..dd0c814 100644 --- a/TikTok/Queries/User.py +++ b/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 diff --git a/TikTok/ValidationModels/RestAPI.py b/TikTok/ValidationModels/RestAPI.py index b1a9df1..bd60ac2 100644 --- a/TikTok/ValidationModels/RestAPI.py +++ b/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/" ) diff --git a/TikTok/ValidationModels/User.py b/TikTok/ValidationModels/User.py index c6d9ab0..369b403 100644 --- a/TikTok/ValidationModels/User.py +++ b/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