Environment selection (#17)

* 🧹 Less generic name for token file
* ✨ Option to choose between staging and production API
* ⚡ Update test and example

Co-authored-by: Rudolf Offereins <rudolf.offereins@netivity.nl>

Author: Sholofly

.gitignore

@@ -18,5 +18,5 @@ var/
 logs/
 __pycache__/
 .pytest_cache/
-token.py
+my_token.py
 .pypirc

examples/control.py

@@ -1,12 +1,15 @@
 """Asynchronous Python client for the Geocaching API."""
 import asyncio
 from geocachingapi import GeocachingApi
+from geocachingapi.models import GeocachingApiEnvironment
 
 async def main():
     """Show example of using the Geocaching API"""
-    async with GeocachingApi() as api:
-        print("test")
+    async with GeocachingApi(token="<insert your token here>", environment=GeocachingApiEnvironment.Staging) as api:
+        status = await api.update()
+        print(status.user.reference_code)
 
 if __name__ == "__main__":
     loop = asyncio.get_event_loop()
-    loop.run_until_complete(main())
+    loop.run_until_complete(main())
+    

geocachingapi/__init__.py

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

geocachingapi/const.py

@@ -1,9 +1,21 @@
 """Geocaching Api Constants."""
-GEOCACHING_API_SCHEME = "https"
-GEOCACHING_API_HOST = "staging.api.groundspeak.com"
-GEOCACHING_API_PORT = 443
-GEOCACHING_API_BASE_PATH = "/"
-GEOCACHING_API_VERSION = "v1"
+from .models import GeocachingApiEnvironment, GeocachingApiEnvironmentSettings
+
+ENVIRONMENT_SETTINGS = {
+    GeocachingApiEnvironment.Staging : GeocachingApiEnvironmentSettings(   
+            api_scheme= "https",
+            api_host= "staging.api.groundspeak.com",
+            api_port = 443,
+            api_base_bath="/v1",
+            ),
+            
+    GeocachingApiEnvironment.Production : GeocachingApiEnvironmentSettings(   
+            api_scheme= "https",
+            api_host= "api.groundspeak.com",
+            api_port = 443,
+            api_base_bath="/v1",
+            )
+}
 
 MEMBERSHIP_LEVELS = {
     0: "Unknown",

geocachingapi/geocachingapi.py

@@ -12,13 +12,7 @@
 from aiohttp import ClientResponse, ClientSession, ClientError
 
 from typing import Any, Awaitable, Callable, Dict, List, Optional
-from .const import (
-    GEOCACHING_API_BASE_PATH,
-    GEOCACHING_API_HOST,
-    GEOCACHING_API_PORT,
-    GEOCACHING_API_SCHEME,
-    GEOCACHING_API_VERSION,
-)
+from .const import ENVIRONMENT_SETTINGS
 from .exceptions import (
     GeocachingApiConnectionError,
     GeocachingApiConnectionTimeoutError,
@@ -28,7 +22,9 @@
 
 from .models import (
     GeocachingStatus,
-    GeocachingSettings
+    GeocachingSettings,
+    GeocachingApiEnvironment,
+    GeocachingApiEnvironmentSettings
 )
 
 _LOGGER = logging.getLogger(__name__)
@@ -38,9 +34,11 @@ class GeocachingApi:
     _close_session: bool = False
     _status: GeocachingStatus = None
     _settings: GeocachingSettings = None
+    _environment_settings: GeocachingApiEnvironmentSettings = None
     def __init__(
         self,
         *,
+        environment: GeocachingApiEnvironment,
         token: str,
         settings: GeocachingSettings = None,
         request_timeout: int = 8,
@@ -49,6 +47,7 @@ def __init__(
 
     ) -> None:
         """Initialize connection with the Geocaching API."""
+        self._environment_settings = ENVIRONMENT_SETTINGS[environment]
         self._status = GeocachingStatus()
         self._settings = settings or GeocachingSettings(False)
         self._session = session
@@ -67,11 +66,12 @@ async def _request(self, method, uri, **kwargs) -> ClientResponse:
             _LOGGER.debug(f'Token refresh method called.')
 
         url = URL.build(
-            scheme=GEOCACHING_API_SCHEME,
-            host=GEOCACHING_API_HOST,
-            port=GEOCACHING_API_PORT,
-            path=GEOCACHING_API_BASE_PATH,
-        ).join(URL(uri))
+            scheme=self._environment_settings["api_scheme"],
+            host=self._environment_settings["api_host"],
+            port=self._environment_settings["api_port"],
+            path=self._environment_settings["api_base_bath"],
+        ) 
+        url = str(url) + uri
         _LOGGER.debug(f'Executing {method} API request to {url}.')
         headers = kwargs.get("headers")
 
@@ -155,7 +155,7 @@ async def _update_user(self, data: Dict[str, Any] = None) -> None:
                 "awardedFavoritePoints",
                 "membershipLevelId"
             ])
-            data = await self._request("GET", f"/{GEOCACHING_API_VERSION}/users/me?fields={fields}")
+            data = await self._request("GET", f"/users/me?fields={fields}")
         self._status.update_user_from_dict(data)
         _LOGGER.debug(f'User updated.')
 

geocachingapi/models.py

@@ -1,21 +1,32 @@
 from __future__ import annotations
-from typing import Any, Dict, Optional
+from enum import Enum
+from typing import Any, Dict, Optional, TypedDict
 
 from dataclasses import dataclass
 from datetime import datetime
 from .utils import try_get_from_dict
 
+class GeocachingApiEnvironmentSettings(TypedDict):
+    """Class to represent API environment settings"""
+    api_scheme:str
+    api_host:str
+    api_port: int
+    api_base_bath:str
+
+class GeocachingApiEnvironment(Enum):
+    """Enum to represent API environment"""
+    Staging = 1,
+    Production = 2,
 
 class GeocachingSettings:
     """Class to hold the Geocaching Api settings"""
-    fetch_trackables: bool = False
-    def __init__(self, fetch_trackables:bool = False) -> None:
+    fetch_trackables: bool
+    environment: GeocachingApiEnvironment
+
+    def __init__(self, fetch_trackables:bool = False, environment:GeocachingApiEnvironment = GeocachingApiEnvironment.Production ) -> None:
         """Initialize settings"""
         self.fetch_trackables = fetch_trackables
 
-
-
-
 @dataclass
 class GeocachingUser:
     """Class to hold the Geocaching user information"""

tests/test.py

@@ -1,27 +1,19 @@
 """Test for Geocaching Api integration."""
-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
+from geocachingapi import GeocachingApi, GeocachingStatus
+from geocachingapi.models import GeocachingApiEnvironment
+from my_token import TOKEN
 logging.basicConfig(level=logging.DEBUG)
-
 mylogger = logging.getLogger()
 
-@pytest.mark.asyncio
 async def test():
+    """Function to test GeocachingAPI  integration"""
     status:GeocachingStatus = None
-    api = GeocachingApi(token=TOKEN)
+    api = GeocachingApi(token=TOKEN, environment=GeocachingApiEnvironment.Staging)
     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)
-
-
-
-
+loop = asyncio.get_event_loop()
+loop.run_until_complete(test())
+loop.close()