Merge pull request meraki#119 from coreGreenberet/asyncio_semaphore

Asyncio semaphore

Author: Shiyue Cheng

meraki/aio/rest_session.py

@@ -12,49 +12,7 @@
 from ..config import *
 from ..exceptions import *
 from ..__init__ import __version__
-
-
-def user_agent_extended(be_geo_id, caller):
-    # Generate extended portion of the User-Agent
-    user_agent_extended = be_geo_id
-    user_agent_extended = {}
-
-    # Mimic pip system data collection per https://github.com/pypa/pip/blob/master/src/pip/_internal/network/session.py
-    user_agent_extended['implementation'] = {
-            "name": platform.python_implementation(),
-        }
-
-    if user_agent_extended["implementation"]["name"] in ('CPython','Jython','IronPython'):
-        user_agent_extended["implementation"]["version"] = platform.python_version()
-    elif user_agent_extended["implementation"]["name"] == 'PyPy':
-        if sys.pypy_version_info.releaselevel == 'final':
-            pypy_version_info = sys.pypy_version_info[:3]
-        else:
-            pypy_version_info = sys.pypy_version_info
-        user_agent_extended["implementation"]["version"] = ".".join(
-            [str(x) for x in pypy_version_info]
-        )
-
-    if sys.platform.startswith("darwin") and platform.mac_ver()[0]:
-        user_agent_extended["distro"] = {"name": "macOS", "version": platform.mac_ver()[0]}
-
-    if platform.system():
-        user_agent_extended.setdefault("system", {})["name"] = platform.system()
-
-    if platform.release():
-        user_agent_extended.setdefault("system", {})["release"] = platform.release()
-
-    if platform.machine():
-        user_agent_extended["cpu"] = platform.machine()
-
-    if be_geo_id:
-        user_agent_extended["be_geo_id"] = be_geo_id
-
-    if caller:
-        user_agent_extended["caller"] = caller
-
-    return urllib.parse.quote(json.dumps(user_agent_extended))
-
+from ..rest_session import user_agent_extended
 
 # Main module interface
 class AsyncRestSession:
@@ -93,8 +51,7 @@ def __init__(
         self._retry_4xx_error_wait_time = retry_4xx_error_wait_time
         self._maximum_retries = maximum_retries
         self._simulate = simulate
-        self._maximum_concurrent_sessions = maximum_concurrent_requests
-        self._current_sessions = 0
+        self._concurrent_requests_semaphore = asyncio.Semaphore(maximum_concurrent_requests)
         self._be_geo_id = be_geo_id
         self._caller = caller
 
@@ -133,14 +90,8 @@ def __init__(
             self._logger.info(f"Meraki dashboard API session initialized with these parameters: {self._parameters}")
 
     async def request(self, metadata, method, url, **kwargs):
-        while self._current_sessions >= self._maximum_concurrent_sessions:
-            await asyncio.sleep(0.3)  # wait for a free slot
-
-        self._current_sessions = self._current_sessions + 1
-        try:
+        async with self._concurrent_requests_semaphore:
             return await self._request(metadata, method, url, allow_redirects=False, **kwargs)
-        finally:
-            self._current_sessions = self._current_sessions - 1
 
     async def _request(self, metadata, method, url, **kwargs):
         # Metadata on endpoint

meraki_v0/aio/rest_session.py

@@ -11,49 +11,7 @@
 from ..config import *
 from ..exceptions import *
 from ..__init__ import __version__
-
-
-def user_agent_extended(be_geo_id, caller):
-    # Generate extended portion of the User-Agent
-    user_agent_extended = be_geo_id
-    user_agent_extended = {}
-
-    # Mimic pip system data collection per https://github.com/pypa/pip/blob/master/src/pip/_internal/network/session.py
-    user_agent_extended['implementation'] = {
-            "name": platform.python_implementation(),
-        }
-
-    if user_agent_extended["implementation"]["name"] in ('CPython','Jython','IronPython'):
-        user_agent_extended["implementation"]["version"] = platform.python_version()
-    elif user_agent_extended["implementation"]["name"] == 'PyPy':
-        if sys.pypy_version_info.releaselevel == 'final':
-            pypy_version_info = sys.pypy_version_info[:3]
-        else:
-            pypy_version_info = sys.pypy_version_info
-        user_agent_extended["implementation"]["version"] = ".".join(
-            [str(x) for x in pypy_version_info]
-        )
-
-    if sys.platform.startswith("darwin") and platform.mac_ver()[0]:
-        user_agent_extended["distro"] = {"name": "macOS", "version": platform.mac_ver()[0]}
-
-    if platform.system():
-        user_agent_extended.setdefault("system", {})["name"] = platform.system()
-
-    if platform.release():
-        user_agent_extended.setdefault("system", {})["release"] = platform.release()
-
-    if platform.machine():
-        user_agent_extended["cpu"] = platform.machine()
-
-    if be_geo_id:
-        user_agent_extended["be_geo_id"] = be_geo_id
-
-    if caller:
-        user_agent_extended["caller"] = caller
-
-    return urllib.parse.quote(json.dumps(user_agent_extended))
-
+from ..rest_session import user_agent_extended
 
 # Main module interface
 class AsyncRestSession:
@@ -92,8 +50,7 @@ def __init__(
         self._retry_4xx_error_wait_time = retry_4xx_error_wait_time
         self._maximum_retries = maximum_retries
         self._simulate = simulate
-        self._maximum_concurrent_sessions = maximum_concurrent_requests
-        self._current_sessions = 0
+        self._concurrent_requests_semaphore = asyncio.Semaphore(maximum_concurrent_requests)
         self._be_geo_id = be_geo_id
         self._caller = caller
 
@@ -132,14 +89,8 @@ def __init__(
             self._logger.info(f"Meraki dashboard API session initialized with these parameters: {self._parameters}")
 
     async def request(self, metadata, method, url, **kwargs):
-        while self._current_sessions >= self._maximum_concurrent_sessions:
-            await asyncio.sleep(0.3)  # wait for a free slot
-
-        self._current_sessions = self._current_sessions + 1
-        try:
+        async with self._concurrent_requests_semaphore:
             return await self._request(metadata, method, url, allow_redirects=False, **kwargs)
-        finally:
-            self._current_sessions = self._current_sessions - 1
 
     async def _request(self, metadata, method, url, **kwargs):
         # Metadata on endpoint