move upload code to transferer

b2sdk/api.py

@@ -10,24 +10,20 @@
 
 import six
 
-from .account_info.sqlite_account_info import SqliteAccountInfo
-from .account_info.exception import MissingAccountData
-from .b2http import B2Http
 from .bucket import Bucket, BucketFactory
-from .cache import AuthInfoCache, DummyCache
-from .transferer import Transferer
 from .exception import NonExistentBucket, RestrictedBucket
-from .file_version import FileVersionInfoFactory, FileIdAndName
-from .part import PartFactory
-from .raw_api import API_VERSION, B2RawApi
+from .file_version import FileIdAndName
+from .large_file.services import LargeFileServices
+from .raw_api import API_VERSION
 from .session import B2Session
+from .transfer import (
+    CopyManager,
+    DownloadManager,
+    Emerger,
+    UploadManager,
+)
 from .utils import B2TraceMeta, b2_url_encode, limit_trace_arguments
 
-try:
-    import concurrent.futures as futures
-except ImportError:
-    import futures
-
 
 def url_for_api(info, api_name):
     """
@@ -44,6 +40,11 @@ def url_for_api(info, api_name):
     return '%s/b2api/%s/%s' % (base, API_VERSION, api_name)
 
 
+class Services(object):
+    def __init__(self, session):
+        self.large_file = LargeFileServices(session)
+
+
 @six.add_metaclass(B2TraceMeta)
 class B2Api(object):
     """
@@ -68,7 +69,14 @@ class handles several things that simplify the task of uploading
     BUCKET_FACTORY_CLASS = staticmethod(BucketFactory)
     BUCKET_CLASS = staticmethod(Bucket)
 
-    def __init__(self, account_info=None, cache=None, raw_api=None, max_upload_workers=10):
+    def __init__(
+        self,
+        account_info=None,
+        cache=None,
+        raw_api=None,
+        max_upload_workers=10,
+        max_copy_workers=10
+    ):
         """
         Initialize the API using the given account info.
 
@@ -77,14 +85,12 @@ def __init__(self, account_info=None, cache=None, raw_api=None, max_upload_worke
                       :class:`~b2sdk.v1.AbstractAccountInfo`
                       To learn more about Account Info objects, see here
                       :class:`~b2sdk.v1.SqliteAccountInfo`
-
         :param cache: an instance of the one of the following classes:
                       :class:`~b2sdk.cache.DummyCache`, :class:`~b2sdk.cache.InMemoryCache`,
                       :class:`~b2sdk.cache.AuthInfoCache`,
                       or any custom class derived from :class:`~b2sdk.cache.AbstractCache`
                       It is used by B2Api to cache the mapping between bucket name and bucket ids.
                       default is :class:`~b2sdk.cache.DummyCache`
-
         :param raw_api: an instance of one of the following classes:
                         :class:`~b2sdk.raw_api.B2RawApi`, :class:`~b2sdk.raw_simulator.RawSimulator`,
                         or any custom class derived from :class:`~b2sdk.raw_api.AbstractRawApi`
@@ -93,59 +99,40 @@ def __init__(self, account_info=None, cache=None, raw_api=None, max_upload_worke
                         default is :class:`~b2sdk.raw_api.B2RawApi`
 
         :param int max_upload_workers: a number of upload threads, default is 10
+        :param int max_copy_workers: a number of copy threads, default is 10
         """
-        self.raw_api = raw_api or B2RawApi(B2Http())
-        if account_info is None:
-            account_info = SqliteAccountInfo()
-            if cache is None:
-                cache = AuthInfoCache(account_info)
-        self.session = B2Session(self, self.raw_api)
-        self.transferer = Transferer(self.session, account_info)
-        self.account_info = account_info
-        if cache is None:
-            cache = DummyCache()
-        self.cache = cache
-        self.upload_executor = None
-        self.max_workers = max_upload_workers
-
-    def set_thread_pool_size(self, max_workers):
-        """
-        Set the size of the thread pool to use for uploads and downloads.
-
-        Must be called before any work starts, or the thread pool will get
-        the default size of 1.
+        self.session = B2Session(account_info=account_info, cache=cache, raw_api=raw_api)
+        self.services = Services(self.session)
+        self.download_manager = DownloadManager(self.session)
+        self.upload_manager = UploadManager(
+            self.session, self.services, max_upload_workers=max_upload_workers
+        )
+        self.copy_manager = CopyManager(
+            self.session, self.services, max_copy_workers=max_copy_workers
+        )
+        self.emerger = Emerger(
+            self.session, self.services, self.download_manager, self.upload_manager,
+            self.copy_manager
+        )
 
-        .. todo::
-            move set_thread_pool_size and get_thread_pool to transferer
+    @property
+    def account_info(self):
+        return self.session.account_info
 
-        :param int max_workers: maximum allowed number of workers in a pool
-        """
-        if self.upload_executor is not None:
-            raise Exception('thread pool already created')
-        self.max_workers = max_workers
+    @property
+    def cache(self):
+        return self.session.cache
 
-    def get_thread_pool(self):
-        """
-        Return the thread pool executor to use for uploads and downloads.
-        """
-        if self.upload_executor is None:
-            self.upload_executor = futures.ThreadPoolExecutor(max_workers=self.max_workers)
-        return self.upload_executor
+    @property
+    def raw_api(self):
+        return self.session.raw_api
 
     def authorize_automatically(self):
         """
         Perform automatic account authorization, retrieving all account data
         from account info object passed during initialization.
         """
-        try:
-            self.authorize_account(
-                self.account_info.get_realm(),
-                self.account_info.get_application_key_id(),
-                self.account_info.get_application_key(),
-            )
-        except MissingAccountData:
-            return False
-        return True
+        return self.session.authorize_automatically()
 
     @limit_trace_arguments(only=('self', 'realm'))
     def authorize_account(self, realm, application_key_id, application_key):
@@ -156,32 +143,7 @@ def authorize_account(self, realm, application_key_id, application_key):
         :param str application_key_id: :term:`application key ID`
         :param str application_key: user's :term:`application key`
         """
-        # Clean up any previous account info if it was for a different account.
-        try:
-            old_account_id = self.account_info.get_account_id()
-            old_realm = self.account_info.get_realm()
-            if application_key_id != old_account_id or realm != old_realm:
-                self.cache.clear()
-        except MissingAccountData:
-            self.cache.clear()
-
-        # Authorize
-        realm_url = self.account_info.REALM_URLS[realm]
-        response = self.raw_api.authorize_account(realm_url, application_key_id, application_key)
-        allowed = response['allowed']
-
-        # Store the auth data
-        self.account_info.set_auth_data(
-            response['accountId'],
-            response['authorizationToken'],
-            response['apiUrl'],
-            response['downloadUrl'],
-            response['recommendedPartSize'],
-            application_key,
-            realm,
-            allowed,
-            application_key_id,
-        )
+        self.session.authorize_account(realm, application_key_id, application_key)
 
     def get_account_id(self):
         """
@@ -245,11 +207,10 @@ def download_file_by_id(self, file_id, download_dest, progress_listener=None, ra
         position, and the second one is the end position in the file
         :return: context manager that returns an object that supports iter_content()
         """
-        url = self.session.get_download_url_by_id(
-            file_id,
-            url_factory=self.account_info.get_download_url,
+        url = self.session.get_download_url_by_id(file_id)
+        return self.download_manager.download_file_from_url(
+            url, download_dest, progress_listener, range_
         )
-        return self.transferer.download_file_from_url(url, download_dest, progress_listener, range_)
 
     def get_bucket_by_id(self, bucket_id):
         """
@@ -340,14 +301,9 @@ def list_parts(self, file_id, start_part_number=None, batch_size=None):
         :param int batch_size: the number of parts to fetch at a time from the server
         :rtype: generator
         """
-        batch_size = batch_size or 100
-        while True:
-            response = self.session.list_parts(file_id, start_part_number, batch_size)
-            for part_dict in response['parts']:
-                yield PartFactory.from_list_parts_dict(part_dict)
-            start_part_number = response.get('nextPartNumber')
-            if start_part_number is None:
-                break
+        return self.services.large_file.list_parts(
+            file_id, start_part_number=start_part_number, batch_size=batch_size
+        )
 
     # delete/cancel
     def cancel_large_file(self, file_id):
@@ -357,8 +313,7 @@ def cancel_large_file(self, file_id):
         :param str file_id: a file ID
         :rtype: None
         """
-        response = self.session.cancel_large_file(file_id)
-        return FileVersionInfoFactory.from_cancel_large_file_response(response)
+        return self.services.large_file.cancel_large_file(file_id)
 
     def delete_file_version(self, file_id, file_name):
         """