rest_session.py extractions and minor improvements (#296)

* Extractathon!

* Extracting parts of rest_session.py into common.py
* Creating input validation for caller and be_geo_id arguments
* Improving help text on config.py

* Update pyproject.toml

Tested and verified working on pytest 8.3.5. Bumping version requirement.

* Fix missing return statement

* Clean up rest_session.py extractions

Author: TKIPisalegacycipher

meraki/aio/rest_session.py

@@ -5,7 +5,7 @@
 import sys
 import time
 import urllib.parse
-from datetime import datetime
+from datetime import datetime, timezone
 
 import aiohttp
 
@@ -66,18 +66,14 @@ def __init__(
         check_python_version()
 
         # Check base URL
-        if "v0" in self._base_url:
-            sys.exit(f'This library does not support dashboard API v0 ({self._base_url} was configured as the base'
-                     f' URL).  API v0 has been end of life since 2020 August 5.')
-        elif self._base_url[-1] == "/":
-            self._base_url = self._base_url[:-1]
+        reject_v0_base_url(self)
 
         # Update the headers for the session
         self._headers = {
             "Authorization": "Bearer " + self._api_key,
             "Content-Type": "application/json",
             "User-Agent": f"python-meraki/aio-{self._version} "
-                          + user_agent_extended(self._be_geo_id, self._caller),
+                          + validate_user_agent(self._be_geo_id, self._caller),
         }
         if self._certificate_path:
             self._sslcontext = ssl.create_default_context()

meraki/common.py

@@ -1,5 +1,8 @@
 import platform
 from meraki.exceptions import *
+import re
+import sys
+import urllib.parse
 
 
 def check_python_version():
@@ -20,3 +23,63 @@ def check_python_version():
         )
 
         raise PythonVersionError(message)
+
+
+def validate_user_agent(be_geo_id, caller):
+    # Generate extended portion of the User Agent
+    # Validate that it follows the expected format
+    user_agent = dict()
+
+    allowed_format_in_regex = r'^[A-Za-z0-9]+(?:/[0-9A-Za-z]+(?:\.[0-9A-Za-z]+)*(-[a-z]+)?)? [A-Za-z-0-9]+$'
+
+    if caller and re.match(allowed_format_in_regex, caller):
+        user_agent["caller"] = caller
+    elif be_geo_id and re.match(allowed_format_in_regex, be_geo_id):
+        user_agent["caller"] = be_geo_id
+    else:
+        if caller:
+            message = "Please follow the user agent format prescribed in our User Agents guide, available here:"
+            doc_link = "https://developer.cisco.com/meraki/api-v1/user-agents-overview/"
+            raise SessionInputError("MERAKI_PTYHON_SDK_CALLER", caller, message, doc_link)
+        elif be_geo_id:
+            message = "Use of be_geo_id is deprecated. Please use the argument MERAKI_PTYHON_SDK_CALLER instead."
+            doc_link = "https://developer.cisco.com/meraki/api-v1/user-agents-overview/"
+            raise SessionInputError("BE_GEO_ID", caller, message, doc_link)
+        else:
+            user_agent["caller"] = "unidentified"
+
+    caller_string = f'Caller/({user_agent["caller"]})'
+
+    return caller_string
+
+
+def reject_v0_base_url(self):
+    if 'v0' in self._base_url:
+        sys.exit(f'This library does not support dashboard API v0 ({self._base_url} was configured as the base'
+                 f' URL).  API v0 has been end of life since 2020 August 5.')
+    elif self._base_url[-1] == '/':
+        self._base_url = self._base_url[:-1]
+
+
+def iterator_for_get_pages_bool(self):
+    return self._use_iterator_for_get_pages
+
+
+def use_iterator_for_get_pages_setter(self, value):
+    if value:
+        self.get_pages = self._get_pages_iterator
+    else:
+        self.get_pages = self._get_pages_legacy
+
+    self._use_iterator_for_get_pages = value
+
+
+def validate_base_url(self, url):
+    allowed_domains = ['meraki.com', 'meraki.ca', 'meraki.cn', 'meraki.in', 'gov-meraki.com']
+    parsed_url = urllib.parse.urlparse(url)
+    if any(domain in parsed_url.netloc for domain in allowed_domains):
+        abs_url = url
+    else:
+        abs_url = self._base_url + url
+    return abs_url
+

meraki/config.py

@@ -82,13 +82,15 @@
 # 1. Application name precedes vendor name in all cases.
 # 2. If your application or vendor name normally contains spaces or special casing, you should omit them in favor of
 #    normal CamelCasing here.
-# 3. The optional slash and version number are optional. Leave both out if you like.
+# 3. The slash and version number are optional. Leave both out if you like.
 # 4. The slash is a forward slash, '/' -- not a backslash.
-# 5. Don't use the 'Meraki' name in your application name here. Maybe in general? I'm a config file, not a lawyer.
+# 5. Don't use the 'Meraki' or 'Cisco' names in your application name here. Maybe in general? I'm a config file, not a
+#    lawyer.
 # Example 1: if your application is named 'Mambo', version number is 5.0, and your vendor/company name is Vega, then
 # you would use, at minimum: 'Mambo Vega'. Optionally: 'Mambo/5.0 Vega'.
 # Example 2: if your application is named 'Sunshine Rainbows', and company name is 'hunter2 for Life', and if you
 # don't want to report version number, then you would use, at minimum: 'SunshineRainbows hunter2ForLife'
 # The choice is yours as long as you follow the format. You should **not** include other information in this string.
 # If you are an official ecosystem partner, this is required.
+# For more guidance, please refer to https://developer.cisco.com/meraki/api-v1/user-agents-overview/
 MERAKI_PYTHON_SDK_CALLER = ""

meraki/exception_handler.py

@@ -57,7 +57,7 @@ def __init__(self, metadata, response):
         except ValueError:
             self.message = self.response.content[:100].decode("UTF-8").strip()
             if (
-                type(self.message) == str
+                isinstance(self.message, str)
                 and self.status == 404
                 and self.reason == "Not Found"
             ):
@@ -85,7 +85,7 @@ def __init__(self, metadata, response, message):
             response.reason if response is not None and response.reason else None
         )
         self.message = message
-        if type(self.message) == str:
+        if isinstance(self.message, str):
             self.message = self.message.strip()
             if self.status == 404 and self.reason == "Not Found":
                 self.message += (
@@ -107,3 +107,15 @@ def __init__(self, message):
         self.message = message
 
         super().__init__(self.message)
+
+
+class SessionInputError(Exception): 
+    """Exception raised for unsupported session inputs."""
+    
+    def __init__(self, argument, value, message, doc_link):
+        self.argument = argument
+        self.value = value
+        self.message = message
+        self.doc_link = doc_link
+        
+        super().__init__(f'{self.message} {self.doc_link}')

meraki/exceptions.py

@@ -0,0 +1,8 @@
+def handle_3xx(self, response):
+    abs_url = response.headers['Location']
+    substring = 'meraki.com/api/v'
+    if substring not in abs_url:
+        substring = 'meraki.cn/api/v'
+    self._base_url = abs_url[:abs_url.find(substring) + len(substring) + 1]
+    return abs_url
+