Merge pull request #248 from luigi311/reliable

Improve reliability

pyproject.toml

@@ -18,5 +18,7 @@ lint = [
     "ruff>=0.9.6",
 ]
 dev = [
+    "mypy>=1.15.0",
     "pytest>=8.3.4",
+    "types-requests>=2.32.0.20250306",
 ]

src/black_white.py

@@ -12,7 +12,7 @@ def setup_black_white_lists(
     whitelist_users: list[str] | None,
     library_mapping: dict[str, str] | None = None,
     user_mapping: dict[str, str] | None = None,
-):
+) -> tuple[list[str], list[str], list[str], list[str], list[str], list[str]]:
     blacklist_library, blacklist_library_type, blacklist_users = setup_x_lists(
         blacklist_library,
         blacklist_library_type,

src/connection.py

@@ -25,17 +25,17 @@ def jellyfin_emby_server_connection(
             f"{server_type.upper()}_BASEURL and {server_type.upper()}_TOKEN must have the same number of entries"
         )
 
-    for i, baseurl in enumerate(server_baseurls):
-        baseurl = baseurl.strip()
-        if baseurl[-1] == "/":
-            baseurl = baseurl[:-1]
+    for i, base_url in enumerate(server_baseurls):
+        base_url = base_url.strip()
+        if base_url[-1] == "/":
+            base_url = base_url[:-1]
 
         if server_type == "jellyfin":
-            server = Jellyfin(baseurl=baseurl, token=server_tokens[i].strip())
+            server = Jellyfin(base_url=base_url, token=server_tokens[i].strip())
             servers.append(server)
 
         elif server_type == "emby":
-            server = Emby(baseurl=baseurl, token=server_tokens[i].strip())
+            server = Emby(base_url=base_url, token=server_tokens[i].strip())
             servers.append(server)
         else:
             raise Exception("Unknown server type")
@@ -66,11 +66,11 @@ def generate_server_connections() -> list[Plex | Jellyfin | Emby]:
 
         for i, url in enumerate(plex_baseurl):
             server = Plex(
-                baseurl=url.strip(),
+                base_url=url.strip(),
                 token=plex_token[i].strip(),
-                username=None,
+                user_name=None,
                 password=None,
-                servername=None,
+                server_name=None,
                 ssl_bypass=ssl_bypass,
             )
 
@@ -92,11 +92,11 @@ def generate_server_connections() -> list[Plex | Jellyfin | Emby]:
 
         for i, username in enumerate(plex_username):
             server = Plex(
-                baseurl=None,
+                base_url=None,
                 token=None,
-                username=username.strip(),
+                user_name=username.strip(),
                 password=plex_password[i].strip(),
-                servername=plex_servername[i].strip(),
+                server_name=plex_servername[i].strip(),
                 ssl_bypass=ssl_bypass,
             )
 

src/emby.py

@@ -1,9 +1,10 @@
 from src.jellyfin_emby import JellyfinEmby
 from packaging.version import parse, Version
+from loguru import logger
 
 
 class Emby(JellyfinEmby):
-    def __init__(self, baseurl, token):
+    def __init__(self, base_url: str, token: str) -> None:
         authorization = (
             "Emby , "
             'Client="JellyPlex-Watched", '
@@ -18,8 +19,14 @@ class Emby(JellyfinEmby):
         }
 
         super().__init__(
-            server_type="Emby", baseurl=baseurl, token=token, headers=headers
+            server_type="Emby", base_url=base_url, token=token, headers=headers
         )
 
     def is_partial_update_supported(self, server_version: Version) -> bool:
-        return server_version > parse("4.4")
+        if not server_version >= parse("4.4"):
+            logger.info(
+                f"{self.server_type}: Server version {server_version} does not support updating playback position.",
+            )
+            return False
+
+        return True

src/functions.py

@@ -16,7 +16,7 @@ def log_marked(
     movie_show: str,
     episode: str | None = None,
     duration: float | None = None,
-):
+) -> None:
     output = f"{server_type}/{server_name}/{username}/{library}/{movie_show}"
 
     if episode:

src/jellyfin.py

@@ -1,9 +1,10 @@
 from src.jellyfin_emby import JellyfinEmby
 from packaging.version import parse, Version
+from loguru import logger
 
 
 class Jellyfin(JellyfinEmby):
-    def __init__(self, baseurl, token):
+    def __init__(self, base_url: str, token: str) -> None:
         authorization = (
             "MediaBrowser , "
             'Client="JellyPlex-Watched", '
@@ -18,8 +19,14 @@ class Jellyfin(JellyfinEmby):
         }
 
         super().__init__(
-            server_type="Jellyfin", baseurl=baseurl, token=token, headers=headers
+            server_type="Jellyfin", base_url=base_url, token=token, headers=headers
         )
 
     def is_partial_update_supported(self, server_version: Version) -> bool:
-        return server_version >= parse("10.9.0")
+        if not server_version >= parse("10.9.0"):
+            logger.info(
+                f"{self.server_type}: Server version {server_version} does not support updating playback position.",
+            )
+            return False
+
+        return True

src/jellyfin_emby.py

@@ -30,28 +30,34 @@ generate_guids = str_to_bool(os.getenv("GENERATE_GUIDS", "True"))
 generate_locations = str_to_bool(os.getenv("GENERATE_LOCATIONS", "True"))
 
 
-def extract_identifiers_from_item(server_type, item: dict) -> MediaIdentifiers:
-    title = item.get("Name", None)
+def extract_identifiers_from_item(
+    server_type: str, item: dict[str, Any]
+) -> MediaIdentifiers:
+    title = item.get("Name")
     id = None
     if not title:
         id = item.get("Id")
-        logger.info(f"{server_type}: Name not found in {id}")
+        logger.info(f"{server_type}: Name not found for {id}")
 
     guids = {}
     if generate_guids:
-        guids = {k.lower(): v for k, v in item["ProviderIds"].items()}
+        guids = {k.lower(): v for k, v in item.get("ProviderIds", {}).items()}
         if not guids:
             logger.info(
                 f"{server_type}: {title if title else id} has no guids",
             )
 
-    locations = tuple()
+    locations: tuple[str, ...] = tuple()
     if generate_locations:
-        if "Path" in item:
-            locations = tuple([item.get("Path").split("/")[-1]])
-        elif "MediaSources" in item:
+        if item.get("Path"):
+            locations = tuple([item["Path"].split("/")[-1]])
+        elif item.get("MediaSources"):
             locations = tuple(
-                [x["Path"].split("/")[-1] for x in item["MediaSources"] if "Path" in x]
+                [
+                    x["Path"].split("/")[-1]
+                    for x in item["MediaSources"]
+                    if x.get("Path")
+                ]
             )
 
         if not locations:
@@ -60,18 +66,20 @@ def extract_identifiers_from_item(server_type, item: dict) -> MediaIdentifiers:
     return MediaIdentifiers(
         title=title,
         locations=locations,
-        imdb_id=guids.get("imdb", None),
-        tvdb_id=guids.get("tvdb", None),
-        tmdb_id=guids.get("tmdb", None),
+        imdb_id=guids.get("imdb"),
+        tvdb_id=guids.get("tvdb"),
+        tmdb_id=guids.get("tmdb"),
     )
 
 
-def get_mediaitem(server_type, item: dict) -> MediaItem:
+def get_mediaitem(server_type: str, item: dict[str, Any]) -> MediaItem:
     return MediaItem(
         identifiers=extract_identifiers_from_item(server_type, item),
         status=WatchedStatus(
-            completed=item["UserData"]["Played"],
-            time=floor(item["UserData"]["PlaybackPositionTicks"] / 10000),
+            completed=item.get("UserData", {}).get("Played"),
+            time=floor(
+                item.get("UserData", {}).get("PlaybackPositionTicks", 0) / 10000
+            ),
         ),
     )
 
@@ -80,27 +88,31 @@ class JellyfinEmby:
     def __init__(
         self,
         server_type: Literal["Jellyfin", "Emby"],
-        baseurl: str,
+        base_url: str,
         token: str,
         headers: dict[str, str],
-    ):
+    ) -> None:
         if server_type not in ["Jellyfin", "Emby"]:
             raise Exception(f"Server type {server_type} not supported")
-        self.server_type = server_type
-        self.baseurl = baseurl
-        self.token = token
-        self.headers = headers
-        self.timeout = int(os.getenv("REQUEST_TIMEOUT", 300))
+        self.server_type: str = server_type
+        self.base_url: str = base_url
+        self.token: str = token
+        self.headers: dict[str, str] = headers
+        self.timeout: int = int(os.getenv("REQUEST_TIMEOUT", 300))
 
-        if not self.baseurl:
-            raise Exception(f"{self.server_type} baseurl not set")
+        if not self.base_url:
+            raise Exception(f"{self.server_type} base_url not set")
 
         if not self.token:
             raise Exception(f"{self.server_type} token not set")
 
         self.session = requests.Session()
-        self.users = self.get_users()
-        self.server_name = self.info(name_only=True)
+        self.users: dict[str, str] = self.get_users()
+        self.server_name: str = self.info(name_only=True)
+        self.server_version: Version = self.info(version_only=True)
+        self.update_partial: bool = self.is_partial_update_supported(
+            self.server_version
+        )
 
     def query(
         self,
@@ -108,15 +120,13 @@ class JellyfinEmby:
         query_type: Literal["get", "post"],
         identifiers: dict[str, str] | None = None,
         json: dict[str, float] | None = None,
-    ) -> dict[str, Any] | list[dict[str, Any]] | None:
+    ) -> list[dict[str, Any]] | dict[str, Any] | None:
         try:
-            results: (
-                dict[str, list[Any] | dict[str, str]] | list[dict[str, Any]] | None
-            ) = None
+            results = None
 
             if query_type == "get":
                 response = self.session.get(
-                    self.baseurl + query, headers=self.headers, timeout=self.timeout
+                    self.base_url + query, headers=self.headers, timeout=self.timeout
                 )
                 if response.status_code not in [200, 204]:
                     raise Exception(
@@ -129,7 +139,7 @@ class JellyfinEmby:
 
             elif query_type == "post":
                 response = self.session.post(
-                    self.baseurl + query,
+                    self.base_url + query,
                     headers=self.headers,
                     json=json,
                     timeout=self.timeout,
@@ -143,12 +153,12 @@ class JellyfinEmby:
                 else:
                     results = response.json()
 
-            if results is not None:
+            if results:
                 if not isinstance(results, list) and not isinstance(results, dict):
                     raise Exception("Query result is not of type list or dict")
 
             # append identifiers to results
-            if identifiers and results:
+            if identifiers and isinstance(results, dict):
                 results["Identifiers"] = identifiers
 
             return results
@@ -165,13 +175,13 @@ class JellyfinEmby:
         try:
             query_string = "/System/Info/Public"
 
-            response: dict[str, Any] = self.query(query_string, "get")
+            response = self.query(query_string, "get")
 
-            if response:
+            if response and isinstance(response, dict):
                 if name_only:
-                    return response["ServerName"]
+                    return response.get("ServerName")
                 elif version_only:
-                    return parse(response["Version"])
+                    return parse(response.get("Version", ""))
 
                 return f"{self.server_type} {response.get('ServerName')}: {response.get('Version')}"
             else:
@@ -186,13 +196,11 @@ class JellyfinEmby:
             users: dict[str, str] = {}
 
             query_string = "/Users"
-            response: list[dict[str, str | bool]] = self.query(query_string, "get")
+            response = self.query(query_string, "get")
 
-            # If response is not empty
-            if response:
+            if response and isinstance(response, list):
                 for user in response:
-                    if isinstance(user["Name"], str) and isinstance(user["Id"], str):
-                        users[user["Name"]] = user["Id"]
+                    users[user["Name"]] = user["Id"]
 
             return users
         except Exception as e:
@@ -201,17 +209,26 @@ class JellyfinEmby:
 
     def get_libraries(self) -> dict[str, str]:
         try:
-            libraries = {}
+            libraries: dict[str, str] = {}
 
             # Theres no way to get all libraries so individually get list of libraries from all users
             users = self.get_users()
 
             for user_name, user_id in users.items():
-                user_libraries: dict = self.query(f"/Users/{user_id}/Views", "get")
-                logger.debug(f"{self.server_type}: All Libraries for {user_name} {[library.get("Name") for library in user_libraries["Items"]]}")
+                user_libraries = self.query(f"/Users/{user_id}/Views", "get")
 
-                for library in user_libraries["Items"]:
-                    library_title = library["Name"]
+                if not user_libraries or not isinstance(user_libraries, dict):
+                    logger.error(
+                        f"{self.server_type}: Failed to get libraries for {user_name}"
+                    )
+                    return libraries
+
+                logger.debug(
+                    f"{self.server_type}: All Libraries for {user_name} {[library.get('Name') for library in user_libraries.get('Items', [])]}"
+                )
+
+                for library in user_libraries.get("Items", []):
+                    library_title = library.get("Name")
                     library_type = library.get("CollectionType")
 
                     if library_type not in ["movies", "tvshows"]:
@@ -228,7 +245,12 @@ class JellyfinEmby:
             raise Exception(e)
 
     def get_user_library_watched(
-        self, user_name, user_id, library_type, library_id, library_title
+        self,
+        user_name: str,
+        user_id: str,
+        library_type: Literal["Movie", "Series", "Episode"],
+        library_id: str,
+        library_title: str,
     ) -> LibraryData:
         user_name = user_name.lower()
         try:
@@ -239,85 +261,104 @@ class JellyfinEmby:
 
             # Movies
             if library_type == "Movie":
+                movie_items = []
                 watched_items = self.query(
                     f"/Users/{user_id}/Items"
                     + f"?ParentId={library_id}&Filters=IsPlayed&IncludeItemTypes=Movie&Recursive=True&Fields=ItemCounts,ProviderIds,MediaSources",
                     "get",
-                ).get("Items", [])
+                )
+
+                if watched_items and isinstance(watched_items, dict):
+                    movie_items += watched_items.get("Items", [])
 
                 in_progress_items = self.query(
                     f"/Users/{user_id}/Items"
                     + f"?ParentId={library_id}&Filters=IsResumable&IncludeItemTypes=Movie&Recursive=True&Fields=ItemCounts,ProviderIds,MediaSources",
                     "get",
-                ).get("Items", [])
+                )
 
-                for movie in watched_items + in_progress_items:
+                if in_progress_items and isinstance(in_progress_items, dict):
+                    movie_items += in_progress_items.get("Items", [])
+
+                for movie in movie_items:
                     # Skip if theres no user data which means the movie has not been watched
-                    if "UserData" not in movie:
+                    if not movie.get("UserData"):
                         continue
 
                     # Skip if theres no media tied to the movie
-                    if "MediaSources" not in movie or movie["MediaSources"] == {}:
+                    if not movie.get("MediaSources"):
                         continue
 
                     # Skip if not watched or watched less than a minute
                     if (
-                        movie["UserData"]["Played"] == True
-                        or movie["UserData"]["PlaybackPositionTicks"] > 600000000
+                        movie["UserData"].get("Played")
+                        or movie["UserData"].get("PlaybackPositionTicks", 0) > 600000000
                     ):
                         watched.movies.append(get_mediaitem(self.server_type, movie))
 
             # TV Shows
             if library_type in ["Series", "Episode"]:
                 # Retrieve a list of watched TV shows
-                watched_shows = self.query(
+                all_shows = self.query(
                     f"/Users/{user_id}/Items"
                     + f"?ParentId={library_id}&isPlaceHolder=false&IncludeItemTypes=Series&Recursive=True&Fields=ProviderIds,Path,RecursiveItemCount",
                     "get",
-                ).get("Items", [])
+                )
+
+                if not all_shows or not isinstance(all_shows, dict):
+                    logger.debug(
+                        f"{self.server_type}: Failed to get shows for {user_name} in {library_title}"
+                    )
+                    return watched
 
                 # Filter the list of shows to only include those that have been partially or fully watched
                 watched_shows_filtered = []
-                for show in watched_shows:
-                    if "UserData" not in show:
+                for show in all_shows.get("Items", []):
+                    if not show.get("UserData"):
                         continue
 
-                    if "PlayedPercentage" in show["UserData"]:
-                        if show["UserData"]["PlayedPercentage"] > 0:
-                            watched_shows_filtered.append(show)
+                    if show["UserData"].get("PlayedPercentage", 0) > 0:
+                        watched_shows_filtered.append(show)
 
                 # Retrieve the watched/partially watched list of episodes of each watched show
                 for show in watched_shows_filtered:
-                    show_guids = {k.lower(): v for k, v in show["ProviderIds"].items()}
+                    show_name = show.get("Name")
+                    show_guids = {
+                        k.lower(): v for k, v in show.get("ProviderIds", {}).items()
+                    }
                     show_locations = (
                         tuple([show["Path"].split("/")[-1]])
-                        if "Path" in show
+                        if show.get("Path")
                         else tuple()
                     )
 
                     show_episodes = self.query(
-                        f"/Shows/{show['Id']}/Episodes"
+                        f"/Shows/{show.get('Id')}/Episodes"
                         + f"?userId={user_id}&isPlaceHolder=false&Fields=ProviderIds,MediaSources",
                         "get",
-                    ).get("Items", [])
+                    )
+
+                    if not show_episodes or not isinstance(show_episodes, dict):
+                        logger.debug(
+                            f"{self.server_type}: Failed to get episodes for {user_name} {library_title} {show_name}"
+                        )
+                        continue
 
                     # Iterate through the episodes
                     # Create a list to store the episodes
                     episode_mediaitem = []
-                    for episode in show_episodes:
-                        if "UserData" not in episode:
+                    for episode in show_episodes.get("Items", []):
+                        if not episode.get("UserData"):
                             continue
 
-                        if (
-                            "MediaSources" not in episode
-                            or episode["MediaSources"] == {}
-                        ):
+                        if not episode.get("MediaSources"):
                             continue
 
                         # If watched or watched more than a minute
                         if (
-                            episode["UserData"]["Played"] == True
-                            or episode["UserData"]["PlaybackPositionTicks"] > 600000000
+                            episode["UserData"].get("Played")
+                            or episode["UserData"].get("PlaybackPositionTicks", 0)
+                            > 600000000
                         ):
                             episode_mediaitem.append(
                                 get_mediaitem(self.server_type, episode)
@@ -329,9 +370,9 @@ class JellyfinEmby:
                                 identifiers=MediaIdentifiers(
                                     title=show.get("Name"),
                                     locations=show_locations,
-                                    imdb_id=show_guids.get("imdb", None),
-                                    tvdb_id=show_guids.get("tvdb", None),
-                                    tmdb_id=show_guids.get("tmdb", None),
+                                    imdb_id=show_guids.get("imdb"),
+                                    tvdb_id=show_guids.get("tvdb"),
+                                    tmdb_id=show_guids.get("tmdb"),
                                 ),
                                 episodes=episode_mediaitem,
                             )
@@ -348,7 +389,7 @@ class JellyfinEmby:
             )
 
             logger.error(traceback.format_exc())
-            return {}
+            return LibraryData(title=library_title)
 
     def get_watched(
         self, users: dict[str, str], sync_libraries: list[str]
@@ -425,9 +466,8 @@ class JellyfinEmby:
         library_data: LibraryData,
         library_name: str,
         library_id: str,
-        update_partial: bool,
         dryrun: bool,
-    ):
+    ) -> None:
         try:
             # If there are no movies or shows to update, exit early.
             if not library_data.series and not library_data.movies:
@@ -445,7 +485,14 @@ class JellyfinEmby:
                     + "&isPlayed=false&Fields=ItemCounts,ProviderIds,MediaSources&IncludeItemTypes=Movie",
                     "get",
                 )
-                for jellyfin_video in jellyfin_search["Items"]:
+
+                if not jellyfin_search or not isinstance(jellyfin_search, dict):
+                    logger.debug(
+                        f"{self.server_type}: Failed to get movies for {user_name} {library_name}"
+                    )
+                    return
+
+                for jellyfin_video in jellyfin_search.get("Items", []):
                     jelly_identifiers = extract_identifiers_from_item(
                         self.server_type, jellyfin_video
                     )
@@ -454,7 +501,7 @@ class JellyfinEmby:
                         if check_same_identifiers(
                             jelly_identifiers, stored_movie.identifiers
                         ):
-                            jellyfin_video_id = jellyfin_video["Id"]
+                            jellyfin_video_id = jellyfin_video.get("Id")
                             if stored_movie.status.completed:
                                 msg = f"{self.server_type}: {jellyfin_video.get('Name')} as watched for {user_name} in {library_name}"
                                 if not dryrun:
@@ -471,11 +518,11 @@ class JellyfinEmby:
                                     library_name,
                                     jellyfin_video.get("Name"),
                                 )
-                            elif update_partial:
+                            elif self.update_partial:
                                 msg = f"{self.server_type}: {jellyfin_video.get('Name')} as partially watched for {floor(stored_movie.status.time / 60_000)} minutes for {user_name} in {library_name}"
 
                                 if not dryrun:
-                                    playback_position_payload = {
+                                    playback_position_payload: dict[str, float] = {
                                         "PlaybackPositionTicks": stored_movie.status.time
                                         * 10_000,
                                     }
@@ -507,7 +554,13 @@ class JellyfinEmby:
                     + "&Fields=ItemCounts,ProviderIds,Path&IncludeItemTypes=Series",
                     "get",
                 )
-                jellyfin_shows = [x for x in jellyfin_search["Items"]]
+                if not jellyfin_search or not isinstance(jellyfin_search, dict):
+                    logger.debug(
+                        f"{self.server_type}: Failed to get shows for {user_name} {library_name}"
+                    )
+                    return
+
+                jellyfin_shows = [x for x in jellyfin_search.get("Items", [])]
 
                 for jellyfin_show in jellyfin_shows:
                     jellyfin_show_identifiers = extract_identifiers_from_item(
@@ -518,19 +571,27 @@ class JellyfinEmby:
                         if check_same_identifiers(
                             jellyfin_show_identifiers, stored_series.identifiers
                         ):
-                            logger.info(
+                            logger.trace(
                                 f"Found matching show for '{jellyfin_show.get('Name')}'",
                             )
                             # Now update episodes.
                             # Get the list of Plex episodes for this show.
-                            jellyfin_show_id = jellyfin_show["Id"]
+                            jellyfin_show_id = jellyfin_show.get("Id")
                             jellyfin_episodes = self.query(
                                 f"/Shows/{jellyfin_show_id}/Episodes"
                                 + f"?userId={user_id}&Fields=ItemCounts,ProviderIds,MediaSources",
                                 "get",
                             )
 
-                            for jellyfin_episode in jellyfin_episodes["Items"]:
+                            if not jellyfin_episodes or not isinstance(
+                                jellyfin_episodes, dict
+                            ):
+                                logger.debug(
+                                    f"{self.server_type}: Failed to get episodes for {user_name} {library_name} {jellyfin_show.get('Name')}"
+                                )
+                                return
+
+                            for jellyfin_episode in jellyfin_episodes.get("Items", []):
                                 jellyfin_episode_identifiers = (
                                     extract_identifiers_from_item(
                                         self.server_type, jellyfin_episode
@@ -541,10 +602,10 @@ class JellyfinEmby:
                                         jellyfin_episode_identifiers,
                                         stored_ep.identifiers,
                                     ):
-                                        jellyfin_episode_id = jellyfin_episode["Id"]
+                                        jellyfin_episode_id = jellyfin_episode.get("Id")
                                         if stored_ep.status.completed:
                                             msg = (
-                                                f"{self.server_type}: {jellyfin_episode['SeriesName']} {jellyfin_episode['SeasonName']} Episode {jellyfin_episode.get('IndexNumber')} {jellyfin_episode.get('Name')}"
+                                                f"{self.server_type}: {jellyfin_episode.get('SeriesName')} {jellyfin_episode.get('SeasonName')} Episode {jellyfin_episode.get('IndexNumber')} {jellyfin_episode.get('Name')}"
                                                 + f" as watched for {user_name} in {library_name}"
                                             )
                                             if not dryrun:
@@ -564,9 +625,9 @@ class JellyfinEmby:
                                                 jellyfin_episode.get("SeriesName"),
                                                 jellyfin_episode.get("Name"),
                                             )
-                                        elif update_partial:
+                                        elif self.update_partial:
                                             msg = (
-                                                f"{self.server_type}: {jellyfin_episode['SeriesName']} {jellyfin_episode['SeasonName']} Episode {jellyfin_episode.get('IndexNumber')} {jellyfin_episode.get('Name')}"
+                                                f"{self.server_type}: {jellyfin_episode.get('SeriesName')} {jellyfin_episode.get('SeasonName')} Episode {jellyfin_episode.get('IndexNumber')} {jellyfin_episode.get('Name')}"
                                                 + f" as partially watched for {floor(stored_ep.status.time / 60_000)} minutes for {user_name} in {library_name}"
                                             )
 
@@ -614,19 +675,11 @@ class JellyfinEmby:
     def update_watched(
         self,
         watched_list: dict[str, UserData],
-        user_mapping=None,
-        library_mapping=None,
-        dryrun=False,
-    ):
+        user_mapping: dict[str, str] | None = None,
+        library_mapping: dict[str, str] | None = None,
+        dryrun: bool = False,
+    ) -> None:
         try:
-            server_version = self.info(version_only=True)
-            update_partial = self.is_partial_update_supported(server_version)
-
-            if not update_partial:
-                logger.info(
-                    f"{self.server_type}: Server version {server_version} does not support updating playback position.",
-                )
-
             for user, user_data in watched_list.items():
                 user_other = None
                 user_name = None
@@ -647,7 +700,7 @@ class JellyfinEmby:
                         user_name = key
                         break
 
-                if not user_id:
+                if not user_id or not user_name:
                     logger.info(f"{user} {user_other} not found in Jellyfin")
                     continue
 
@@ -655,7 +708,14 @@ class JellyfinEmby:
                     f"/Users/{user_id}/Views",
                     "get",
                 )
-                jellyfin_libraries = [x for x in jellyfin_libraries["Items"]]
+
+                if not jellyfin_libraries or not isinstance(jellyfin_libraries, dict):
+                    logger.debug(
+                        f"{self.server_type}: Failed to get libraries for {user_name}"
+                    )
+                    continue
+
+                jellyfin_libraries = [x for x in jellyfin_libraries.get("Items", [])]
 
                 for library_name in user_data.libraries:
                     library_data = user_data.libraries[library_name]
@@ -703,7 +763,6 @@ class JellyfinEmby:
                             library_data,
                             library_name,
                             library_id,
-                            update_partial,
                             dryrun,
                         )
 

src/library.py

@@ -5,6 +5,10 @@ from src.functions import (
     search_mapping,
 )
 
+from src.emby import Emby
+from src.jellyfin import Jellyfin
+from src.plex import Plex
+
 
 def check_skip_logic(
     library_title: str,
@@ -54,7 +58,7 @@ def check_blacklist_logic(
     blacklist_library: list[str],
     blacklist_library_type: list[str],
     library_other: str | None = None,
-):
+) -> str | None:
     skip_reason = None
     if isinstance(library_type, (list, tuple, set)):
         for library_type_item in library_type:
@@ -90,7 +94,7 @@ def check_whitelist_logic(
     whitelist_library: list[str],
     whitelist_library_type: list[str],
     library_other: str | None = None,
-):
+) -> str | None:
     skip_reason = None
     if len(whitelist_library_type) > 0:
         if isinstance(library_type, (list, tuple, set)):
@@ -161,8 +165,8 @@ def filter_libaries(
 
 
 def setup_libraries(
-    server_1,
-    server_2,
+    server_1: Plex | Jellyfin | Emby,
+    server_2: Plex | Jellyfin | Emby,
     blacklist_library: list[str],
     blacklist_library_type: list[str],
     whitelist_library: list[str],

src/main.py

@@ -27,7 +27,7 @@ log_file = os.getenv("LOG_FILE", os.getenv("LOGFILE", "log.log"))
 level = os.getenv("DEBUG_LEVEL", "INFO").upper()
 
 
-def configure_logger():
+def configure_logger() -> None:
     # Remove default logger to configure our own
     logger.remove()
 
@@ -111,18 +111,20 @@ def should_sync_server(
     return True
 
 
-def main_loop():
+def main_loop() -> None:
     dryrun = str_to_bool(os.getenv("DRYRUN", "False"))
     logger.info(f"Dryrun: {dryrun}")
 
-    user_mapping = os.getenv("USER_MAPPING", None)
-    if user_mapping:
-        user_mapping = json.loads(user_mapping.lower())
+    user_mapping_env = os.getenv("USER_MAPPING", None)
+    user_mapping = None
+    if user_mapping_env:
+        user_mapping = json.loads(user_mapping_env.lower())
     logger.info(f"User Mapping: {user_mapping}")
 
-    library_mapping = os.getenv("LIBRARY_MAPPING", None)
-    if library_mapping:
-        library_mapping = json.loads(library_mapping)
+    library_mapping_env = os.getenv("LIBRARY_MAPPING", None)
+    library_mapping = None
+    if library_mapping_env:
+        library_mapping = json.loads(library_mapping_env)
     logger.info(f"Library Mapping: {library_mapping}")
 
     # Create (black/white)lists
@@ -241,7 +243,7 @@ def main_loop():
 
 
 @logger.catch
-def main():
+def main() -> None:
     run_only_once = str_to_bool(os.getenv("RUN_ONLY_ONCE", "False"))
     sleep_duration = float(os.getenv("SLEEP_DURATION", "3600"))
     times: list[float] = []

src/plex.py

@@ -10,7 +10,8 @@ from requests.adapters import HTTPAdapter as RequestsHTTPAdapter
 
 from plexapi.video import Show, Episode, Movie
 from plexapi.server import PlexServer
-from plexapi.myplex import MyPlexAccount
+from plexapi.myplex import MyPlexAccount, MyPlexUser
+from plexapi.library import MovieSection, ShowSection
 
 from src.functions import (
     search_mapping,
@@ -35,7 +36,9 @@ generate_locations = str_to_bool(os.getenv("GENERATE_LOCATIONS", "True"))
 
 # Bypass hostname validation for ssl. Taken from https://github.com/pkkid/python-plexapi/issues/143#issuecomment-775485186
 class HostNameIgnoringAdapter(RequestsHTTPAdapter):
-    def init_poolmanager(self, connections, maxsize, block=..., **pool_kwargs):
+    def init_poolmanager(
+        self, connections: int, maxsize: int | None, block=..., **pool_kwargs
+    ) -> None:
         self.poolmanager = PoolManager(
             num_pools=connections,
             maxsize=maxsize,
@@ -53,7 +56,7 @@ def extract_guids_from_item(item: Movie | Show | Episode) -> dict[str, str]:
     guids: dict[str, str] = dict(
         guid.id.split("://")
         for guid in item.guids
-        if guid.id is not None and len(guid.id.strip()) > 0
+        if guid.id and len(guid.id.strip()) > 0
     )
 
     return guids
@@ -69,13 +72,13 @@ def extract_identifiers_from_item(item: Movie | Show | Episode) -> MediaIdentifi
             if generate_locations
             else tuple()
         ),
-        imdb_id=guids.get("imdb", None),
-        tvdb_id=guids.get("tvdb", None),
-        tmdb_id=guids.get("tmdb", None),
+        imdb_id=guids.get("imdb"),
+        tvdb_id=guids.get("tvdb"),
+        tmdb_id=guids.get("tmdb"),
     )
 
 
-def get_mediaitem(item: Movie | Episode, completed=True) -> MediaItem:
+def get_mediaitem(item: Movie | Episode, completed: bool) -> MediaItem:
     return MediaItem(
         identifiers=extract_identifiers_from_item(item),
         status=WatchedStatus(completed=completed, time=item.viewOffset),
@@ -88,7 +91,7 @@ def update_user_watched(
     library_data: LibraryData,
     library_name: str,
     dryrun: bool,
-):
+) -> None:
     try:
         # If there are no movies or shows to update, exit early.
         if not library_data.series and not library_data.movies:
@@ -115,6 +118,7 @@ def update_user_watched(
                             msg = f"Plex: {plex_movie.title} as watched for {user.title} in {library_name}"
                             if not dryrun:
                                 plex_movie.markWatched()
+
                             logger.success(f"{'[DRYRUN] ' if dryrun else ''}{msg}")
                             log_marked(
                                 "Plex",
@@ -154,7 +158,7 @@ def update_user_watched(
                     if check_same_identifiers(
                         plex_show_identifiers, stored_series.identifiers
                     ):
-                        logger.info(f"Found matching show for '{plex_show.title}'")
+                        logger.trace(f"Found matching show for '{plex_show.title}'")
                         # Now update episodes.
                         # Get the list of Plex episodes for this show.
                         plex_episodes = plex_show.episodes()
@@ -216,46 +220,53 @@ def update_user_watched(
 class Plex:
     def __init__(
         self,
-        baseurl=None,
-        token=None,
-        username=None,
-        password=None,
-        servername=None,
-        ssl_bypass=False,
-        session=None,
-    ):
-        self.server_type = "Plex"
-        self.baseurl = baseurl
-        self.token = token
-        self.username = username
-        self.password = password
-        self.servername = servername
-        self.ssl_bypass = ssl_bypass
+        base_url: str | None = None,
+        token: str | None = None,
+        user_name: str | None = None,
+        password: str | None = None,
+        server_name: str | None = None,
+        ssl_bypass: bool = False,
+        session: requests.Session | None = None,
+    ) -> None:
+        self.server_type: str = "Plex"
+        self.ssl_bypass: bool = ssl_bypass
         if ssl_bypass:
             # Session for ssl bypass
             session = requests.Session()
             # By pass ssl hostname check https://github.com/pkkid/python-plexapi/issues/143#issuecomment-775485186
             session.mount("https://", HostNameIgnoringAdapter())
         self.session = session
-        self.plex = self.login(self.baseurl, self.token)
-        self.admin_user = self.plex.myPlexAccount()
-        self.users = self.get_users()
+        self.plex: PlexServer = self.login(
+            base_url, token, user_name, password, server_name
+        )
 
-    def login(self, baseurl, token):
+        self.base_url: str = self.plex._baseurl
+
+        self.admin_user: MyPlexAccount = self.plex.myPlexAccount()
+        self.users: list[MyPlexUser | MyPlexAccount] = self.get_users()
+
+    def login(
+        self,
+        base_url: str | None,
+        token: str | None,
+        user_name: str | None,
+        password: str | None,
+        server_name: str | None,
+    ) -> PlexServer:
         try:
-            if baseurl and token:
-                plex = PlexServer(baseurl, token, session=self.session)
-            elif self.username and self.password and self.servername:
+            if base_url and token:
+                plex: PlexServer = PlexServer(base_url, token, session=self.session)
+            elif user_name and password and server_name:
                 # Login via plex account
-                account = MyPlexAccount(self.username, self.password)
-                plex = account.resource(self.servername).connect()
+                account = MyPlexAccount(user_name, password)
+                plex = account.resource(server_name).connect()
             else:
                 raise Exception("No complete plex credentials provided")
 
             return plex
         except Exception as e:
-            if self.username:
-                msg = f"Failed to login via plex account {self.username}"
+            if user_name:
+                msg = f"Failed to login via plex account {user_name}"
                 logger.error(f"Plex: Failed to login, {msg}, Error: {e}")
             else:
                 logger.error(f"Plex: Failed to login, Error: {e}")
@@ -264,9 +275,9 @@ class Plex:
     def info(self) -> str:
         return f"Plex {self.plex.friendlyName}: {self.plex.version}"
 
-    def get_users(self):
+    def get_users(self) -> list[MyPlexUser | MyPlexAccount]:
         try:
-            users = self.plex.myPlexAccount().users()
+            users: list[MyPlexUser | MyPlexAccount] = self.plex.myPlexAccount().users()
 
             # append self to users
             users.append(self.plex.myPlexAccount())
@@ -302,7 +313,9 @@ class Plex:
             logger.error(f"Plex: Failed to get libraries, Error: {e}")
             raise Exception(e)
 
-    def get_user_library_watched(self, user_name, user_plex, library) -> LibraryData:
+    def get_user_library_watched(
+        self, user_name: str, user_plex: PlexServer, library: MovieSection | ShowSection
+    ) -> LibraryData:
         try:
             logger.info(
                 f"Plex: Generating watched for {user_name} in library {library.title}",
@@ -353,9 +366,9 @@ class Plex:
                                         if generate_locations
                                         else tuple()
                                     ),
-                                    imdb_id=show_guids.get("imdb", None),
-                                    tvdb_id=show_guids.get("tvdb", None),
-                                    tmdb_id=show_guids.get("tmdb", None),
+                                    imdb_id=show_guids.get("imdb"),
+                                    tvdb_id=show_guids.get("tvdb"),
+                                    tmdb_id=show_guids.get("tmdb"),
                                 ),
                                 episodes=episode_mediaitem,
                             )
@@ -369,7 +382,9 @@ class Plex:
             )
             return LibraryData(title=library.title)
 
-    def get_watched(self, users, sync_libraries) -> dict[str, UserData]:
+    def get_watched(
+        self, users: list[MyPlexUser | MyPlexAccount], sync_libraries: list[str]
+    ) -> dict[str, UserData]:
         try:
             users_watched: dict[str, UserData] = {}
 
@@ -379,10 +394,7 @@ class Plex:
                 else:
                     token = user.get_token(self.plex.machineIdentifier)
                     if token:
-                        user_plex = self.login(
-                            self.plex._baseurl,
-                            token,
-                        )
+                        user_plex = self.login(self.base_url, token, None, None, None)
                     else:
                         logger.error(
                             f"Plex: Failed to get token for {user.title}, skipping",
@@ -416,10 +428,10 @@ class Plex:
     def update_watched(
         self,
         watched_list: dict[str, UserData],
-        user_mapping=None,
-        library_mapping=None,
-        dryrun=False,
-    ):
+        user_mapping: dict[str, str] | None = None,
+        library_mapping: dict[str, str] | None = None,
+        dryrun: bool = False,
+    ) -> None:
         try:
             for user, user_data in watched_list.items():
                 user_other = None
@@ -445,15 +457,19 @@ class Plex:
                     user_plex = self.plex
                 else:
                     if isinstance(user, str):
-                        logger.warning(
+                        logger.debug(
                             f"Plex: {user} is not a plex object, attempting to get object for user",
                         )
                         user = self.plex.myPlexAccount().user(user)
 
+                    if not isinstance(user, MyPlexUser):
+                        logger.error(f"Plex: {user} failed to get PlexUser")
+                        continue
+
                     token = user.get_token(self.plex.machineIdentifier)
                     if token:
                         user_plex = PlexServer(
-                            self.plex._baseurl,
+                            self.base_url,
                             token,
                             session=self.session,
                         )
@@ -463,6 +479,10 @@ class Plex:
                         )
                         continue
 
+                if not user_plex:
+                    logger.error(f"Plex: {user} Failed to get PlexServer")
+                    continue
+
                 for library_name in user_data.libraries:
                     library_data = user_data.libraries[library_name]
                     library_other = None

src/users.py

@@ -1,4 +1,4 @@
-from plexapi.myplex import MyPlexAccount
+from plexapi.myplex import MyPlexAccount, MyPlexUser
 from loguru import logger
 
 from src.emby import Emby
@@ -109,7 +109,10 @@ def setup_users(
     blacklist_users: list[str],
     whitelist_users: list[str],
     user_mapping: dict[str, str] | None = None,
-) -> tuple[list[MyPlexAccount] | dict[str, str], list[MyPlexAccount] | dict[str, str]]:
+) -> tuple[
+    list[MyPlexAccount | MyPlexUser] | dict[str, str],
+    list[MyPlexAccount | MyPlexUser] | dict[str, str],
+]:
     server_1_users = generate_user_list(server_1)
     server_2_users = generate_user_list(server_2)
     logger.debug(f"Server 1 users: {server_1_users}")

src/watched.py

@@ -1,6 +1,7 @@
 import copy
 from pydantic import BaseModel, Field
 from loguru import logger
+from typing import Any
 
 from src.functions import search_mapping
 
@@ -103,8 +104,8 @@ def check_remove_entry(item1: MediaItem, item2: MediaItem) -> bool:
 def cleanup_watched(
     watched_list_1: dict[str, UserData],
     watched_list_2: dict[str, UserData],
-    user_mapping=None,
-    library_mapping=None,
+    user_mapping: dict[str, str] | None = None,
+    library_mapping: dict[str, str] | None = None,
 ) -> dict[str, UserData]:
     modified_watched_list_1 = copy.deepcopy(watched_list_1)
 
@@ -199,11 +200,17 @@ def cleanup_watched(
     return modified_watched_list_1
 
 
-def get_other(watched_list, object_1, object_2):
+def get_other(
+    watched_list: dict[str, Any], object_1: str, object_2: str | None
+) -> str | None:
     if object_1 in watched_list:
         return object_1
-    elif object_2 in watched_list:
+
+    if object_2 and object_2 in watched_list:
         return object_2
-    else:
-        logger.info(f"{object_1} and {object_2} not found in watched list 2")
-        return None
+
+    logger.info(
+        f"{object_1}{' and ' + object_2 if object_2 else ''} not found in watched list 2"
+    )
+
+    return None

uv.lock

@@ -1,5 +1,4 @@
 version = 1
-revision = 1
 requires-python = ">=3.12"
 
 [[package]]
@@ -97,7 +96,9 @@ dependencies = [
 
 [package.dev-dependencies]
 dev = [
+    { name = "mypy" },
     { name = "pytest" },
+    { name = "types-requests" },
 ]
 lint = [
     { name = "ruff" },
@@ -114,7 +115,11 @@ requires-dist = [
 ]
 
 [package.metadata.requires-dev]
-dev = [{ name = "pytest", specifier = ">=8.3.4" }]
+dev = [
+    { name = "mypy", specifier = ">=1.15.0" },
+    { name = "pytest", specifier = ">=8.3.4" },
+    { name = "types-requests", specifier = ">=2.32.0.20250306" },
+]
 lint = [{ name = "ruff", specifier = ">=0.9.6" }]
 
 [[package]]
@@ -130,6 +135,40 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/0c/29/0348de65b8cc732daa3e33e67806420b2ae89bdce2b04af740289c5c6c8c/loguru-0.7.3-py3-none-any.whl", hash = "sha256:31a33c10c8e1e10422bfd431aeb5d351c7cf7fa671e3c4df004162264b28220c", size = 61595 },
 ]
 
+[[package]]
+name = "mypy"
+version = "1.15.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "mypy-extensions" },
+    { name = "typing-extensions" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/ce/43/d5e49a86afa64bd3839ea0d5b9c7103487007d728e1293f52525d6d5486a/mypy-1.15.0.tar.gz", hash = "sha256:404534629d51d3efea5c800ee7c42b72a6554d6c400e6a79eafe15d11341fd43", size = 3239717 }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/98/3a/03c74331c5eb8bd025734e04c9840532226775c47a2c39b56a0c8d4f128d/mypy-1.15.0-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:aea39e0583d05124836ea645f412e88a5c7d0fd77a6d694b60d9b6b2d9f184fd", size = 10793981 },
+    { url = "https://files.pythonhosted.org/packages/f0/1a/41759b18f2cfd568848a37c89030aeb03534411eef981df621d8fad08a1d/mypy-1.15.0-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:2f2147ab812b75e5b5499b01ade1f4a81489a147c01585cda36019102538615f", size = 9749175 },
+    { url = "https://files.pythonhosted.org/packages/12/7e/873481abf1ef112c582db832740f4c11b2bfa510e829d6da29b0ab8c3f9c/mypy-1.15.0-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:ce436f4c6d218a070048ed6a44c0bbb10cd2cc5e272b29e7845f6a2f57ee4464", size = 11455675 },
+    { url = "https://files.pythonhosted.org/packages/b3/d0/92ae4cde706923a2d3f2d6c39629134063ff64b9dedca9c1388363da072d/mypy-1.15.0-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:8023ff13985661b50a5928fc7a5ca15f3d1affb41e5f0a9952cb68ef090b31ee", size = 12410020 },
+    { url = "https://files.pythonhosted.org/packages/46/8b/df49974b337cce35f828ba6fda228152d6db45fed4c86ba56ffe442434fd/mypy-1.15.0-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:1124a18bc11a6a62887e3e137f37f53fbae476dc36c185d549d4f837a2a6a14e", size = 12498582 },
+    { url = "https://files.pythonhosted.org/packages/13/50/da5203fcf6c53044a0b699939f31075c45ae8a4cadf538a9069b165c1050/mypy-1.15.0-cp312-cp312-win_amd64.whl", hash = "sha256:171a9ca9a40cd1843abeca0e405bc1940cd9b305eaeea2dda769ba096932bb22", size = 9366614 },
+    { url = "https://files.pythonhosted.org/packages/6a/9b/fd2e05d6ffff24d912f150b87db9e364fa8282045c875654ce7e32fffa66/mypy-1.15.0-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:93faf3fdb04768d44bf28693293f3904bbb555d076b781ad2530214ee53e3445", size = 10788592 },
+    { url = "https://files.pythonhosted.org/packages/74/37/b246d711c28a03ead1fd906bbc7106659aed7c089d55fe40dd58db812628/mypy-1.15.0-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:811aeccadfb730024c5d3e326b2fbe9249bb7413553f15499a4050f7c30e801d", size = 9753611 },
+    { url = "https://files.pythonhosted.org/packages/a6/ac/395808a92e10cfdac8003c3de9a2ab6dc7cde6c0d2a4df3df1b815ffd067/mypy-1.15.0-cp313-cp313-manylinux_2_17_aarch64.manylinux2014_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:98b7b9b9aedb65fe628c62a6dc57f6d5088ef2dfca37903a7d9ee374d03acca5", size = 11438443 },
+    { url = "https://files.pythonhosted.org/packages/d2/8b/801aa06445d2de3895f59e476f38f3f8d610ef5d6908245f07d002676cbf/mypy-1.15.0-cp313-cp313-manylinux_2_17_x86_64.manylinux2014_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:c43a7682e24b4f576d93072216bf56eeff70d9140241f9edec0c104d0c515036", size = 12402541 },
+    { url = "https://files.pythonhosted.org/packages/c7/67/5a4268782eb77344cc613a4cf23540928e41f018a9a1ec4c6882baf20ab8/mypy-1.15.0-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:baefc32840a9f00babd83251560e0ae1573e2f9d1b067719479bfb0e987c6357", size = 12494348 },
+    { url = "https://files.pythonhosted.org/packages/83/3e/57bb447f7bbbfaabf1712d96f9df142624a386d98fb026a761532526057e/mypy-1.15.0-cp313-cp313-win_amd64.whl", hash = "sha256:b9378e2c00146c44793c98b8d5a61039a048e31f429fb0eb546d93f4b000bedf", size = 9373648 },
+    { url = "https://files.pythonhosted.org/packages/09/4e/a7d65c7322c510de2c409ff3828b03354a7c43f5a8ed458a7a131b41c7b9/mypy-1.15.0-py3-none-any.whl", hash = "sha256:5469affef548bd1895d86d3bf10ce2b44e33d86923c29e4d675b3e323437ea3e", size = 2221777 },
+]
+
+[[package]]
+name = "mypy-extensions"
+version = "1.0.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/98/a4/1ab47638b92648243faf97a5aeb6ea83059cc3624972ab6b8d2316078d3f/mypy_extensions-1.0.0.tar.gz", hash = "sha256:75dbf8955dc00442a438fc4d0666508a9a97b6bd41aa2f0ffe9d2f2725af0782", size = 4433 }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/2a/e2/5d3f6ada4297caebe1a2add3b126fe800c96f56dbe5d1988a2cbe0b267aa/mypy_extensions-1.0.0-py3-none-any.whl", hash = "sha256:4392f6c0eb8a5668a69e23d168ffa70f0be9ccfd32b5cc2d26a34ae5b844552d", size = 4695 },
+]
+
 [[package]]
 name = "packaging"
 version = "24.2"
@@ -277,6 +316,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/e8/a8/d71f44b93e3aa86ae232af1f2126ca7b95c0f515ec135462b3e1f351441c/ruff-0.9.6-py3-none-win_arm64.whl", hash = "sha256:0e2bb706a2be7ddfea4a4af918562fdc1bcb16df255e5fa595bbd800ce322a5a", size = 10177499 },
 ]
 
+[[package]]
+name = "types-requests"
+version = "2.32.0.20250306"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "urllib3" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/09/1a/beaeff79ef9efd186566ba5f0d95b44ae21f6d31e9413bcfbef3489b6ae3/types_requests-2.32.0.20250306.tar.gz", hash = "sha256:0962352694ec5b2f95fda877ee60a159abdf84a0fc6fdace599f20acb41a03d1", size = 23012 }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/99/26/645d89f56004aa0ba3b96fec27793e3c7e62b40982ee069e52568922b6db/types_requests-2.32.0.20250306-py3-none-any.whl", hash = "sha256:25f2cbb5c8710b2022f8bbee7b2b66f319ef14aeea2f35d80f18c9dbf3b60a0b", size = 20673 },
+]
+
 [[package]]
 name = "typing-extensions"
 version = "4.12.2"