Skip to content

Asyncio semaphore #119

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
shiyuechengineer merged 2 commits into from
Oct 8, 2020
Merged

Asyncio semaphore #119

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
18a7a9a
removed duplicate function user_agent_extended
coreGreenberet Oct 8, 2020
a08f175
converted the internal concurrent request counter to asyncio.Semaphore
coreGreenberet Oct 8, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 3 additions & 52 deletions meraki/aio/rest_session.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -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
Expand Down
55 changes: 3 additions & 52 deletions meraki_v0/aio/rest_session.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -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
Expand Down