Add trackable support (#1)

* Add trackable support
* Added comments
* Logger not constructed..

* CR:
- Adding some newlines
- Removed double import

Author: Sholofly

geocachingapi/__init__.py

@@ -1,2 +1,3 @@
 """Python client for connecting to the Geocaching API"""
-from .geocachingapi import GeocachingApi, GeocachingStatus
+from .geocachingapi import GeocachingApi
+from .models import GeocachingSettings, GeocachingStatus

geocachingapi/__version__.py

@@ -1,2 +1,2 @@
 """Python client for accessing the Geocaching API"""
-__version__ = "0.0.1"
+__version__ = "0.0.7"

geocachingapi/const.py

@@ -3,4 +3,11 @@
 GEOCACHING_API_HOST = "staging.api.groundspeak.com"
 GEOCACHING_API_PORT = 443
 GEOCACHING_API_BASE_PATH = "/"
-GEOCACHING_API_VERSION = "v1"
+GEOCACHING_API_VERSION = "v1"
+
+MEMBERSHIP_LEVELS = {
+    0: "Unknown",
+    1: "Basic",
+    2: "Charter",
+    3: "Premium"
+}

geocachingapi/exceptions.py

@@ -13,4 +13,5 @@ class GeocachingApiConnectionTimeoutError(GeocachingApiConnectionError):
 
 
 class GeocachingApiRateLimitError(GeocachingApiConnectionError):
-    """GeocachingApi Rate Limit exception."""
+    """GeocachingApi Rate Limit exception."""
+    

geocachingapi/geocachingapi.py

@@ -3,9 +3,11 @@
 
 import asyncio
 import json
+import logging
 import socket
 import async_timeout
 import backoff
+
 from yarl import URL
 from aiohttp import ClientResponse, ClientSession, ClientError
 
@@ -26,21 +28,29 @@
 
 from .models import (
     GeocachingStatus,
+    GeocachingSettings
 )
 
-class GeocachingApi:
+_LOGGER = logging.getLogger(__name__)
 
+class GeocachingApi:
+    """ Main class to control the Geocaching API"""
     _close_session: bool = False
-    _status: GeocachingStatus = GeocachingStatus()
+    _status: GeocachingStatus = None
+    _settings: GeocachingSettings = None
     def __init__(
         self,
         *,
         token: str,
+        settings: GeocachingSettings = None,
         request_timeout: int = 8,
         session: Optional[ClientSession] = None,
-        token_refresh_method: Optional[Callable[[], Awaitable[str]]] = None,
+        token_refresh_method: Optional[Callable[[], Awaitable[str]]] = None
+       
     ) -> None:
         """Initialize connection with the Geocaching API."""
+        self._status = GeocachingStatus()
+        self._settings = settings or GeocachingSettings(False)
         self._session = session
         self.request_timeout = request_timeout
         self.token = token
@@ -54,13 +64,14 @@ async def _request(self, method, uri, **kwargs) -> ClientResponse:
         """Make a request."""
         if self.token_refresh_method is not None:
             self.token = await self.token_refresh_method()
-        
+        _LOGGER.debug(f'Received API request with token {self.token}')
         url = URL.build(
             scheme=GEOCACHING_API_SCHEME,
             host=GEOCACHING_API_HOST,
             port=GEOCACHING_API_PORT,
             path=GEOCACHING_API_BASE_PATH,
         ).join(URL(uri))
+        _LOGGER.debug(f'URL: {url}')
         headers = kwargs.get("headers")
 
         if headers is None:
@@ -109,13 +120,19 @@ async def _request(self, method, uri, **kwargs) -> ClientResponse:
         # Handle empty response
         if response.status == 204:
             return
-
+        
         if "application/json" in content_type:
-            return await response.json()
-        return await response.text()
+            result =  await response.json()
+            _LOGGER.debug(f'response: {str(result)}')
+            return result
+        result =  await response.text()
+        _LOGGER.debug(f'response: {str(result)}')
+        return result
 
     async def update(self) -> GeocachingStatus:
         await self._update_user(None)
+        if self._settings.fetch_trackables:
+            await self._update_trackables()
         return self._status
 
     async def _update_user(self, data: Dict[str, Any] = None) -> None:
@@ -128,11 +145,27 @@ async def _update_user(self, data: Dict[str, Any] = None) -> None:
                 "hideCount",
                 "favoritePoints",
                 "souvenirCount",
-                "awardedFavoritePoints"
+                "awardedFavoritePoints",
+                "membershipLevelId"
             ])
             data = await self._request("GET", f"/{GEOCACHING_API_VERSION}/users/me?fields={fields}")
-        self._status.user.update_from_dict(data)
+        self._status.update_user_from_dict(data)
 
+    async def _update_trackables(self, data: Dict[str, Any] = None) -> None:
+        assert self._status
+        if data is None:
+            fields = ",".join([
+                "referenceCode",
+                "name",
+                "holder",
+                "trackingNumber",
+                "kilometersTraveled",
+                "currentGeocacheCode",
+                "currentGeocacheName"
+            ])
+            data = await self._request("GET", f"/{GEOCACHING_API_VERSION}/trackables?fields={fields}&type=3")
+        self._status.update_trackables_from_dict(data)
+
     async def close(self) -> None:
         """Close open client session."""
         if self._session and self._close_session:
@@ -144,4 +177,5 @@ async def __aenter__(self) -> GeocachingApi:
 
     async def __aexit__(self, *exc_info) -> None:
         """Async exit."""
-        await self.close()
+        await self.close()
+        

geocachingapi/models.py

@@ -1,10 +1,24 @@
 from __future__ import annotations
-from typing import Any, Callable, Dict, Optional
+from typing import Any, Dict, Optional
 
 from dataclasses import dataclass
+from datetime import datetime
+from .utils import try_get_from_dict
+
+
+class GeocachingSettings:
+    """Class to hold the Geocaching Api settings"""
+    fetch_trackables: bool = False
+    def __init__(self, fetch_trackables:bool = False) -> None:
+        """Initialize settings"""
+        self.fetch_trackables = fetch_trackables
+
+
+
 
 @dataclass
 class GeocachingUser:
+    """Class to hold the Geocaching user information"""
     reference_code: Optional[str] = None
     username: Optional[str] = None
     find_count: Optional[int] = None
@@ -14,27 +28,63 @@ class GeocachingUser:
     awarded_favorite_points: Optional[int] = None
 
     def update_from_dict(self, data: Dict[str, Any]) -> None:
-        if  "username" in data:
-            self.username = data["username"]
-        if  "referenceCode" in data:
-            self.reference_code = data["referenceCode"]
-        if  "findCount" in data:
-            self.find_count = data["findCount"]
-        if  "hideCount" in data:
-            self.hide_count = data["hideCount"]
-        if  "favoritePoints" in data:
-            self.favorite_points = data["favoritePoints"]
-        if  "souvenirCount" in data:
-            self.souvenir_count = data["souvenirCount"]
-        if  "awardedFavoritePoints" in data:
-            self.awarded_favorite_points = data["awardedFavoritePoints"]
-
-        pass
+        """Update user from the API result"""
+        self.reference_code = try_get_from_dict(data, "referenceCode", self.reference_code)
+        self.username = try_get_from_dict(data, "username", self.username)
+        self.find_count = try_get_from_dict(data, "findCount", self.find_count)
+        self.hide_count = try_get_from_dict(data, "hideCount", self.hide_count)
+        self.favorite_points = try_get_from_dict(data, "favoritePoints", self.favorite_points)
+        self.souvenir_count = try_get_from_dict(data, "souvenirCount", self.souvenir_count)
+        self.awarded_favorite_points = try_get_from_dict(data, "awardedFavoritePoints", self.awarded_favorite_points)
+
+@dataclass
+class GeocachingTrackable:
+    """Class to hold the Geocaching trackable information"""
+    reference_code: Optional[str] = None
+    name: Optional[str] = None
+    holder: GeocachingUser = None
+    tracking_number: Optional[str] = None
+    kilometers_traveled: Optional[datetime] = None
+    current_geocache_code: Optional[str] = None
+    current_geocache_name: Optional[str] = None
+
+    def update_from_dict(self, data: Dict[str, Any]) -> None:
+        """Update trackble from the API"""
+        self.reference_code = try_get_from_dict(data, "referenceCode", self.reference_code)
+        self.name = try_get_from_dict(data, "name", self.name)
+        if data["holder"] is not None:
+            if self.holder is None :
+                holder = GeocachingUser()
+            holder.update_from_dict(data["holder"])
+        else:
+            holder = None
+
+        self.tracking_number = try_get_from_dict(data, "trackingNumber", self.tracking_number)
+        self.kilometers_traveled = try_get_from_dict(data, "kilometersTraveled", self.kilometers_traveled)
+        self.current_geocache_code = try_get_from_dict(data, "currectGeocacheCode", self.current_geocache_code)
+        self.current_geocache_name = try_get_from_dict(data, "currentGeocacheName", self.current_geocache_name)
 
 class GeocachingStatus:
-    user: GeocachingUser = GeocachingUser()
-    def __init__(self) -> None:
-        pass
+    """Class to hold all account status information"""
+    user: GeocachingUser = None
+    trackables: Dict[str, GeocachingTrackable] = None
+
+    def __init__(self):
+        """Initialize GeocachingStatus"""
+        self.user = GeocachingUser()
+        self.trackables = {}
 
-    def update_from_dict(self, data: Dict[str, Any]) -> GeocachingStatus:
-        return self
+    def update_user_from_dict(self, data: Dict[str, Any]) -> None:
+        """Update user from the API result"""
+        self.user.update_from_dict(data)
+    
+    def update_trackables_from_dict(self, data: Any) -> None:
+        """Update trackables from the API result"""
+        if not any(data):
+            pass
+        for trackable in data:
+            reference_code = trackable["referenceCode"]
+            if not reference_code in self.trackables.keys():
+                self.trackables[reference_code] = GeocachingTrackable()
+            self.trackables[reference_code].update_from_dict(trackable)
+            

geocachingapi/utils.py

@@ -0,0 +1,14 @@
+"""Utils for consuming the API"""
+from typing import Dict, Any, Callable, Optional
+
+def try_get_from_dict(data: Dict[str, Any], key: str, original_value: Any, conversion: Optional[Callable[[Any], Any]] = None) -> Any:
+    """Try to get value from dict, otherwise set default value"""
+    if not key in data:
+        return None
+    
+    value = data[key]
+    if value is None:
+        return original_value
+    if conversion is None:
+        return value 
+    return conversion(value)

setup.py

@@ -44,7 +44,7 @@ def read(*parts):
         "Programming Language :: Python :: 3",
         "Topic :: Software Development :: Libraries :: Python Modules",
     ],
-    python_requires='>=3.9',
+    python_requires='>=3.8',
     zip_safe=False,
     include_package_data=True,
 )

tests/pytest.ini

@@ -1,3 +1,5 @@
 [pytest]
+log_cli = true
+log_cli_level = 0
 log_format = %(asctime)s %(levelname)s %(message)s
 log_date_format = %Y-%m-%d %H:%M:%S

tests/test.py

@@ -1,15 +1,27 @@
 """Test for Geocaching Api integration."""
-from geocachingapi import GeocachingApi, GeocachingStatus
+from geocachingapi import GeocachingApi, GeocachingStatus, GeocachingSettings
 #For adding your token add a file call token.py to current folder and add TOKEN = "<your token here>"
 from .token import TOKEN
 import asyncio
 import pytest
 
+# @pytest.mark.asyncio
+import logging
+logging.basicConfig(level=logging.DEBUG)
+
+mylogger = logging.getLogger()
+
 @pytest.mark.asyncio
 async def test():
     status:GeocachingStatus = None
     api = GeocachingApi(token=TOKEN)
     status = await api.update()
+    print(status.user.reference_code)
+    assert(status.user.reference_code is not None)
+    assert(status.user.find_count is not None)
+    await api.close()
+    assert(False)
+
+
+
 
-    assert(status.user.username is not None)
-    assert(status.user.find_count is not None)