Saimon/direct io streamer max workers (#318)

* add max workers, update change-log, update session expires error message to log in again

* list root directory if no path was provided

* add tls error and ancestors error exceptions

* Catch session expired and log when trying to log out on an expired session
Catch Forbidden and throw AuthenticationError for asynchronous login sessions to Portal

* introduce handling for conflicts when copying and moving files. allow to resolve conflicts by overriding, renaming or skipping files

* add ancestors exception and resolve unit test errors

* update file browser and changelog

* pass flake8

* resolve pylint errors

* pass final lint

* update docs to pass lint

* fix lint

* resolve ut errors and update changelog version

* ensure destination is writeable target, update to pass exceptions

* resolve regression in edge filer browser exceptions

* update to pass lint and flake8

---------

Co-authored-by: Saimon Michelson <saimon@localhost.localdomain>

Author: saimonation

cterasdk/asynchronous/core/files/browser.py

@@ -1,5 +1,6 @@
-from ....cio.core import CorePath
+from ....cio.core import CorePath, a_await_or_future
 from ....lib.storage import asynfs, commonfs
+from ....exceptions.io import FileConflict
 from ..base_command import BaseCommand
 from . import io
 
@@ -61,11 +62,11 @@ async def download_many(self, target, objects, destination=None):
         handle = await self.handle_many(target, *objects)
         return await asynfs.write(directory, name, handle)
 
-    async def listdir(self, path, depth=None, include_deleted=False):
+    async def listdir(self, path=None, depth=None, include_deleted=False):
         """
         List Directory
 
-        :param str path: Path
+        :param str,optional path: Path, defaults to the Cloud Drive root
         :param bool,optional include_deleted: Include deleted files, defaults to False
         """
         return await io.listdir(self._core, self.normalize(path), depth=depth, include_deleted=include_deleted)
@@ -105,18 +106,34 @@ async def public_link(self, path, access='RO', expire_in=30):
         """
         return await io.public_link(self._core, self.normalize(path), access, expire_in)
 
-    async def copy(self, *paths, destination=None, wait=False):
+    async def _try_with_resolver(self, func, *paths, destination=None, resolver=None, cursor=None, wait=False):
+        async def wrapper(resume_from=None):
+            ref = await func(self._core, *paths, destination=destination, resolver=resolver, cursor=resume_from)
+            return await a_await_or_future(self._core, ref, wait)
+
+        try:
+            return await wrapper(cursor)
+        except FileConflict as e:
+            if resolver:
+                return await wrapper(e.cursor)
+            raise
+
+    async def copy(self, *paths, destination=None, resolver=None, cursor=None, wait=False):
         """
         Copy one or more files or folders
 
         :param list[str] paths: List of paths
         :param str destination: Destination
+        :param cterasdk.core.types.ConflictResolver resolver: Conflict resolver, defaults to ``None``
+        :param cterasdk.common.object.Object cursor: Resume copy from cursor
         :param bool,optional wait: ``True`` Wait for task to complete, or ``False`` to return an awaitable task object.
         :returns: Task status object, or an awaitable task object
         :rtype: cterasdk.common.object.Object or :class:`cterasdk.lib.tasks.AwaitablePortalTask`
         """
         try:
-            return await io.copy(self._core, *[self.normalize(path) for path in paths], destination=self.normalize(destination), wait=wait)
+            return await self._try_with_resolver(io.copy, *[self.normalize(path) for path in paths],
+                                                 destination=self.normalize(destination),
+                                                 resolver=resolver, cursor=cursor, wait=wait)
         except ValueError:
             raise ValueError('Copy destination was not specified.')
 
@@ -188,7 +205,8 @@ async def rename(self, path, name, *, wait=False):
         :returns: Task status object, or an awaitable task object
         :rtype: cterasdk.common.object.Object or :class:`cterasdk.lib.tasks.AwaitablePortalTask`
         """
-        return await io.rename(self._core, self.normalize(path), name, wait=wait)
+        ref = await io.rename(self._core, self.normalize(path), name)
+        return await a_await_or_future(self._core, ref, wait)
 
     async def delete(self, *paths, wait=False):
         """
@@ -199,7 +217,8 @@ async def delete(self, *paths, wait=False):
         :returns: Task status object, or an awaitable task object
         :rtype: cterasdk.common.object.Object or :class:`cterasdk.lib.tasks.AwaitablePortalTask`
         """
-        return await io.remove(self._core, *[self.normalize(path) for path in paths], wait=wait)
+        ref = await io.remove(self._core, *[self.normalize(path) for path in paths])
+        return await a_await_or_future(self._core, ref, wait)
 
     async def undelete(self, *paths, wait=False):
         """
@@ -210,19 +229,24 @@ async def undelete(self, *paths, wait=False):
         :returns: Task status object, or an awaitable task object
         :rtype: cterasdk.common.object.Object or :class:`cterasdk.lib.tasks.AwaitablePortalTask`
         """
-        return await io.recover(self._core, *[self.normalize(path) for path in paths], wait=wait)
+        ref = await io.recover(self._core, *[self.normalize(path) for path in paths])
+        return await a_await_or_future(self._core, ref, wait)
 
-    async def move(self, *paths, destination=None, wait=False):
+    async def move(self, *paths, destination=None, resolver=None, cursor=None, wait=False):
         """
         Move one or more files or folders
 
         :param list[str] paths: List of paths
         :param str destination: Destination
+        :param cterasdk.core.types.ConflictResolver resolver: Conflict resolver, defaults to ``None``
+        :param cterasdk.common.object.Object cursor: Resume copy from cursor
         :param bool,optional wait: ``True`` Wait for task to complete, or ``False`` to return an awaitable task object.
         :returns: Task status object, or an awaitable task object
         :rtype: cterasdk.common.object.Object or :class:`cterasdk.lib.tasks.AwaitablePortalTask`
         """
         try:
-            return await io.move(self._core, *[self.normalize(path) for path in paths], destination=self.normalize(destination), wait=wait)
+            return await self._try_with_resolver(io.move, *[self.normalize(path) for path in paths],
+                                                 destination=self.normalize(destination),
+                                                 resolver=resolver, cursor=cursor, wait=wait)
         except ValueError:
             raise ValueError('Move destination was not specified.')

cterasdk/asynchronous/core/files/io.py

@@ -1,7 +1,7 @@
 import logging
-from ....cio.common import encode_request_parameter, a_await_or_future
+from ....cio.common import encode_request_parameter
 from ....cio import core as fs
-from ....exceptions.io import ResourceNotFoundError, NotADirectory, ResourceExistsError
+from ....exceptions.io import ResourceNotFoundError, ResourceExistsError
 from .. import query
 
 
@@ -68,40 +68,34 @@ async def makedirs(core, path):
             logger.debug('Resource already exists: %s', path.reference.as_posix())
 
 
-async def rename(core, path, name, *, wait=False):
+async def rename(core, path, name):
     with fs.rename(path, name) as param:
-        ref = await core.v1.api.execute('', 'moveResources', param)
-        return await a_await_or_future(core, ref, wait)
+        return await core.v1.api.execute('', 'moveResources', param)
 
 
-async def remove(core, *paths, wait=False):
+async def remove(core, *paths):
     with fs.delete(*paths) as param:
-        ref = await core.v1.api.execute('', 'deleteResources', param)
-        return await a_await_or_future(core, ref, wait)
+        return await core.v1.api.execute('', 'deleteResources', param)
 
 
-async def recover(core, *paths, wait=False):
+async def recover(core, *paths):
     with fs.recover(*paths) as param:
-        ref = await core.v1.api.execute('', 'restoreResources', param)
-        return await a_await_or_future(core, ref, wait)
+        return await core.v1.api.execute('', 'restoreResources', param)
 
 
-async def copy(core, *paths, destination=None, wait=False):
-    with fs.copy(*paths, destination=destination) as param:
-        ref = await core.v1.api.execute('', 'copyResources', param)
-        return await a_await_or_future(core, ref, wait)
+async def copy(core, *paths, destination=None, resolver=None, cursor=None):
+    with fs.copy(*paths, destination=destination, resolver=resolver, cursor=cursor) as param:
+        return await core.v1.api.execute('', 'copyResources', param)
 
 
-async def move(core, *paths, destination=None, wait=False):
-    with fs.move(*paths, destination=destination) as param:
-        ref = await core.v1.api.execute('', 'moveResources', param)
-        return await a_await_or_future(core, ref, wait)
+async def move(core, *paths, destination=None, resolver=None, cursor=None):
+    with fs.move(*paths, destination=destination, resolver=resolver, cursor=cursor) as param:
+        return await core.v1.api.execute('', 'moveResources', param)
 
 
 async def ensure_directory(core, directory, suppress_error=False):
     present, resource = await metadata(core, directory, suppress_error=True)
-    if (not present or not resource.isFolder) and not suppress_error:
-        raise NotADirectory(directory.absolute)
+    fs.ensure_directory(present, resource, directory, suppress_error)
     return resource.isFolder if present else False, resource
 
 
@@ -151,7 +145,9 @@ async def _validate_destination(core, name, destination):
     is_dir, resource = await ensure_directory(core, destination, suppress_error=True)
     if not is_dir:
         is_dir, resource = await ensure_directory(core, destination.parent)
+        fs.ensure_writeable(resource, destination.parent)
         return resource.cloudFolderInfo.uid, destination.name, destination.parent
+    fs.ensure_writeable(resource, destination)
     return resource.cloudFolderInfo.uid, name, destination
 
 

cterasdk/asynchronous/core/login.py

@@ -1,7 +1,9 @@
 import logging
 
 from .base_command import BaseCommand
-from ...exceptions import CTERAException
+from ...exceptions.transport import Forbidden
+from ...exceptions.session import SessionExpired
+from ...exceptions.auth import AuthenticationError
 
 
 logger = logging.getLogger('cterasdk.core')
@@ -23,9 +25,9 @@ async def login(self, username, password):
         try:
             await self._core.v1.api.form_data('/login', {'j_username': username, 'j_password': password})
             logger.info("User logged in. %s", {'host': host, 'user': username})
-        except CTERAException:
-            logger.error("Login failed. %s", {'host': host, 'user': username})
-            raise
+        except Forbidden as error:
+            logger.error('Login failed. %s', {'host': host, 'user': username})
+            raise AuthenticationError() from error
 
     async def sso(self, ctera_ticket):
         """
@@ -40,5 +42,8 @@ async def logout(self):
         """
         Log out of the portal
         """
-        await self._core.v1.api.form_data('/logout', {})
-        logger.info("User logged out. %s", {'host': self._core.host()})
+        try:
+            await self._core.v1.api.form_data('/logout', {})
+            logger.info("User logged out. %s", {'host': self._core.host()})
+        except SessionExpired:
+            logger.info("Session expired and is no longer active.")

cterasdk/asynchronous/core/notifications.py

@@ -7,7 +7,7 @@
 from ...common import Object
 from ...lib import CursorResponse
 from ...exceptions.transport import HTTPError
-from ...exceptions.notifications import NotificationsError
+from ...exceptions.notifications import NotificationsError, AncestorsError
 
 
 logger = logging.getLogger('cterasdk.notifications')
@@ -86,12 +86,12 @@ async def ancestors(self, descendant):
         param = Object()
         param.folder_id = descendant.folder_id
         param.guid = descendant.guid
-        logger.debug('Getting ancestors. %s', {'guid': param.guid, 'folder_id': param.folder_id})
+        logger.debug('Getting ancestors for: %s:%s', param.folder_id, param.guid)
         try:
             return await self._core.v2.api.post('/metadata/ancestors', param)
-        except HTTPError:
-            logger.error('Could not retrieve ancestors. %s', {'folder_id': param.folder_id, 'guid': param.guid})
-            raise
+        except HTTPError as error:
+            logger.error('Could not retrieve ancestors for: %s:%s', param.folder_id, param.guid)
+            raise AncestorsError(param.folder_id, param.guid) from error
 
 
 class Service(BaseCommand):

cterasdk/cio/common.py

@@ -77,27 +77,3 @@ def encode_request_parameter(param):
     return dict(
         inputXML=utf8_decode(toxmlstr(param))
     )
-
-
-def await_or_future(ctera, ref, wait):
-    """
-    Wait for task completion, or return an awaitable task object.
-
-    :param str ref: Task reference
-    :param bool wait: ``True`` to wait for task completion, ``False`` to return an awaitable task object
-    """
-    if wait:
-        return ctera.tasks.wait(ref)
-    return ctera.tasks.awaitable_task(ref)
-
-
-async def a_await_or_future(ctera, ref, wait):
-    """
-    Wait for task completion, or return an awaitable task object.
-
-    :param str ref: Task reference
-    :param bool wait: ``True`` to wait for task completion, ``False`` to return an awaitable task object
-    """
-    if wait:
-        return await ctera.tasks.wait(ref)
-    return ctera.tasks.awaitable_task(ref)