diff --git a/.bumpversion.toml b/.bumpversion.toml index c47ea01..c85a6e9 100644 --- a/.bumpversion.toml +++ b/.bumpversion.toml @@ -1,5 +1,5 @@ [tool.bumpversion] -current_version = "0.3.0" +current_version = "0.4.0" parse = "(?P\\d+)\\.(?P\\d+)\\.(?P\\d+)(\\.(?Pdev)\\d+\\+[-_a-zA-Z0-9]+)?" diff --git a/README.md b/README.md index 6f53d76..cb03249 100644 --- a/README.md +++ b/README.md @@ -10,7 +10,7 @@ pip install otf-api ## Overview -To use the API, you need to create an instance of the `Api` class, providing your email address and password. This will authenticate you with the API and allow you to make requests. When the `Api` object is created it automatically grabs your member details and home studio, to simplify the process of making requests. +To use the API, you need to create an instance of the `Otf` class, providing your email address and password. This will authenticate you with the API and allow you to make requests. When the `Otf` object is created it automatically grabs your member details and home studio, to simplify the process of making requests. See the [examples](./examples) for more information on how to use the API. diff --git a/docs/usage.md b/docs/usage.md index 050fe3a..c972615 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -2,7 +2,7 @@ ## Overview -To use the API, you need to create an instance of the `Api` class, providing your email address and password. This will authenticate you with the API and allow you to make requests. When the `Api` object is created it automatically grabs your member details and home studio, to simplify the process of making requests. +To use the API, you need to create an instance of the `Otf` class, providing your email address and password. This will authenticate you with the API and allow you to make requests. When the `Otf` object is created it automatically grabs your member details and home studio, to simplify the process of making requests. ### Data diff --git a/examples/challenge_tracker_examples.py b/examples/challenge_tracker_examples.py index 43b83fa..bb40a4f 100644 --- a/examples/challenge_tracker_examples.py +++ b/examples/challenge_tracker_examples.py @@ -1,7 +1,7 @@ import asyncio import os -from otf_api import Api +from otf_api import Otf from otf_api.models.responses import ChallengeType, EquipmentType USERNAME = os.getenv("OTF_EMAIL") @@ -9,7 +9,7 @@ async def main(): - otf = await Api.create(USERNAME, PASSWORD) + otf = Otf(USERNAME, PASSWORD) # challenge tracker content is an overview of the challenges OTF runs # and your participation in them diff --git a/examples/class_bookings_examples.py b/examples/class_bookings_examples.py index e2ca1f4..61257da 100644 --- a/examples/class_bookings_examples.py +++ b/examples/class_bookings_examples.py @@ -2,7 +2,7 @@ import os from collections import Counter -from otf_api import Api +from otf_api import Otf from otf_api.models.responses.bookings import BookingStatus from otf_api.models.responses.classes import ClassType @@ -11,7 +11,7 @@ async def main(): - otf = await Api.create(USERNAME, PASSWORD) + otf = otf = Otf(USERNAME, PASSWORD) resp = await otf.get_member_purchases() print(resp.model_dump_json(indent=4)) diff --git a/examples/studio_examples.py b/examples/studio_examples.py index d9fd8bd..f2194b4 100644 --- a/examples/studio_examples.py +++ b/examples/studio_examples.py @@ -1,14 +1,14 @@ import asyncio import os -from otf_api import Api +from otf_api import Otf USERNAME = os.getenv("OTF_EMAIL") PASSWORD = os.getenv("OTF_PASSWORD") async def main(): - otf = await Api.create(USERNAME, PASSWORD) + otf = otf = Otf(USERNAME, PASSWORD) # if you need to figure out what studios are in an area, you can call `search_studios_by_geo` # which takes latitude, longitude, distance, page_index, and page_size as arguments diff --git a/examples/workout_examples.py b/examples/workout_examples.py index a2e84b9..7baa630 100644 --- a/examples/workout_examples.py +++ b/examples/workout_examples.py @@ -1,14 +1,14 @@ import asyncio import os -from otf_api import Api +from otf_api import Otf USERNAME = os.getenv("OTF_EMAIL") PASSWORD = os.getenv("OTF_PASSWORD") async def main(): - otf = await Api.create(USERNAME, PASSWORD) + otf = otf = Otf(USERNAME, PASSWORD) resp = await otf.get_member_lifetime_stats() print(resp.model_dump_json(indent=4)) diff --git a/poetry.lock b/poetry.lock index 8924d9e..47ab350 100644 --- a/poetry.lock +++ b/poetry.lock @@ -3236,4 +3236,4 @@ test = ["big-O", "importlib-resources", "jaraco.functools", "jaraco.itertools", [metadata] lock-version = "2.0" python-versions = "^3.10" -content-hash = "d1a5514d076a835f5d2c3477c3383a055deab498d1a6c4766677dcfa00905e55" +content-hash = "68d3c18e5b7f31d6cf08a8a5651343f4a899b20c6a72e3319265603bec837708" diff --git a/pyproject.toml b/pyproject.toml index 08a50af..9931c20 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "otf-api" -version = "0.3.0" +version = "0.4.0" description = "Python OrangeTheory Fitness API Client" authors = ["Jessica Smith "] license = "MIT" @@ -23,36 +23,37 @@ classifiers = [ [tool.poetry.dependencies] python = "^3.10" aiohttp = "3.9.5" -loguru = "0.7.2" -pydantic = "2.7.3" -pycognito = "2024.5.1" -typer = { version = "^0.12.3", extras = ["cli"] } -pendulum = { version = "^3.0.0", extras = ["cli"] } - - -readchar = { version = "^4.1.0", extras = ["cli"] } humanize = "^4.9.0" -python-box = "^7.2.0" inflection = "0.5.*" +loguru = "0.7.2" +pendulum = "^3.0.0" pint = "0.24.*" +pycognito = "2024.5.1" +pydantic = "2.7.3" +python-box = "^7.2.0" +readchar = "^4.1.0" +typer = "^0.12.3" + [tool.poetry.group.dev.dependencies] +aioresponses = "0.7.6" +black = "^24.4.2" +build = "1.2.1" +bump-my-version = "^0.23.0" +httpx = "^0.27.0" +mypy = "1.10.0" +pre-commit = "3.7.1" pytest = "8.2.2" -pytest-loguru = "0.4.0" pytest-asyncio = "0.23.7" -aioresponses = "0.7.6" -tox = "4.15.1" pytest-cov = "5.0.0" -build = "1.2.1" +pytest-loguru = "0.4.0" ruff = "0.4.9" -pre-commit = "3.7.1" -mypy = "1.10.0" +tox = "4.15.1" twine = "5.1.0" -black = "^24.4.2" -httpx = "^0.27.0" -bump-my-version = "^0.23.0" [tool.poetry.group.docs.dependencies] +griffe-fieldz = "0.1.2" +mike = "2.1.1" mkdocs = "1.6.0" mkdocs-autorefs = "1.0.1" mkdocs-gen-files = "0.5.0" @@ -64,10 +65,8 @@ mkdocs-material-extensions = "1.3.1" mkdocs-section-index = "0.3.9" mkdocstrings = "0.25.1" mkdocstrings-python = "1.10.3" -griffe-fieldz = "0.1.2" -mike = "2.1.1" -setuptools = "^70.0.0" pkginfo = "^1.11.1" +setuptools = "^70.0.0" virtualenv = "^20.26.2" diff --git a/src/otf_api/__init__.py b/src/otf_api/__init__.py index efac0da..bbcbbb6 100644 --- a/src/otf_api/__init__.py +++ b/src/otf_api/__init__.py @@ -3,13 +3,13 @@ from loguru import logger -from .api import Api -from .models.auth import User +from .api import Otf +from .auth import OtfUser -__version__ = "0.3.0" +__version__ = "0.4.0" -__all__ = ["Api", "User"] +__all__ = ["Otf", "OtfUser"] logger.remove() logger.add(sink=sys.stdout, level=os.getenv("OTF_LOG_LEVEL", "INFO")) diff --git a/src/otf_api/api.py b/src/otf_api/api.py index 82b9400..6fef960 100644 --- a/src/otf_api/api.py +++ b/src/otf_api/api.py @@ -7,36 +7,41 @@ from typing import Any import aiohttp +import requests from loguru import logger from yarl import URL -from otf_api.models.auth import User -from otf_api.models.responses.body_composition_list import BodyCompositionList -from otf_api.models.responses.book_class import BookClass -from otf_api.models.responses.cancel_booking import CancelBooking -from otf_api.models.responses.classes import ClassType, DoW, OtfClassList -from otf_api.models.responses.favorite_studios import FavoriteStudioList -from otf_api.models.responses.lifetime_stats import StatsResponse, StatsTime -from otf_api.models.responses.performance_summary_detail import PerformanceSummaryDetail -from otf_api.models.responses.performance_summary_list import PerformanceSummaryList -from otf_api.models.responses.studio_detail import Pagination, StudioDetail, StudioDetailList -from otf_api.models.responses.telemetry import Telemetry -from otf_api.models.responses.telemetry_hr_history import TelemetryHrHistory -from otf_api.models.responses.telemetry_max_hr import TelemetryMaxHr - -from .models import ( +from otf_api.auth import OtfUser +from otf_api.models import ( + BodyCompositionList, + BookClass, BookingList, BookingStatus, + CancelBooking, ChallengeTrackerContent, ChallengeTrackerDetailList, ChallengeType, + ClassType, + DoW, EquipmentType, + FavoriteStudioList, LatestAgreement, MemberDetail, MemberMembership, MemberPurchaseList, + OtfClassList, OutOfStudioWorkoutHistoryList, + Pagination, + PerformanceSummaryDetail, + PerformanceSummaryList, + StatsResponse, + StatsTime, + StudioDetail, + StudioDetailList, StudioServiceList, + Telemetry, + TelemetryHrHistory, + TelemetryMaxHr, TotalClasses, WorkoutList, ) @@ -56,40 +61,55 @@ class AlreadyBookedError(Exception): REQUEST_HEADERS = {"Authorization": None, "Content-Type": "application/json", "Accept": "application/json"} -class Api: - """The main class of the otf-api library. Create an instance using the async method `create`. +class Otf: + logger: "Logger" = logger + user: OtfUser + _session: aiohttp.ClientSession - Example: + def __init__( + self, + username: str | None = None, + password: str | None = None, + access_token: str | None = None, + id_token: str | None = None, + refresh_token: str | None = None, + device_key: str | None = None, + user: OtfUser | None = None, + ): + """Create a new Otf instance. + + Authentication methods: --- - ```python - import asyncio - from otf_api import Api - - async def main(): - otf = await Api.create("username", "password") - print(otf.member) - - if __name__ == "__main__": - asyncio.run(main()) - ``` - """ + - Provide a username and password. + - Provide an access token and id token. + - Provide a user object. - logger: "Logger" = logger - user: User - session: aiohttp.ClientSession + Args: + username (str, optional): The username of the user. Default is None. + password (str, optional): The password of the user. Default is None. + access_token (str, optional): The access token. Default is None. + id_token (str, optional): The id token. Default is None. + refresh_token (str, optional): The refresh token. Default is None. + device_key (str, optional): The device key. Default is None. + user (OtfUser, optional): A user object. Default is None. + """ - def __init__(self, username: str, password: str): self.member: MemberDetail - self.home_studio: StudioDetail - - self.user = User.login(username, password) - - headers = { - "Authorization": f"Bearer {self.user.cognito.id_token}", - "Content-Type": "application/json", - "Accept": "application/json", - } - self.session = aiohttp.ClientSession(headers=headers) + self.home_studio_uuid: str + + if user: + self.user = user + elif username and password or (access_token and id_token): + self.user = OtfUser( + username=username, + password=password, + access_token=access_token, + id_token=id_token, + refresh_token=refresh_token, + device_key=device_key, + ) + else: + raise ValueError("No valid authentication method provided") # simplify access to member_id and member_uuid self._member_id = self.user.member_id @@ -98,26 +118,48 @@ def __init__(self, username: str, password: str): "koji-member-id": self._member_id, "koji-member-email": self.user.id_claims_data.email, } + self.member = self._get_member_details_sync() + self.home_studio_uuid = self.member.home_studio.studio_uuid - @classmethod - async def create(cls, username: str, password: str) -> "Api": - """Create a new API instance. The username and password are required arguments because even though - we cache the token, they expire so quickly that we usually end up needing to re-authenticate. + def _get_member_details_sync(self) -> MemberDetail: + """Get the member details synchronously. - Args: - username (str): The username of the user. - password (str): The password of the user. + This is used to get the member details when the API is first initialized, to let use initialize + without needing to await the member details. + + Returns: + MemberDetail: The member details. """ - self = cls(username, password) - self.member = await self.get_member_detail() - self.home_studio = await self.get_studio_detail(self.member.home_studio.studio_uuid) - return self + url = f"https://{API_BASE_URL}/member/members/{self._member_id}" + resp = requests.get(url, headers=self.headers) + return MemberDetail(**resp.json()["data"]) + + @property + def headers(self) -> dict[str, str]: + """Get the headers for the API request.""" + + # check the token before making a request in case it has expired + + self.user.cognito.check_token() + return { + "Authorization": f"Bearer {self.user.cognito.id_token}", + "Content-Type": "application/json", + "Accept": "application/json", + } + + @property + def session(self) -> aiohttp.ClientSession: + """Get the aiohttp session.""" + if not getattr(self, "_session", None): + self._session = aiohttp.ClientSession(headers=self.headers) + + return self._session def __del__(self) -> None: if not hasattr(self, "session"): return + try: - loop = asyncio.get_event_loop() asyncio.create_task(self._close_session()) # noqa except RuntimeError: loop = asyncio.new_event_loop() @@ -145,6 +187,12 @@ async def _do( logger.debug(f"Making {method!r} request to {full_url}, params: {params}") + # ensure we have headers that contain the most up-to-date token + if not headers: + headers = self.headers + else: + headers.update(self.headers) + text = None async with self.session.request(method, full_url, headers=headers, params=params, **kwargs) as response: with contextlib.suppress(Exception): @@ -155,10 +203,8 @@ async def _do( except aiohttp.ClientResponseError as e: logger.exception(f"Error making request: {e}") logger.exception(f"Response: {text}") - # raise except Exception as e: logger.exception(f"Error making request: {e}") - # raise return await response.json() @@ -242,9 +288,9 @@ class types can be provided, if there are multiple there will be a call per clas """ if not studio_uuids: - studio_uuids = [self.home_studio.studio_uuid] - elif include_home_studio and self.home_studio.studio_uuid not in studio_uuids: - studio_uuids.append(self.home_studio.studio_uuid) + studio_uuids = [self.home_studio_uuid] + elif include_home_studio and self.home_studio_uuid not in studio_uuids: + studio_uuids.append(self.home_studio_uuid) path = "/v1/classes" @@ -280,7 +326,7 @@ class types can be provided, if there are multiple there will be a call per clas classes_list.classes = [c for c in classes_list.classes if not c.canceled] for otf_class in classes_list.classes: - otf_class.is_home_studio = otf_class.studio.id == self.home_studio.studio_uuid + otf_class.is_home_studio = otf_class.studio.id == self.home_studio_uuid if day_of_week: classes_list.classes = [c for c in classes_list.classes if c.day_of_week_enum in day_of_week] @@ -415,7 +461,7 @@ async def get_bookings( for booking in data.bookings: if not booking.otf_class: continue - if booking.otf_class.studio.studio_uuid == self.home_studio.studio_uuid: + if booking.otf_class.studio.studio_uuid == self.home_studio_uuid: booking.is_home_studio = True else: booking.is_home_studio = False @@ -660,7 +706,7 @@ async def get_studio_services(self, studio_uuid: str | None = None) -> StudioSer Returns: StudioServiceList: The services available at the studio. """ - studio_uuid = studio_uuid or self.home_studio.studio_uuid + studio_uuid = studio_uuid or self.home_studio_uuid data = await self._default_request("GET", f"/member/studios/{studio_uuid}/services") return StudioServiceList(data=data["data"]) @@ -711,7 +757,7 @@ async def get_studio_detail(self, studio_uuid: str | None = None) -> StudioDetai Returns: StudioDetail: Detailed information about the studio. """ - studio_uuid = studio_uuid or self.home_studio.studio_uuid + studio_uuid = studio_uuid or self.home_studio_uuid path = f"/mobile/v1/studios/{studio_uuid}" params = {"include": "locations"} @@ -747,8 +793,11 @@ async def search_studios_by_geo( """ path = "/mobile/v1/studios" - latitude = latitude or self.home_studio.studio_location.latitude - longitude = longitude or self.home_studio.studio_location.longitude + if not latitude and not longitude: + home_studio = await self.get_studio_detail() + + latitude = home_studio.studio_location.latitude + longitude = home_studio.studio_location.longitude if page_size > 50: self.logger.warning("The API does not support more than 50 results per page, limiting to 50.") diff --git a/src/otf_api/auth.py b/src/otf_api/auth.py new file mode 100644 index 0000000..ce11d9d --- /dev/null +++ b/src/otf_api/auth.py @@ -0,0 +1,314 @@ +import typing +from typing import Any + +import jwt +import pendulum +from loguru import logger +from pycognito import AWSSRP, Cognito, MFAChallengeException +from pycognito.exceptions import TokenVerificationException +from pydantic import Field +from pydantic.config import ConfigDict + +from otf_api.models.base import OtfItemBase + +if typing.TYPE_CHECKING: + from boto3.session import Session + from botocore.config import Config + +CLIENT_ID = "65knvqta6p37efc2l3eh26pl5o" # from otlive +USER_POOL_ID = "us-east-1_dYDxUeyL1" + + +class OtfCognito(Cognito): + _device_key: str | None = None + + def __init__( + self, + user_pool_id: str, + client_id: str, + user_pool_region: str | None = None, + username: str | None = None, + id_token: str | None = None, + refresh_token: str | None = None, + access_token: str | None = None, + client_secret: str | None = None, + access_key: str | None = None, + secret_key: str | None = None, + session: "Session|None" = None, + botocore_config: "Config|None" = None, + boto3_client_kwargs: dict[str, Any] | None = None, + device_key: str | None = None, + ): + super().__init__( + user_pool_id, + client_id, + user_pool_region=user_pool_region, + username=username, + id_token=id_token, + refresh_token=refresh_token, + access_token=access_token, + client_secret=client_secret, + access_key=access_key, + secret_key=secret_key, + session=session, + botocore_config=botocore_config, + boto3_client_kwargs=boto3_client_kwargs, + ) + self.device_key = device_key + + @property + def device_key(self) -> str | None: + return self._device_key + + @device_key.setter + def device_key(self, value: str | None): + if not value: + logger.info("Clearing device key") + self._device_key = value + return + + redacted_value = value[:4] + "*" * (len(value) - 8) + value[-4:] + logger.info(f"Setting device key: {redacted_value}") + self._device_key = value + + def _set_tokens(self, tokens: dict[str, Any]): + """Set the tokens and device metadata from the response. + + Args: + tokens (dict): The response from the Cognito service. + """ + super()._set_tokens(tokens) + + if new_metadata := tokens["AuthenticationResult"].get("NewDeviceMetadata"): + self.device_key = new_metadata["DeviceKey"] + + def authenticate(self, password: str, client_metadata: dict[str, Any] | None = None, device_key: str | None = None): + """ + Authenticate the user using the SRP protocol. Overridden to add `confirm_device` call. + + Args: + password (str): The user's password + client_metadata (dict, optional): Any additional client metadata to send to Cognito + """ + aws = AWSSRP( + username=self.username, + password=password, + pool_id=self.user_pool_id, + client_id=self.client_id, + client=self.client, + client_secret=self.client_secret, + ) + try: + tokens = aws.authenticate_user(client_metadata=client_metadata) + except MFAChallengeException as mfa_challenge: + self.mfa_tokens = mfa_challenge.get_tokens() + raise mfa_challenge + + # Set the tokens and device metadata + self._set_tokens(tokens) + + if not device_key: + # Confirm the device so we can use the refresh token + aws.confirm_device(tokens) + else: + self.device_key = device_key + try: + self.renew_access_token() + except TokenVerificationException: + logger.error("Failed to renew access token. Confirming device.") + self.device_key = None + aws.confirm_device(tokens) + + def check_token(self, renew: bool = True) -> bool: + """ + Checks the exp attribute of the access_token and either refreshes + the tokens by calling the renew_access_tokens method or does nothing + :param renew: bool indicating whether to refresh on expiration + :return: bool indicating whether access_token has expired + """ + if not self.access_token: + raise AttributeError("Access Token Required to Check Token") + now = pendulum.now() + dec_access_token = jwt.decode(self.access_token, options={"verify_signature": False}) + + exp = pendulum.DateTime.fromtimestamp(dec_access_token["exp"]) + if now > exp.subtract(minutes=15): + expired = True + if renew: + self.renew_access_token() + else: + expired = False + return expired + + def renew_access_token(self): + """Sets a new access token on the User using the cached refresh token and device metadata.""" + auth_params = {"REFRESH_TOKEN": self.refresh_token} + self._add_secret_hash(auth_params, "SECRET_HASH") + + if self.device_key: + logger.info("Using device key for refresh token") + auth_params["DEVICE_KEY"] = self.device_key + + refresh_response = self.client.initiate_auth( + ClientId=self.client_id, AuthFlow="REFRESH_TOKEN_AUTH", AuthParameters=auth_params + ) + self._set_tokens(refresh_response) + + @classmethod + def from_token( + cls, access_token: str, id_token: str, refresh_token: str | None = None, device_key: str | None = None + ) -> "OtfCognito": + """Create an OtfCognito instance from an id token. + + Args: + access_token (str): The access token. + id_token (str): The id token. + refresh_token (str, optional): The refresh token. Defaults to None. + device_key (str, optional): The device key. Defaults + + Returns: + OtfCognito: The user instance + """ + cognito = OtfCognito( + USER_POOL_ID, + CLIENT_ID, + access_token=access_token, + id_token=id_token, + refresh_token=refresh_token, + device_key=device_key, + ) + cognito.verify_tokens() + cognito.check_token() + return cognito + + @classmethod + def login(cls, username: str, password: str) -> "OtfCognito": + """Create an OtfCognito instance from a username and password. + + Args: + username (str): The username to login with. + password (str): The password to login with. + + Returns: + OtfCognito: The logged in user. + """ + cognito_user = OtfCognito(USER_POOL_ID, CLIENT_ID, username=username) + cognito_user.authenticate(password) + cognito_user.check_token() + return cognito_user + + +class IdClaimsData(OtfItemBase): + sub: str + email_verified: bool + iss: str + cognito_username: str = Field(alias="cognito:username") + given_name: str + locale: str + home_studio_id: str = Field(alias="custom:home_studio_id") + aud: str + event_id: str + token_use: str + auth_time: int + exp: int + is_migration: str = Field(alias="custom:isMigration") + iat: int + family_name: str + email: str + koji_person_id: str = Field(alias="custom:koji_person_id") + + @property + def member_uuid(self) -> str: + return self.cognito_username + + @property + def full_name(self) -> str: + return f"{self.given_name} {self.family_name}" + + +class AccessClaimsData(OtfItemBase): + sub: str + device_key: str + iss: str + client_id: str + event_id: str + token_use: str + scope: str + auth_time: int + exp: int + iat: int + jti: str + username: str + + @property + def member_uuid(self) -> str: + return self.username + + +class OtfUser(OtfItemBase): + model_config = ConfigDict(arbitrary_types_allowed=True) + cognito: OtfCognito + + def __init__( + self, + username: str | None = None, + password: str | None = None, + id_token: str | None = None, + access_token: str | None = None, + refresh_token: str | None = None, + device_key: str | None = None, + cognito: OtfCognito | None = None, + ): + """Create a User instance. + + Args: + username (str, optional): The username to login with. Defaults to None. + password (str, optional): The password to login with. Defaults to None. + id_token (str, optional): The id token. Defaults to None. + access_token (str, optional): The access token. Defaults to None. + refresh_token (str, optional): The refresh token. Defaults to None. + device_key (str, optional): The device key. Defaults to None. + cognito (OtfCognito, optional): A Cognito instance. Defaults to None. + + Raises: + ValueError: Must provide either username and password or id token + + + """ + if cognito: + cognito = cognito + elif username and password: + cognito = OtfCognito.login(username, password) + elif access_token and id_token: + cognito = OtfCognito.from_token(access_token, id_token, refresh_token, device_key) + else: + raise ValueError("Must provide either username and password or id token.") + + super().__init__(cognito=cognito) + + @property + def member_id(self) -> str: + return self.id_claims_data.cognito_username + + @property + def member_uuid(self) -> str: + return self.access_claims_data.sub + + @property + def access_claims_data(self) -> AccessClaimsData: + return AccessClaimsData(**self.cognito.access_claims) + + @property + def id_claims_data(self) -> IdClaimsData: + return IdClaimsData(**self.cognito.id_claims) + + def get_tokens(self) -> dict[str, str]: + return { + "id_token": self.cognito.id_token, + "access_token": self.cognito.access_token, + "refresh_token": self.cognito.refresh_token, + } + + @property + def device_key(self) -> str: + return self.cognito.device_key diff --git a/src/otf_api/cli/app.py b/src/otf_api/cli/app.py index ea405bb..0830a27 100644 --- a/src/otf_api/cli/app.py +++ b/src/otf_api/cli/app.py @@ -12,10 +12,9 @@ import otf_api from otf_api.cli._utilities import is_async_fn, with_cli_exception_handling -from otf_api.models.auth import User if typing.TYPE_CHECKING: - from otf_api.api import Api + from otf_api import Otf class OutputType(str, Enum): @@ -79,7 +78,7 @@ def __init__(self, *args: typing.Any, **kwargs: typing.Any): self.console = Console(highlight=False, theme=theme, color_system="auto") # TODO: clean these up later, just don't want warnings everywhere that these could be None - self.api: Api = None # type: ignore + self.api: Otf = None # type: ignore self.username: str = None # type: ignore self.password: str = None # type: ignore self.output: OutputType = None # type: ignore @@ -91,10 +90,6 @@ def set_username(self, username: str | None = None) -> None: self.username = username return - if User.cache_file_exists(): - self.username = User.username_from_disk() - return - raise ValueError("Username not provided and not found in cache") def set_log_level(self, level: str) -> None: diff --git a/src/otf_api/cli/bookings.py b/src/otf_api/cli/bookings.py index 01be9af..206d9ec 100644 --- a/src/otf_api/cli/bookings.py +++ b/src/otf_api/cli/bookings.py @@ -56,7 +56,7 @@ async def list_bookings( bk_status = BookingStatus.get_from_key_insensitive(status.value) if status else None if not base_app.api: - base_app.api = await otf_api.Api.create(base_app.username, base_app.password) + base_app.api = await otf_api.Otf.create(base_app.username, base_app.password) bookings = await base_app.api.get_bookings(start_date, end_date, bk_status, limit, exclude_cancelled) if base_app.output == "json": @@ -82,7 +82,7 @@ async def book(class_uuid: str = typer.Option(help="Class UUID to cancel")) -> N logger.info(f"Booking class {class_uuid}") if not base_app.api: - base_app.api = await otf_api.Api.create(base_app.username, base_app.password) + base_app.api = await otf_api.Otf.create(base_app.username, base_app.password) booking = await base_app.api.book_class(class_uuid) base_app.console.print(booking) @@ -115,7 +115,7 @@ async def book_interactive( class_type_enums = None if not base_app.api: - base_app.api = await otf_api.Api.create(base_app.username, base_app.password) + base_app.api = await otf_api.Otf.create(base_app.username, base_app.password) classes = await base_app.api.get_classes( studio_uuids, @@ -152,7 +152,7 @@ async def cancel_interactive() -> None: with base_app.console.status("Getting bookings...", spinner="arc"): if not base_app.api: - base_app.api = await otf_api.Api.create(base_app.username, base_app.password) + base_app.api = await otf_api.Otf.create(base_app.username, base_app.password) bookings = await base_app.api.get_bookings() result = prompt_select_from_table( @@ -177,7 +177,7 @@ async def cancel(booking_uuid: str = typer.Option(help="Booking UUID to cancel") logger.info(f"Cancelling booking {booking_uuid}") if not base_app.api: - base_app.api = await otf_api.Api.create(base_app.username, base_app.password) + base_app.api = await otf_api.Otf.create(base_app.username, base_app.password) booking = await base_app.api.cancel_booking(booking_uuid) base_app.console.print(booking) @@ -211,7 +211,7 @@ async def list_classes( class_type_enum = ClassType.get_from_key_insensitive(class_type.value) if class_type else None if not base_app.api: - base_app.api = await otf_api.Api.create(base_app.username, base_app.password) + base_app.api = await otf_api.Otf.create(base_app.username, base_app.password) classes = await base_app.api.get_classes( studio_uuids, include_home_studio, start_date, end_date, limit, class_type_enum, exclude_cancelled ) diff --git a/src/otf_api/models/__init__.py b/src/otf_api/models/__init__.py index 3d5a4d6..e3387d7 100644 --- a/src/otf_api/models/__init__.py +++ b/src/otf_api/models/__init__.py @@ -1,5 +1,5 @@ -from .auth import User from .responses import ( + BodyCompositionList, BookClass, BookingList, BookingStatus, @@ -7,6 +7,8 @@ ChallengeTrackerContent, ChallengeTrackerDetailList, ChallengeType, + ClassType, + DoW, EquipmentType, FavoriteStudioList, HistoryClassStatus, @@ -16,8 +18,11 @@ MemberPurchaseList, OtfClassList, OutOfStudioWorkoutHistoryList, + Pagination, PerformanceSummaryDetail, PerformanceSummaryList, + StatsResponse, + StatsTime, StudioDetail, StudioDetailList, StudioServiceList, @@ -29,31 +34,37 @@ ) __all__ = [ - "User", - "ChallengeType", - "BookingStatus", - "EquipmentType", - "HistoryClassStatus", + "BodyCompositionList", + "BookClass", "BookingList", + "BookingStatus", + "CancelBooking", "ChallengeTrackerContent", "ChallengeTrackerDetailList", + "ChallengeType", + "ClassType", + "DoW", + "EquipmentType", + "FavoriteStudioList", + "HistoryClassStatus", "LatestAgreement", "MemberDetail", "MemberMembership", "MemberPurchaseList", + "OtfClassList", "OutOfStudioWorkoutHistoryList", + "Pagination", + "PerformanceSummaryDetail", + "PerformanceSummaryList", + "StatsResponse", + "StatsTime", + "StudioDetail", + "StudioDetailList", "StudioServiceList", - "TotalClasses", - "WorkoutList", - "FavoriteStudioList", - "OtfClassList", - "TelemetryHrHistory", + "StudioStatus", "Telemetry", + "TelemetryHrHistory", "TelemetryMaxHr", - "StudioDetail", - "StudioDetailList", - "PerformanceSummaryDetail", - "PerformanceSummaryList", - "BookClass", - "CancelBooking", + "TotalClasses", + "WorkoutList", ] diff --git a/src/otf_api/models/auth.py b/src/otf_api/models/auth.py deleted file mode 100644 index f77ff15..0000000 --- a/src/otf_api/models/auth.py +++ /dev/null @@ -1,147 +0,0 @@ -import json -from pathlib import Path -from typing import ClassVar - -from pycognito import Cognito, TokenVerificationException -from pydantic import Field - -from otf_api.models.base import OtfItemBase - -CLIENT_ID = "65knvqta6p37efc2l3eh26pl5o" # from otlive -USER_POOL_ID = "us-east-1_dYDxUeyL1" - - -class IdClaimsData(OtfItemBase): - sub: str - email_verified: bool - iss: str - cognito_username: str = Field(alias="cognito:username") - given_name: str - locale: str - home_studio_id: str = Field(alias="custom:home_studio_id") - aud: str - event_id: str - token_use: str - auth_time: int - exp: int - is_migration: str = Field(alias="custom:isMigration") - iat: int - family_name: str - email: str - koji_person_id: str = Field(alias="custom:koji_person_id") - - @property - def member_uuid(self) -> str: - return self.cognito_username - - @property - def full_name(self) -> str: - return f"{self.given_name} {self.family_name}" - - -class AccessClaimsData(OtfItemBase): - sub: str - device_key: str - iss: str - client_id: str - event_id: str - token_use: str - scope: str - auth_time: int - exp: int - iat: int - jti: str - username: str - - @property - def member_uuid(self) -> str: - return self.username - - -class User: - token_path: ClassVar[Path] = Path("~/.otf/.tokens").expanduser() - cognito: Cognito - - def __init__(self, cognito: Cognito): - self.cognito = cognito - - @property - def member_id(self) -> str: - return self.id_claims_data.cognito_username - - @property - def member_uuid(self) -> str: - return self.access_claims_data.sub - - @property - def access_claims_data(self) -> AccessClaimsData: - return AccessClaimsData(**self.cognito.access_claims) - - @property - def id_claims_data(self) -> IdClaimsData: - return IdClaimsData(**self.cognito.id_claims) - - def save_to_disk(self) -> None: - self.token_path.parent.mkdir(parents=True, exist_ok=True) - data = { - "username": self.cognito.username, - "id_token": self.cognito.id_token, - "access_token": self.cognito.access_token, - "refresh_token": self.cognito.refresh_token, - } - self.token_path.write_text(json.dumps(data)) - - @classmethod - def cache_file_exists(cls) -> bool: - return cls.token_path.exists() - - @classmethod - def username_from_disk(cls) -> str: - val: str = json.loads(cls.token_path.read_text())["username"] - return val - - @classmethod - def load_from_disk(cls, username: str, password: str) -> "User": - """Load a User instance from disk. If the token is invalid, reauthenticate with the provided credentials. - - Args: - username (str): The username to reauthenticate with. - password (str): The password to reauthenticate with. - - Returns: - User: The loaded user. - - """ - attr_dict = json.loads(cls.token_path.read_text()) - - cognito_user = Cognito(USER_POOL_ID, CLIENT_ID, **attr_dict) - try: - cognito_user.verify_tokens() - return cls(cognito=cognito_user) - except TokenVerificationException: - user = cls.login(username, password) - return user - - @classmethod - def login(cls, username: str, password: str) -> "User": - """Login and return a User instance. After a successful login, the user is saved to disk. - - Args: - username (str): The username to login with. - password (str): The password to login with. - - Returns: - User: The logged in user. - """ - cognito_user = Cognito(USER_POOL_ID, CLIENT_ID, username=username) - cognito_user.authenticate(password) - cognito_user.check_token() - user = cls(cognito=cognito_user) - user.save_to_disk() - return user - - def refresh_token(self) -> "User": - """Refresh the user's access token.""" - self.cognito.check_token() - self.save_to_disk() - return self diff --git a/src/otf_api/models/responses/__init__.py b/src/otf_api/models/responses/__init__.py index 13265e1..f5309d1 100644 --- a/src/otf_api/models/responses/__init__.py +++ b/src/otf_api/models/responses/__init__.py @@ -1,19 +1,21 @@ +from .body_composition_list import BodyCompositionList from .book_class import BookClass from .bookings import BookingList, BookingStatus from .cancel_booking import CancelBooking from .challenge_tracker_content import ChallengeTrackerContent from .challenge_tracker_detail import ChallengeTrackerDetailList -from .classes import OtfClassList +from .classes import ClassType, DoW, OtfClassList from .enums import ChallengeType, EquipmentType, HistoryClassStatus from .favorite_studios import FavoriteStudioList from .latest_agreement import LatestAgreement +from .lifetime_stats import StatsResponse, StatsTime from .member_detail import MemberDetail from .member_membership import MemberMembership from .member_purchases import MemberPurchaseList from .out_of_studio_workout_history import OutOfStudioWorkoutHistoryList from .performance_summary_detail import PerformanceSummaryDetail from .performance_summary_list import PerformanceSummaryList -from .studio_detail import StudioDetail, StudioDetailList +from .studio_detail import Pagination, StudioDetail, StudioDetailList from .studio_services import StudioServiceList from .telemetry import Telemetry from .telemetry_hr_history import TelemetryHrHistory @@ -22,31 +24,37 @@ from .workouts import WorkoutList __all__ = [ + "BodyCompositionList", + "BookClass", "BookingList", + "BookingStatus", + "CancelBooking", "ChallengeTrackerContent", "ChallengeTrackerDetailList", + "ChallengeType", + "ClassType", + "DoW", + "EquipmentType", + "FavoriteStudioList", + "HistoryClassStatus", "LatestAgreement", "MemberDetail", "MemberMembership", "MemberPurchaseList", + "OtfClassList", "OutOfStudioWorkoutHistoryList", + "Pagination", + "PerformanceSummaryDetail", + "PerformanceSummaryList", + "StatsResponse", + "StatsTime", + "StudioDetail", + "StudioDetailList", "StudioServiceList", - "TotalClasses", - "WorkoutList", - "ChallengeType", - "BookingStatus", - "EquipmentType", - "HistoryClassStatus", "StudioStatus", - "FavoriteStudioList", - "OtfClassList", - "TelemetryHrHistory", "Telemetry", + "TelemetryHrHistory", "TelemetryMaxHr", - "StudioDetail", - "StudioDetailList", - "PerformanceSummaryDetail", - "PerformanceSummaryList", - "BookClass", - "CancelBooking", + "TotalClasses", + "WorkoutList", ] diff --git a/tests/test_api.py b/tests/test_api.py index fdee331..4f9c1f6 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -1,8 +1,8 @@ import pytest -from otf_api.api import Api +from otf_api.api import Otf @pytest.mark.asyncio async def test_api_raises_error_if_no_username_password(): with pytest.raises(TypeError): - await Api.create() + await Otf.create()