Strava fetch methods return model objects rather than Any

pyproject.toml

@@ -36,6 +36,7 @@ warn_return_any = true
 strict_optional = true
 disallow_incomplete_defs = true
 exclude = ["tests"]
+plugins = ['pydantic.mypy']
 
 [tool.ruff]
 line-length = 88

src/pardner/services/strava.py

@@ -1,9 +1,13 @@
-from typing import Any, Iterable, Optional, override
+from collections import defaultdict
+from datetime import datetime, timedelta
+from typing import Any, Iterable, Literal, Optional, override
+from urllib.parse import urljoin
 
 from pardner.exceptions import UnsupportedRequestException, UnsupportedVerticalException
 from pardner.services import BaseTransferService
 from pardner.services.utils import scope_as_set, scope_as_string
-from pardner.verticals import PhysicalActivityVertical, Vertical
+from pardner.verticals import PhysicalActivityVertical, SocialPostingVertical, Vertical
+from pardner.verticals.sub_verticals import AssociatedMediaSubVertical
 
 
 class StravaTransferService(BaseTransferService):
@@ -64,9 +68,150 @@ def scope_for_verticals(self, verticals: Iterable[Vertical]) -> set[str]:
                 sub_scopes.update(['activity:read', 'profile:read_all'])
         return sub_scopes
 
+    def _convert_to_datetime(self, raw_datetime: str | None) -> datetime | None:
+        if raw_datetime:
+            return datetime.strptime(raw_datetime, '%Y-%m-%dT%H:%M:%SZ')
+        return None
+
+    def _parse_social_posting(self, raw_data: Any) -> SocialPostingVertical | None:
+        """
+        Given the response from the API request, creates a
+        :class:`SocialPostingVertical` model object, if possible.
+
+        :param raw_data: the JSON representation of the data returned by the request.
+
+        :returns: :class:`SocialPostingVertical` or ``None``, depending on whether it
+        was possible to extract data from the response
+        """
+        if not isinstance(raw_data, dict):
+            return None
+        raw_data_dict = defaultdict(dict, raw_data)
+
+        created_at = raw_data_dict.get('start_date')
+        if created_at:
+            created_at = self._convert_to_datetime(created_at)
+
+        url_str = urljoin(
+            'https://www.strava.com/activities/', str(raw_data_dict.get('id'))
+        )
+        interaction_count = raw_data_dict.get('kudos_count', 0) + raw_data_dict.get(
+            'comment_count', 0
+        )
+
+        status: Literal['public', 'private', 'restricted'] = 'public'
+        if raw_data_dict.get('private'):
+            status = 'private'
+        elif raw_data_dict.get('visibility') == 'followers_only':
+            status = 'restricted'
+
+        associated_media_list = []
+        if raw_data_dict.get('total_photo_count', 0) > 0:
+            photo_urls = (
+                raw_data_dict['photos'].get('primary', {}).get('urls', {}).values()
+            )
+            associated_media_list = [
+                AssociatedMediaSubVertical(image_url=photo_url)
+                for photo_url in photo_urls
+            ]
+
+        return SocialPostingVertical(
+            creator_user_id=str(raw_data_dict['athlete'].get('id')),
+            service=self._service_name,
+            created_at=created_at,
+            url=url_str,
+            associated_media=associated_media_list,
+            interaction_count=interaction_count,
+            status=status,
+            text=raw_data_dict.get('description'),
+            title=raw_data_dict.get('name'),
+        )
+
+    def fetch_social_posting_vertical(
+        self, request_params: dict[str, Any] = {}, count: int = 30
+    ) -> tuple[list[SocialPostingVertical | None], Any]:
+        """
+        Fetches and returns social postings created by the authorized user.
+
+        :param count: number of posts to request. At most 30 at a time.
+        :param request_params: any other endpoint-specific parameters to be sent
+        to the endpoint. Depending on the parameters passed, this could override
+        the other arguments to this method.
+
+        :returns: two elements: the first, a list of :class:`SocialPostingVertical`s
+        or ``None``, if unable to parse; the second, the raw response from making the
+        request.
+
+        :raises: :class:`UnsupportedRequestException` if the request is unable to be
+        made.
+        """
+        max_count = 30
+        if count <= max_count:
+            raw_social_postings = self._get_resource_from_path(
+                'athlete/activities', params={'per_page': count, **request_params}
+            ).json()
+            return [
+                self._parse_social_posting(raw_social_posting)
+                for raw_social_posting in raw_social_postings
+            ], raw_social_postings
+        raise UnsupportedRequestException(
+            self._service_name,
+            f'can only make a request for at most {max_count} posts at a time.',
+        )
+
+    def _parse_physical_activity(
+        self, raw_data: Any
+    ) -> PhysicalActivityVertical | None:
+        """
+        Given the response from the API request, creates a
+        :class:`PhysicalActivityVertical` model object, if possible.
+
+        :param raw_data: the JSON representation of the data returned by the request.
+
+        :returns: :class:`PhysicalActivityVertical` or ``None``, depending on whether it
+        was possible to extract data from the response
+        """
+        social_posting = self._parse_social_posting(raw_data)
+        if not social_posting:
+            return None
+
+        social_posting_dict = social_posting.model_dump()
+        raw_data_dict = defaultdict(dict, raw_data)
+
+        start_datetime = self._convert_to_datetime(raw_data_dict.get('start_date'))
+        duration_s = raw_data_dict.get('elapsed_time')
+        duration_timedelta = timedelta(seconds=duration_s) if duration_s else None
+        end_datetime = (
+            start_datetime + duration_timedelta
+            if start_datetime and duration_timedelta
+            else None
+        )
+
+        start_latlng = raw_data_dict.get('start_latlng')
+        end_latlng = raw_data_dict.get('end_latlng')
+
+        social_posting_dict.update(
+            {
+                'vertical_name': 'physical_activity',
+                'activity_type': raw_data_dict.get('sport_type'),
+                'distance': raw_data_dict.get('distance'),
+                'elevation_high': raw_data_dict.get('elev_high'),
+                'elevation_low': raw_data_dict.get('elev_low'),
+                'kilocalories': raw_data_dict.get('calories'),
+                'max_speed': raw_data_dict.get('max_speed'),
+                'start_datetime': start_datetime,
+                'end_datetime': end_datetime,
+                'start_latitude': start_latlng[0] if start_latlng else None,
+                'start_longitude': start_latlng[1] if start_latlng else None,
+                'end_latitude': end_latlng[0] if end_latlng else None,
+                'end_longitude': end_latlng[1] if end_latlng else None,
+            }
+        )
+
+        return PhysicalActivityVertical.model_validate(social_posting_dict)
+
     def fetch_physical_activity_vertical(
         self, request_params: dict[str, Any] = {}, count: int = 30
-    ) -> list[Any]:
+    ) -> tuple[list[PhysicalActivityVertical | None], Any]:
         """
         Fetches and returns activities completed by the authorized user.
 
@@ -75,19 +220,22 @@ def fetch_physical_activity_vertical(
         to the endpoint. Depending on the parameters passed, this could override
         the other arguments to this method.
 
-        :returns: a list of dictionary objects with information for the activities from
-        the authorized user.
+        :returns: two elements: the first, a list of :class:`PhysicalActivityVertical`s
+        or ``None``, if unable to parse; the second, the raw response from making the
+        request.
 
         :raises: :class:`UnsupportedRequestException` if the request is unable to be
         made.
         """
         max_count = 30
         if count <= max_count:
-            return list(
-                self._get_resource_from_path(
-                    'athlete/activities', params={'per_page': count, **request_params}
-                ).json()
-            )
+            raw_activities = self._get_resource_from_path(
+                'athlete/activities', params={'per_page': count, **request_params}
+            ).json()
+            return [
+                self._parse_physical_activity(raw_activity)
+                for raw_activity in raw_activities
+            ], raw_activities
         raise UnsupportedRequestException(
             self._service_name,
             f'can only make a request for at most {max_count} activities at a time.',

src/pardner/verticals/base.py

@@ -1,3 +1,4 @@
+import uuid
 from abc import ABC
 from datetime import datetime
 from typing import Type
@@ -12,7 +13,7 @@ class BaseVertical(BaseModel, ABC):
     supported by every transfer service.
     """
 
-    id: str
+    id: str = Field(default_factory=lambda: uuid.uuid4().hex)
     creator_user_id: str
     service: str = Field(
         description='The name of the service the data was pulled from.'