Merge pull request #98 from NodeJSmith/feature/add_get_class_from_booking

Feature/add get class from booking

Author: NodeJSmith

.bumpversion.toml

@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.14.1"
+current_version = "0.15.0"
 
 parse = "(?P<major>\\d+)\\.(?P<minor>\\d+)\\.(?P<patch>\\d+)(?:-(?P<rc_l>rc)(?P<rc>0|[1-9]\\d*))?"
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "otf-api"
-version = "0.14.1"
+version = "0.15.0"
 description = "Python OrangeTheory Fitness API Client"
 authors = [{ name = "Jessica Smith", email = "[email protected]" }]
 requires-python = ">=3.11"

source/conf.py

@@ -14,7 +14,7 @@
 project = "OrangeTheory API"
 copyright = "2025, Jessica Smith"
 author = "Jessica Smith"
-release = "0.14.1"
+release = "0.15.0"
 
 # -- General configuration ---------------------------------------------------
 # https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration

src/otf_api/__init__.py

@@ -36,7 +36,7 @@ def _setup_logging() -> None:
 
 _setup_logging()
 
-__version__ = "0.14.1"
+__version__ = "0.15.0"
 
 
 __all__ = ["Otf", "OtfUser", "models"]

src/otf_api/api/bookings/booking_api.py

@@ -80,11 +80,37 @@ def get_bookings_new(
             ends_before=end_date, starts_after=start_date, include_canceled=include_canceled, expand=expand
         )
 
-        results = [models.BookingV2.create(**b, api=self.otf) for b in bookings_resp]
+        # filter out bookings with ids that start with "no-booking-id"
+        # no idea what these are, but I am praying for the poor sap stuck with maintaining OTF's data model
+        results: list[models.BookingV2] = []
+
+        for b in bookings_resp:
+            if not b.get("id", "").startswith("no-booking-id"):
+                try:
+                    results.append(models.BookingV2.create(**b, api=self.otf))
+                except ValueError as e:
+                    LOGGER.warning(f"Failed to create BookingV2 from response: {e}. Booking data:\n{b}")
+                    continue
 
         if not remove_duplicates:
             return results
 
+        results = self._deduplicate_bookings(results, exclude_cancelled=exclude_cancelled)
+
+        return results
+
+    def _deduplicate_bookings(
+        self, results: list[models.BookingV2], exclude_cancelled: bool = True
+    ) -> list[models.BookingV2]:
+        """Deduplicate bookings by class_id, keeping the most recent booking.
+
+        Args:
+            results (list[BookingV2]): The list of bookings to deduplicate.
+            exclude_cancelled (bool): If True, will not include cancelled bookings in the results.
+
+        Returns:
+            list[BookingV2]: The deduplicated list of bookings.
+        """
         # remove duplicates by class_id, keeping the one with the most recent updated_at timestamp
         seen_classes: dict[str, models.BookingV2] = {}
 
@@ -188,7 +214,11 @@ def get_classes(
         for c in classes_resp:
             c["studio"] = studio_dict[c["studio"]["id"]]  # the one (?) place where ID actually means UUID
             c["is_home_studio"] = c["studio"].studio_uuid == self.otf.home_studio_uuid
-            classes.append(models.OtfClass.create(**c, api=self.otf))
+            try:
+                classes.append(models.OtfClass.create(**c, api=self.otf))
+            except ValueError as e:
+                LOGGER.warning(f"Failed to create OtfClass from response: {e}. Class data:\n{c}")
+                continue
 
         # additional data filtering and enrichment
 
@@ -240,7 +270,7 @@ def get_booking_from_class(self, otf_class: str | models.OtfClass) -> models.Boo
             Booking: The booking.
 
         Raises:
-            BookingNotFoundError: If the booking does not exist.
+            ResourceNotFoundError: If the booking does not exist.
             ValueError: If class_uuid is None or empty string.
         """
         class_uuid = utils.get_class_uuid(otf_class)
@@ -250,7 +280,7 @@ def get_booking_from_class(self, otf_class: str | models.OtfClass) -> models.Boo
         if booking := next((b for b in all_bookings if b.class_uuid == class_uuid), None):
             return booking
 
-        raise exc.BookingNotFoundError(f"Booking for class {class_uuid} not found.")
+        raise exc.ResourceNotFoundError(f"Booking for class {class_uuid} not found.")
 
     def get_booking_from_class_new(self, otf_class: str | models.OtfClass | models.BookingV2Class) -> models.BookingV2:
         """Get a specific booking by class_uuid or OtfClass object.
@@ -262,7 +292,7 @@ def get_booking_from_class_new(self, otf_class: str | models.OtfClass | models.B
             BookingV2: The booking.
 
         Raises:
-            BookingNotFoundError: If the booking does not exist.
+            ResourceNotFoundError: If the booking does not exist.
             ValueError: If class_uuid is None or empty string.
         """
         class_uuid = utils.get_class_uuid(otf_class)
@@ -272,7 +302,78 @@ def get_booking_from_class_new(self, otf_class: str | models.OtfClass | models.B
         if booking := next((b for b in all_bookings if b.class_uuid == class_uuid), None):
             return booking
 
-        raise exc.BookingNotFoundError(f"Booking for class {class_uuid} not found.")
+        raise exc.ResourceNotFoundError(f"Booking for class {class_uuid} not found.")
+
+    def get_class_from_booking(self, booking: models.Booking | models.BookingV2) -> models.OtfClass:
+        """Get the class details from a Booking or BookingV2 object.
+
+        Args:
+            booking (Booking | BookingV2): The booking to get the class details from.
+
+        Returns:
+            OtfClass: The class details.
+
+        Raises:
+            ValueError: If the booking does not have a class_id.
+        """
+        if isinstance(booking, models.BookingV2):
+            return self.get_class_from_booking_new(booking)
+
+        if not booking.otf_class.class_uuid:
+            raise ValueError("Booking does not have a class_uuid")
+
+        if not booking.otf_class.studio:
+            LOGGER.warning("Booking does not have a studio, will attempt to use the home studio to get class details.")
+            studio_uuid = self.otf.home_studio_uuid
+        else:
+            studio_uuid = booking.otf_class.studio.studio_uuid
+
+        classes = self.otf.bookings.get_classes(
+            start_date=booking.starts_at.date(),
+            end_date=booking.starts_at.date(),
+            studio_uuids=[studio_uuid],
+        )
+        if classes:
+            otf_class = next((c for c in classes if c.class_uuid == booking.otf_class.class_uuid), None)
+            if otf_class:
+                return otf_class
+
+        raise exc.ResourceNotFoundError(
+            f"Class for booking {booking.otf_class.name} ({booking.booking_uuid}) not found."
+        )
+
+    def get_class_from_booking_new(self, booking: models.BookingV2) -> models.OtfClass:
+        """Get the class details from a BookingV2 object.
+
+        Args:
+            booking (BookingV2): The booking to get the class details from.
+
+        Returns:
+            OtfClass: The class details.
+
+        Raises:
+            ValueError: If the booking does not have a class_id.
+        """
+        if not booking.otf_class.class_id:
+            raise ValueError("Booking does not have a class_id")
+
+        if not booking.otf_class.studio:
+            LOGGER.warning("Booking does not have a studio, will attempt to use the home studio to get class details.")
+            studio_uuid = self.otf.home_studio_uuid
+        else:
+            studio_uuid = booking.otf_class.studio.studio_uuid
+
+        classes = self.otf.bookings.get_classes(
+            start_date=booking.starts_at.date(),
+            end_date=booking.starts_at.date(),
+            studio_uuids=[studio_uuid],
+        )
+        if classes:
+            otf_class = next((c for c in classes if c.class_id == booking.otf_class.class_id), None)
+            if otf_class:
+                return otf_class
+
+        raise exc.ResourceNotFoundError(f"Class for booking {booking.otf_class.name} ({booking.booking_id}) not found.")
 
     def book_class(self, otf_class: str | models.OtfClass) -> models.Booking:
         """Book a class by providing either the class_uuid or the OtfClass object.
@@ -287,7 +388,7 @@ def book_class(self, otf_class: str | models.OtfClass) -> models.Booking:
             AlreadyBookedError: If the class is already booked.
             OutsideSchedulingWindowError: If the class is outside the scheduling window.
             ValueError: If class_uuid is None or empty string.
-            OtfException: If there is an error booking the class.
+            OtfError: If there is an error booking the class.
         """
         class_uuid = utils.get_class_uuid(otf_class)
 
@@ -297,7 +398,7 @@ def book_class(self, otf_class: str | models.OtfClass) -> models.Booking:
                 raise exc.AlreadyBookedError(
                     f"Class {class_uuid} is already booked.", booking_uuid=existing_booking.booking_uuid
                 )
-        except exc.BookingNotFoundError:
+        except exc.ResourceNotFoundError:
             pass
 
         if isinstance(otf_class, models.OtfClass):
@@ -328,7 +429,7 @@ def book_class_new(self, class_id: str | models.BookingV2Class) -> models.Bookin
             BookingV2: The booking.
 
         Raises:
-            OtfException: If there is an error booking the class.
+            OtfError: If there is an error booking the class.
             TypeError: If the input is not a string or BookingV2Class.
         """
         class_id = utils.get_class_id(class_id)
@@ -349,7 +450,7 @@ def cancel_booking(self, booking: str | models.Booking) -> None:
 
         Raises:
             ValueError: If booking_uuid is None or empty string
-            BookingNotFoundError: If the booking does not exist.
+            ResourceNotFoundError: If the booking does not exist.
         """
         if isinstance(booking, models.BookingV2):
             LOGGER.warning("BookingV2 object provided, using the new cancel booking endpoint (`cancel_booking_new`)")
@@ -370,7 +471,7 @@ def cancel_booking_new(self, booking: str | models.BookingV2) -> None:
 
         Raises:
             ValueError: If booking_id is None or empty string
-            BookingNotFoundError: If the booking does not exist.
+            ResourceNotFoundError: If the booking does not exist.
         """
         if isinstance(booking, models.Booking):
             LOGGER.warning("Booking object provided, using the old cancel booking endpoint (`cancel_booking`)")
@@ -443,7 +544,15 @@ def get_bookings(
             b["class"]["studio"] = studios[b["class"]["studio"]["studioUUId"]]
             b["is_home_studio"] = b["class"]["studio"].studio_uuid == self.otf.home_studio_uuid
 
-        bookings = [models.Booking.create(**b, api=self.otf) for b in resp]
+        bookings: list[models.Booking] = []
+
+        for b in resp:
+            try:
+                bookings.append(models.Booking.create(**b, api=self.otf))
+            except ValueError as e:
+                LOGGER.warning(f"Failed to create Booking from response: {e}. Booking data:\n{b}")
+                continue
+
         bookings = sorted(bookings, key=lambda x: x.otf_class.starts_at)
 
         if exclude_cancelled:

src/otf_api/api/bookings/booking_client.py

@@ -51,7 +51,7 @@ def put_class(self, body: dict) -> dict:
         Raises:
             AlreadyBookedError: If the class is already booked.
             OutsideSchedulingWindowError: If the class is outside the scheduling window.
-            OtfException: If there is an error booking the class.
+            OtfError: If there is an error booking the class.
         """
         return self.client.default_request("PUT", f"/member/members/{self.member_uuid}/bookings", json=body)["data"]
 

src/otf_api/api/client.py

@@ -152,7 +152,7 @@ def _map_http_error(
 
         if re.match(r"^/member/members/.*?/bookings", path):
             if code == "NOT_AUTHORIZED" and error_msg.startswith("This class booking has been cancelled"):
-                raise exc.BookingNotFoundError("Booking was already cancelled")
+                raise exc.ResourceNotFoundError("Booking was already cancelled")
             if error_code == "603":
                 raise exc.AlreadyBookedError("Class is already booked")
             if error_code == "602":

src/otf_api/api/studios/studio_api.py

@@ -56,7 +56,15 @@ def add_favorite_studio(self, studio_uuids: list[str] | str) -> list[models.Stud
 
         new_faves = resp.get("studios", [])
 
-        return [models.StudioDetail.create(**studio, api=self.otf) for studio in new_faves]
+        studios: list[models.StudioDetail] = []
+        for studio in new_faves:
+            try:
+                studios.append(models.StudioDetail.create(**studio, api=self.otf))
+            except ValueError as e:
+                LOGGER.error(f"Failed to create StudioDetail for studio {studio}: {e}")
+                continue
+
+        return studios
 
     def remove_favorite_studio(self, studio_uuids: list[str] | str) -> None:
         """Remove a studio from the member's favorite studios.
@@ -141,7 +149,16 @@ def search_studios_by_geo(
         longitude = longitude or self.otf.home_studio.location.longitude
 
         results = self.client.get_studios_by_geo(latitude, longitude, distance)
-        return [models.StudioDetail.create(**studio, api=self.otf) for studio in results]
+
+        studios: list[models.StudioDetail] = []
+        for studio in results:
+            try:
+                studios.append(models.StudioDetail.create(**studio, api=self.otf))
+            except ValueError as e:
+                LOGGER.error(f"Failed to create StudioDetail for studio {studio}: {e}")
+                continue
+
+        return studios
 
     def _get_all_studios(self) -> list[models.StudioDetail]:
         """Gets all studios. Marked as private to avoid random users calling it.
@@ -153,7 +170,16 @@ def _get_all_studios(self) -> list[models.StudioDetail]:
         """
         # long/lat being None will cause the endpoint to return all studios
         results = self.client.get_studios_by_geo(None, None)
-        return [models.StudioDetail.create(**studio, api=self.otf) for studio in results]
+
+        studios: list[models.StudioDetail] = []
+        for studio in results:
+            try:
+                studios.append(models.StudioDetail.create(**studio, api=self.otf))
+            except ValueError as e:
+                LOGGER.error(f"Failed to create StudioDetail for studio {studio}: {e}")
+                continue
+
+        return studios
 
     def _get_studio_detail_threaded(self, studio_uuids: list[str]) -> dict[str, models.StudioDetail]:
         """Get detailed information about multiple studios in a threaded manner.
@@ -168,7 +194,13 @@ def _get_studio_detail_threaded(self, studio_uuids: list[str]) -> dict[str, mode
             dict[str, StudioDetail]: A dictionary mapping studio UUIDs to their detailed information.
         """
         studio_dicts = self.client.get_studio_detail_threaded(studio_uuids)
-        return {
-            studio_uuid: models.StudioDetail.create(**studio, api=self.otf)
-            for studio_uuid, studio in studio_dicts.items()
-        }
+
+        studios: dict[str, models.StudioDetail] = {}
+        for studio_uuid, studio in studio_dicts.items():
+            try:
+                studios[studio_uuid] = models.StudioDetail.create(**studio, api=self.otf)
+            except ValueError as e:
+                LOGGER.error(f"Failed to create StudioDetail for studio {studio_uuid}: {e}")
+                continue
+
+        return studios

src/otf_api/api/workouts/workout_api.py

@@ -221,7 +221,6 @@ def get_workout_from_booking(self, booking: str | models.BookingV2) -> models.Wo
             Workout: The member's workout.
 
         Raises:
-            BookingNotFoundError: If the booking does not exist.
             ResourceNotFoundError: If the workout does not exist.
             TypeError: If the booking is an old Booking model, as these do not have the necessary fields.
         """
@@ -271,14 +270,18 @@ def get_workouts(
 
         workouts: list[models.Workout] = []
         for perf_id, perf_summary in perf_summaries_dict.items():
-            workout = models.Workout.create(
-                **perf_summary,
-                v2_booking=bookings_dict[perf_id],
-                telemetry=telemetry_dict.get(perf_id),
-                class_uuid=perf_summary_to_class_uuid_map.get(perf_id),
-                api=self.otf,
-            )
-            workouts.append(workout)
+            try:
+                workout = models.Workout.create(
+                    **perf_summary,
+                    v2_booking=bookings_dict[perf_id],
+                    telemetry=telemetry_dict.get(perf_id),
+                    class_uuid=perf_summary_to_class_uuid_map.get(perf_id),
+                    api=self.otf,
+                )
+                workouts.append(workout)
+            except ValueError as e:
+                LOGGER.error(f"Failed to create Workout for performance summary {perf_id}: {e}")
+                continue
 
         return workouts