app.db package
Submodules
app.db.base module
- class app.db.base.Base[source]
Bases:
SQLModel- metadata: ClassVar[MetaData] = MetaData()
- model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}
A dictionary of computed field names and their corresponding ComputedFieldInfo objects.
- model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'from_attributes': True, 'registry': PydanticUndefined}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- model_fields: ClassVar[dict[str, FieldInfo]] = {}
Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].
This replaces Model.__fields__ from Pydantic V1.
app.db.repository module
- class app.db.repository.YoutubeDataRepository(session: Session)[source]
Bases:
BaseRepository[Channel]- add_channel(channel_data: ChannelInfoSchema) Channel[source]
Adds a new channel or updates an existing one in the database based on the provided channel data.
- Parameters:
channel_data (ChannelInfoSchema) – A schema instance containing all necessary channel data.
- Returns:
The newly added or updated channel entity.
- Return type:
Channel
- Raises:
SQLAlchemyError – If there is a database operation error.
- Description:
This method checks if a channel exists in the database based on the channel_id provided within the channel_data. If the channel exists, it updates its fields with the new data. If it does not exist, a new channel instance is created and added to the database. It commits the session after adding or updating the channel. Thumbnails associated with the channel are also added by invoking the add_thumbnail method.
- add_channel_history(channel_info: Channel) None[source]
Adds historical data for a channel to the database.
- Parameters:
channel_info (Channel) – The channel object containing the data to record in history.
- Returns:
None
- Description:
This method creates a new ChannelHistory record using the data from the provided Channel object. The historical data includes the channel’s follower count, view count, and video count at the time of this call. The method logs the action and commits the new ChannelHistory record to the database.
- add_tag(tag_name: str) Tag[source]
Adds a new tag to the database or returns the existing one.
- Parameters:
tag_name (str) – The name of the tag to add or find.
- Returns:
The Tag object either retrieved or created.
- Return type:
Tag
- Raises:
Exception – Raises an exception if there’s a problem adding the tag to the database, including integrity errors.
- Description:
This method checks if the tag with the specified name exists in the database. If the tag does not exist, it creates a new Tag object, adds it to the session, and commits the session to save changes. If an error occurs during the database operation, it logs the error, rolls back the transaction, and re-raises the exception to ensure that database integrity is maintained and the error is not silently ignored.
- add_thumbnail(thumbnail_data: ThumbnailSchema, video_id: UUID | None = None, channel_id: str | None = None) Thumbnail[source]
Adds a thumbnail to the database or returns the existing one based on the URL.
- Parameters:
thumbnail_data (ThumbnailSchema) – Data schema for the thumbnail.
video_id (UUID, optional) – UUID of the video the thumbnail is associated with.
channel_id (str, optional) – ID of the channel the thumbnail is associated with.
- Returns:
The Thumbnail object either retrieved or created.
- Return type:
Thumbnail
- Raises:
ValueError – If the provided video_id or channel_id does not correspond to any existing record.
Exception – If there is a problem with adding the thumbnail to the database.
- Description:
This method first checks if there’s an existing video or channel with the provided IDs. It then checks if a thumbnail with the same URL already exists. If it does, it returns that existing thumbnail. Otherwise, it creates a new Thumbnail object using the data provided, adds it to the session, and commits the session to save the changes. If any exception occurs during these operations, the transaction is rolled back and the exception is logged and re-raised.
- add_video(video_schema: VideoSchema, channel_id: str) Video[source]
Adds a new video or updates an existing one based on the provided video schema.
- Parameters:
video_schema (VideoSchema) – The schema containing video data.
channel_id (str) – The ID of the channel to which the video belongs.
- Returns:
The newly added or updated video object.
- Return type:
Video
- Raises:
ValueError – If the channel associated with the video does not exist in the database.
SQLAlchemyError – If there is a database operation error.
- Description:
This method first checks if the channel exists in the database. If not, it raises a ValueError. If the channel exists, it checks if the video already exists. If it does not, it creates a new video object and adds it to the session. It then updates the video’s attributes with data from the video schema and commits the session. The method also manages tags and thumbnails by adding new ones or linking existing ones to the video.
- add_video_format(format_data: YTFormatSchema, youtube_video_id: str) YTFormat[source]
Adds or updates a video format in the database based on the provided format data.
- Parameters:
format_data (YTFormatSchema) – Data schema for the video format.
youtube_video_id (str) – YouTube video ID for which the format is being added or updated.
- Returns:
The newly created or updated YTFormat object.
- Return type:
YTFormat
- Raises:
ValueError – If the video associated with youtube_video_id does not exist.
- Description:
This method retrieves a video by its YouTube video ID. If the video is found, it then checks if a format with the specified format ID already exists for that video. If it exists, the existing format is updated with the new data. If it does not exist, a new YTFormat object is created and added to the session. The changes are then committed to the database. If the video is not found, an error is logged, and None is returned.
- add_video_history(video_info: Video) None[source]
Adds historical data for a video to the database.
- Parameters:
video_info (Video) – The video object containing the data to record in history.
- Returns:
None
- Description:
This method creates a new VideoHistory record using the data from the provided Video object. It captures the video’s view count, like count, and comment count at the time of this call. This historical record helps in tracking the performance of the video over time. The method logs the action and commits the new VideoHistory record to the database.
- bulk_add_tags(tags: list[str]) None[source]
Adds multiple tags to the database if they do not already exist.
- Description:
This method performs a bulk addition of tags to the database. It first retrieves all existing tags to ensure that no duplicates are created. It then filters out any tags already present in the database and adds only the new, unique tags. This approach minimizes database writes and ensures efficiency in handling large sets of tag data. If any integrity errors occur during the process (e.g., due to concurrent modifications), the transaction is rolled back, and an error is logged.
- Parameters:
tags (list[str]) – A list of tag names to be added to the database.
- Raises:
IntegrityError – If a database integrity issue occurs during the tag addition process.
- delete_video(video_id: UUID)[source]
Deletes a video from the database by its unique identifier.
- Parameters:
video_id (UUID) – The unique identifier of the video to be deleted.
- Description:
This method attempts to retrieve a video by its unique identifier from the database. If the video is found, it is deleted from the database and the change is committed. If no video is found with the provided ID, a warning is logged to indicate that the video could not be found and no action is taken.
- get_channel_id_by_url(channel_url: str) str | None[source]
Retrieves the channel ID based on a given channel URL from the database.
- Parameters:
channel_url (str) – The URL of the channel.
- Returns:
The channel ID if found, otherwise None.
- Return type:
str | None
- Description:
This method attempts to find a channel by its URL in the database. If found, it returns the channel’s ID; otherwise, it logs a warning and returns None. This function allows for easy retrieval of channel IDs without exposing the underlying database query details, providing a clean interface for users who need to get channel IDs based on URLs.
- get_channel_videos(channel_id: str) list[Video][source]
Retrieves all videos associated with a specific channel ID from the database.
- Parameters:
channel_id (str) – The unique identifier for the channel.
- Returns:
- A list of Video objects associated with the given channel ID.
The list can be empty if no videos are found for the specified channel.
- Return type:
List[Video]
- Description:
This method fetches all videos linked to a specific channel ID. It queries the Video table filtering by ‘channel_id’. If no videos are found matching the criteria, it returns an empty list and logs a warning. This method ensures that any consumer of the function can handle the output without having to deal with exceptions directly from the query.
- get_channels(limit: int = 50, page: int = 0) list[Channel][source]
Retrieves a paginated list of channels from the database.
- Parameters:
limit (int) – The maximum number of channels to return.
page (int) – The page number to retrieve, based on the limit.
- Returns:
A list of Channel objects. The list can be empty if no channels are found.
- Return type:
List[Channel]
- Description:
This method fetches a paginated list of channels sorted by their published date in ascending order. It utilizes SQL OFFSET for pagination, calculated as page * limit. This allows fetching subsets of channels for large datasets, reducing memory overhead and improving response times. If an error occurs during the query, it logs the error and returns an empty list to ensure the calling function can handle the result gracefully.
- get_new_and_existing_video_ids(video_ids: list[str], channel_id: str) tuple[list[str], list[str]][source]
Determines which video IDs from a given list are new and which already exist in the database for a specific channel.
- Parameters:
video_ids (list[str]) – A list of video IDs to check.
channel_id (str) – The ID of the channel to check against.
- Returns:
- A tuple containing two lists:
The first list contains new video IDs that do not exist in the database.
The second list contains existing video IDs that are already in the database.
- Return type:
tuple[list[str], list[str]]
- Description:
This method checks a list of video IDs against the ‘videos’ table in the database to determine which videos are new and which are already associated with a specific channel. This helps in filtering out videos that need to be added to the database and those that do not require action. The method uses a simple query to fetch existing video IDs for the specified channel and then compares them with the provided list of video IDs.
- get_video_by_id(youtube_video_id: str) Video | None[source]
Retrieves a video from the database using its YouTube video ID.
- Parameters:
youtube_video_id (str) – The YouTube ID of the video to retrieve.
- Returns:
The retrieved video object if found, otherwise None.
- Return type:
Video | None
- Description:
This method searches the database for a video that matches the given YouTube video ID. If found, it returns the Video object; otherwise, it logs a warning and returns None. This function is essential for operations that need to verify the existence of a video before performing further actions or updates based on that video.
- get_video_ids_without_formats(limit: int = 50) list[str][source]
Retrieves a list of video IDs that do not have any associated format entries, up to a specified limit.
- Parameters:
limit (int) – The maximum number of video IDs to retrieve.
- Returns:
A list of video IDs that have no associated formats.
- Return type:
list[str]
- Description:
This method queries the database to find videos that do not have associated format details in the ‘video_formats’ table. It utilizes a subquery to find distinct video IDs in the ‘video_formats’ table and then performs a LEFT OUTER JOIN to determine which videos from the ‘videos’ table do not have corresponding entries in the ‘video_formats’ table. This method is useful for identifying videos that need their format details updated or added.
- get_videos_without_upload_date(limit: int = 30) list[Video][source]
Retrieves a list of videos that do not have an upload date set, up to a specified limit.
- Parameters:
limit (int) – The maximum number of videos to retrieve.
- Returns:
A list of videos without an upload date.
- Return type:
list[Video]
- Description:
This method queries the database to find videos where the upload date is not set. It provides an additional filter to include videos where the like count is either not set or not equal to -1, typically used to denote an error or placeholder value. This method is useful for identifying videos that might have incomplete data, allowing for further updates or corrections.
- model
alias of
Channel
- reset_all_invalid_videos() None[source]
Resets the like count for all videos that have been marked as invalid in the database.
- Description:
This method finds all videos in the database with a like count of -1, indicating they have been marked as invalid or erroneous in some way. It resets their like count to 0 to clear the invalid marker and commits these changes to the database. This operation helps in maintaining data integrity and cleaning up data flags that might have been set due to processing errors or other conditions. It logs the total number of videos updated to provide feedback on the scope of the operation.
- set_video_as_invalid(video_id: str) None[source]
Marks a video as invalid by setting its like count to -1.
- Parameters:
video_id (str) – The unique identifier of the video to mark as invalid.
- Description:
This method searches for a video by its unique video ID. If the video is found, it sets the like count to -1 to mark it as invalid, and commits the change to the database. If the video is not found, it logs an error message indicating that the video could not be found.
- update_channel_details(channel_info: ChannelAPIInfoSchema) None[source]
Updates the details of an existing channel or creates a new channel if it does not exist.
- Parameters:
channel_info (ChannelAPIInfoSchema) – An object containing updated information about the channel.
- Description:
This method attempts to find a channel in the database using the ID provided in the channel_info object. If the channel exists, it updates the channel’s attributes with the new values from channel_info. If the channel does not exist, it creates a new Channel entity with the provided details and adds it to the database. This method handles updating or creating channels based on the dynamically provided channel information, ensuring data consistency and the addition of new channel data as needed.
- update_thumbnail_path(video_id: UUID, thumbnail_url: str, thumbnail_path: Path) None[source]
Updates the storage path for a specific video thumbnail.
- Parameters:
video_id (UUID) – The unique identifier of the video associated with the thumbnail.
thumbnail_url (str) – The URL of the thumbnail to update.
thumbnail_path (Path) – The new file path where the thumbnail should be stored.
- Description:
This method finds a thumbnail by its associated video ID and URL. If the thumbnail is found, its ‘thumbnail_path’ is updated to the new specified path, and the change is committed to the database. If no thumbnail matches the criteria, it logs a warning message indicating that the thumbnail could not be found.
- update_video_details(video_id: str, upload_date: datetime, like_count: int, commentCount: int, tags: list[str], defAudioLang: str) None[source]
Updates the details of an existing video with new information.
- Parameters:
video_id (str) – The ID of the video to update.
upload_date (datetime) – The new upload date for the video.
like_count (int) – The new like count for the video.
commentCount (int) – The new comment count for the video.
tags (List[str]) – A list of new tags associated with the video.
defAudioLang (str) – The default audio language for the video.
- Description:
This method finds an existing video by video_id and updates its properties including upload date, like count, comment count, and default audio language. It also manages the relationship between the video and its tags. If the video is not found, it logs an error.
- update_video_path(video_id: UUID, video_path: Path) None[source]
Updates the file path where the video is stored.
- Parameters:
video_id (UUID) – The unique identifier for the video to update.
video_path (Path) – The new file path for the video.
- Description:
This method updates the storage path of a video in the database. If the video with the specified ID exists, its ‘video_path’ attribute is updated to the new path. The method commits the change to the database. If the video does not exist, it logs a warning.