Saimon/support thread execution (#260)

Author: saimonation

cterasdk/clients/async_requests.py

@@ -17,11 +17,15 @@ def session_parameters(client_settings):
 class Session:
     """Asynchronous HTTP Session"""
 
-    def __init__(self, client_settings):
-        self.initialize(client_settings)
+    def __init__(self, settings):
+        self._settings = settings
+        self._session = None
 
-    def initialize(self, client_settings):
-        self._session = aiohttp.ClientSession(trace_configs=[async_tracers.default()], **session_parameters(client_settings))
+    @property
+    def session(self):
+        if self._session is None:
+            self._session = aiohttp.ClientSession(trace_configs=[async_tracers.default()], **session_parameters(self._settings))
+        return self._session
 
     @property
     def cookies(self):
@@ -41,7 +45,7 @@ async def await_promise(self, r, *, on_response=None):
 
     async def _request(self, r, *, on_response=None):
         try:
-            response = await self._session.request(r.method, r.url, **r.kwargs)
+            response = await self.session.request(r.method, r.url, **r.kwargs)
             return asyncio.create_task(on_response(response))
         except aiohttp.ClientSSLError as error:
             logging.getLogger('cterasdk.http').warning(error)
@@ -61,10 +65,11 @@ async def _request(self, r, *, on_response=None):
 
     @property
     def closed(self):
-        return self._session.closed
+        return self._session.closed if self._session is not None else True
 
     async def close(self):
-        await self._session.close()
+        if self._session is not None:
+            await self._session.close()
 
 
 class CookieJar:

cterasdk/clients/asynchronous/clients.py

@@ -35,7 +35,7 @@ async def delete(self, path, *, on_response=None, **kwargs):
 
     async def _request(self, request, *, on_response=None):
         on_response = on_response if on_response else AsyncResponse.new()
-        response = await self._async_session.await_promise(self.join_headers(request), on_response=on_response)
+        response = await self._session.await_promise(self.join_headers(request), on_response=on_response)
         return await errors.accept(response)
 
 
@@ -45,8 +45,8 @@ class AsyncWebDAV(AsyncClient):
 
 class AsyncJSON(AsyncClient):
 
-    def __init__(self, builder=None, async_session=None, authenticator=None, client_settings=None):
-        super().__init__(builder, async_session, authenticator, client_settings)
+    def __init__(self, builder=None, session=None, settings=None, authenticator=None):
+        super().__init__(builder, session, settings, authenticator)
         self.headers.update_headers({'Content-Type': 'application/json'})
 
     async def get(self, path, **kwargs):

cterasdk/clients/base.py

@@ -1,4 +1,3 @@
-import logging
 from . import async_requests
 from ..common import utils
 
@@ -32,23 +31,38 @@ def update_headers(self, headers):
 class BaseClient:
     """Base Client"""
 
-    def __init__(self, builder=None, async_session=None, authenticator=None, client_settings=None):
+    def __init__(self, builder=None, session=None, settings=None, authenticator=None):
         """
         Initialize a Client
 
         :param builder: Endpoint builder.
-        :param ,optional async_session: Re-use an asynchronous session.
+        :param ,optional session: Session.
+        :param ,optional settings: Client Session Settings.
         :param ,optional authenticator: Authenticator function.
         """
         self._headers = PersistentHeaders()
         self._authenticator = authenticator
         self._builder = builder
-        self._client_settings = client_settings
-        self._async_session = async_session if async_session else async_requests.Session(client_settings)
+        self._session = session if session else async_requests.Session(settings)
+
+    def clone(self, definition, builder=None, authenticator=None):
+        """
+        Clone a Client
+
+        :param class definition: Class definition.
+        :param ,optional builder: Endpoint builder.
+        :param ,optional authenticator: Authenticator function.
+        """
+        return definition(
+            builder if builder is not None else self._builder,
+            self._session,
+            None,
+            authenticator if authenticator is not None else self._authenticator,
+        )
 
     @property
     def cookies(self):
-        return CookieJar(self._async_session.cookies)
+        return CookieJar(self._session.cookies)
 
     @property
     def headers(self):
@@ -62,24 +76,17 @@ def join_headers(self, request):
     def baseurl(self):
         return self._builder()
 
-    def __str__(self):
-        return f"({self.__class__.__name__} client at {hex(hash(self))}, baseurl={self.baseurl})"
-
-    def _before_request(self):
-        if self._async_session.closed:
-            logging.getLogger('cterasdk.http').debug('Session Closed. Renewing.')
-            self._async_session.initialize(self._client_settings)
-
     def request(self, request, *, on_response=None):
-        self._before_request()
         return self._request(request, on_response=on_response)
 
     async def async_request(self, request, *, on_response=None):
-        self._before_request()
         return await self._request(request, on_response=on_response)
 
     async def close(self):
-        await self._async_session.close()
+        await self._session.close()
+
+    def __str__(self):
+        return f"({self.__class__.__name__} client at {hex(hash(self))}, baseurl={self.baseurl})"
 
 
 class BaseResponse:

cterasdk/clients/synchronous/clients.py

@@ -1,7 +1,6 @@
 
-import time
 import asyncio
-import logging
+import threading
 from . import errors
 from ..base import BaseClient
 from ..common import Serializers
@@ -50,10 +49,11 @@ def delete(self, path, *, on_response=None, **kwargs):
 
     def _request(self, request, *, on_response=None):
         on_response = on_response if on_response else SyncResponse.new()
-        return execute_request(self._async_session, self.join_headers(request), on_response=on_response)
+        response = execute(self._session.await_promise, self.join_headers(request), on_response=on_response)
+        return errors.accept(response)
 
     def close(self):  # pylint: disable=invalid-overridden-method
-        return asyncio.get_event_loop().run_until_complete(super().close())
+        return execute(super().close)
 
 
 class Folders(Client):
@@ -190,52 +190,73 @@ def login(self):
         return response.json()
 
 
-def execute_request(async_session, request, *, on_response, max_retries=3, backoff_factor=2):
-    retries = 0
-    while retries < max_retries:
-        try:
-            response = asyncio.get_event_loop().run_until_complete(async_session.await_promise(request, on_response=on_response))
-            return errors.accept(response)
-        except (ConnectionError, TimeoutError):
-            retries += 1
-            if retries < max_retries:
-                delay = backoff_factor ** retries
-                logging.getLogger('cterasdk.http').warning("Retrying in %s seconds.", delay)
-                time.sleep(delay)
-            else:
-                logging.getLogger('cterasdk.http').error("Max retries reached. Request failed.")
-                raise
-    return None
+event_loop = asyncio.new_event_loop()
+
+
+def execute(target, *args, **kwargs):
+    loop = asyncio.get_event_loop()
+
+    if not loop.is_running():
+        return loop.run_until_complete(target(*args, **kwargs))
+
+    return run_threadsafe(event_loop, target, *args, **kwargs)
 
 
 class SyncResponse(AsyncResponse):
     """Synchronous Response Object"""
 
-    def __init__(self, response):
-        super().__init__(response)
-        self._executor = asyncio.get_event_loop()
-
     def iter_content(self, chunk_size=None):
         while True:
             try:
-                yield self._executor.run_until_complete(super().async_iter_content(chunk_size).__anext__())
+                yield execute(super().async_iter_content(chunk_size).__anext__)
             except StopAsyncIteration:
                 break
 
     def text(self):  # pylint: disable=invalid-overridden-method
-        return self._consume_response(super().text)
+        return execute(super().text)
 
     def json(self):  # pylint: disable=invalid-overridden-method
-        return self._consume_response(super().json)
+        return execute(super().json)
 
     def xml(self):  # pylint: disable=invalid-overridden-method
-        return self._consume_response(super().xml)
-
-    def _consume_response(self, consumer):
-        return self._executor.run_until_complete(consumer())
+        return execute(super().xml)
 
     @staticmethod
     def new():
         async def new_response(response):
             return SyncResponse(response)
         return new_response
+
+
+def run_threadsafe(loop, target, *args, **kwargs):
+    event = threading.Event()
+
+    t = Task(loop, event, target, *args, **kwargs)
+    t.start()
+
+    event.wait()
+
+    if t.exception:
+        raise t.exception
+    return t.response
+
+
+class Task(threading.Thread):
+
+    def __init__(self, loop, event, target, *args, **kwargs):
+        super().__init__(name='Thread-safe Executor')
+        self.loop = loop
+        self.event = event
+        self.target = target
+        self.args = args
+        self.kwargs = kwargs
+        self.exception = None
+        self.response = None
+
+    def run(self):
+        try:
+            self.response = self.loop.run_until_complete(self.target(*self.args, **self.kwargs))
+        except Exception as e:  # pylint: disable=broad-exception-caught
+            self.exception = e
+        finally:
+            self.event.set()

cterasdk/direct/client.py

@@ -33,10 +33,12 @@ def __init__(self, baseurl, credentials):
         :param str baseurl: Portal URL
         :param cterasdk.objects.asynchronous.directio.Credentials credentials: Credentials
         """
-        ctera_direct = cterasdk.settings.sessions.ctera_direct
-        self._api = AsyncJSON(EndpointBuilder.new(baseurl, '/directio'), authenticator=lambda *_: True,
-                              client_settings=client_settings(ctera_direct.api))
-        self._client = AsyncClient(DefaultBuilder(), authenticator=lambda *_: True, client_settings=client_settings(ctera_direct.storage))
+        self._api = AsyncJSON(EndpointBuilder.new(baseurl, '/directio'),
+                              settings=client_settings(cterasdk.settings.sessions.ctera_direct.api),
+                              authenticator=lambda *_: True)
+        self._client = AsyncClient(DefaultBuilder(),
+                                   settings=client_settings(cterasdk.settings.sessions.ctera_direct.storage),
+                                   authenticator=lambda *_: True)
         self._credentials = credentials
 
     async def _direct(self, file_id):

cterasdk/objects/asynchronous/core.py

@@ -24,17 +24,13 @@ def __init__(self, core):
 class V1:
 
     def __init__(self, core):
-        session = core._generic._async_session
-        self.api = clients.AsyncAPI(EndpointBuilder.new(core.base, core.context, '/api'), session, core._authenticator,
-                                    core._generic._client_settings)
+        self.api = core.default.clone(clients.AsyncAPI, EndpointBuilder.new(core.base, core.context, '/api'))
 
 
 class V2:
 
     def __init__(self, core):
-        session = core._generic._async_session
-        self.api = clients.AsyncJSON(EndpointBuilder.new(core.base, core.context, '/v2/api'), session,
-                                     core._authenticator, core._generic._client_settings)
+        self.api = core.default.clone(clients.AsyncJSON, EndpointBuilder.new(core.base, core.context, '/v2/api'))
 
 
 class IO:
@@ -46,9 +42,7 @@ def __init__(self, core):
 class WebDAV:
 
     def __init__(self, core):
-        session = core._generic._async_session
-        self._webdav = clients.AsyncWebDAV(EndpointBuilder.new(core.base, core.context, '/webdav'), session, core._authenticator,
-                                           core._generic._client_settings)
+        self._webdav = core.default.clone(clients.AsyncWebDAV, EndpointBuilder.new(core.base, core.context, '/webdav'))
 
     @property
     def download(self):
@@ -62,15 +56,20 @@ async def __aenter__(self):
 
     def __init__(self, host, port=None, https=True):
         super().__init__(host, port, https, base=None)
-        self._generic = clients.AsyncClient(EndpointBuilder.new(self.base), authenticator=self._authenticator,
-                                            client_settings=client_settings(cterasdk.settings.sessions.metadata_connector))
+        self._default = clients.AsyncClient(EndpointBuilder.new(self.base),
+                                            settings=client_settings(cterasdk.settings.sessions.metadata_connector),
+                                            authenticator=self._authenticator)
         self._ctera_session = Session(self.host(), self.context)
         self._ctera_clients = Clients(self)
 
         self.cloudfs = cloudfs.CloudFS(self)
         self.notifications = notifications.Notifications(self)
         self.users = users.Users(self)
 
+    @property
+    def default(self):
+        return self._default
+
     @property
     def v1(self):
         return self.clients.v1
@@ -93,7 +92,7 @@ async def logout(self):
         if self._ctera_session.connected:
             await self._login_object.logout()
             self._ctera_session.stop_session()
-        await self._generic.close()
+        await self.default.close()
 
     @property
     def _login_object(self):
@@ -103,7 +102,7 @@ def _authenticator(self, url):
         return authenticators.core(self.session(), url, self.context)
 
     async def __aexit__(self, exc_type, exc, tb):
-        await self._generic.close()
+        await self.default.close()
 
 
 class AsyncGlobalAdmin(AsyncPortal):

cterasdk/objects/services.py

@@ -91,8 +91,9 @@ def __enter__(self):
 
     def __init__(self, host, port, https, base):
         super().__init__(host, port, https, base)
-        self._generic = clients.Client(endpoints.EndpointBuilder.new(self.base), authenticator=self._authenticator,
-                                       client_settings=client_settings(cterasdk.settings.sessions.management))
+        self._default = clients.Client(endpoints.EndpointBuilder.new(self.base),
+                                       settings=client_settings(cterasdk.settings.sessions.management),
+                                       authenticator=self._authenticator)
 
     def login(self, username, password):
         """
@@ -111,7 +112,7 @@ def logout(self):
         if self._ctera_session.connected:
             self._login_object.logout()
             self._ctera_session.stop_session()
-        self._generic.close()
+        self._default.close()
 
     @abstractmethod
     def test(self):
@@ -122,17 +123,21 @@ def test(self):
     def _session_id_key(self):
         return NotImplementedError("Subclass must implement the '_session_id_key' property")
 
+    @property
+    def default(self):
+        return self._default
+
     def get_session_id(self):
         """
         Get Session Identifier
 
         :return str: Session ID
         """
-        return self._generic.cookies.get(self._session_id_key)
+        return self._default.cookies.get(self._session_id_key)
 
     def set_session_id(self, session_id):
-        self._generic.cookies.update({self._session_id_key: session_id}, self._generic.baseurl)
+        self._default.cookies.update({self._session_id_key: session_id}, self._default.baseurl)
         self._ctera_session.start_session(self)
 
     def __exit__(self, exc_type, exc_value, exc_tb):
-        self._generic.close()
+        self._default.close()