ENH add type annotations

.github/workflows/run_tests.yml

@@ -42,6 +42,11 @@ jobs:
         # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
         flake8 . --count --exit-zero --ignore=E402,C901 --max-line-length=127 --statistics
 
+    - name: Check types with mypy
+      continue-on-error: true
+      run: |
+        python -m mypy --install-types --non-interactive cottoncandy/
+
     - name: Test with pytest
       env:
         DL_BUCKET_NAME: ${{ secrets.DL_BUCKET_NAME }} 

cottoncandy/__init__.py

@@ -2,33 +2,34 @@
 '''
 
 
+from typing import Literal
 import os
 
 from cottoncandy import options
 
+from .browser import BrowserObject
+from .interfaces import DefaultInterface
 from .utils import get_keys, string2bool
 
 
 __version__ = "0.4.0"
 
-ACCESS_KEY = options.config.get('login', 'access_key')
-SECRET_KEY = options.config.get('login', 'secret_key')
-ENDPOINT_URL = options.config.get('login', 'endpoint_url')
+ACCESS_KEY: str = options.config.get('login', 'access_key')
+SECRET_KEY: str = options.config.get('login', 'secret_key')
+ENDPOINT_URL: str = options.config.get('login', 'endpoint_url')
 DEFAULT_SIGNATURE_VERSION = options.config.get('basic', 'signature_version')
 
-default_bucket = options.config.get('basic', 'default_bucket')
-force_bucket_creation = options.config.get('basic', 'force_bucket_creation')
-force_bucket_creation = string2bool(force_bucket_creation)
-
-
-def get_interface(bucket_name=default_bucket,
-                  ACCESS_KEY=ACCESS_KEY,
-                  SECRET_KEY=SECRET_KEY,
-                  endpoint_url=ENDPOINT_URL,
-                  force_bucket_creation=force_bucket_creation,
-                  verbose=True,
-                  backend='s3',
-                  **kwargs):
+default_bucket: str = options.config.get('basic', 'default_bucket')
+force_bucket_creation: bool = bool(string2bool(options.config.get('basic', 'force_bucket_creation')))
+
+def get_interface(bucket_name: str=default_bucket,
+                  ACCESS_KEY: str=ACCESS_KEY,
+                  SECRET_KEY: str=SECRET_KEY,
+                  endpoint_url: str=ENDPOINT_URL,
+                  force_bucket_creation: bool=force_bucket_creation,
+                  verbose: bool=True,
+                  backend: Literal['s3', 'gdrive', 'local']='s3',
+                  **kwargs) -> DefaultInterface:
     """Return an interface to the cloud.
 
     Parameters
@@ -38,7 +39,7 @@ def get_interface(bucket_name=default_bucket,
     SECRET_KEY : str
     endpoint_url : str
         The URL for the S3 gateway
-    backend : 's3'|'gdrive'
+    backend : 's3'|'gdrive'|'local'
         What backend to hook on to
     kwargs :
         S3 only. kwargs passed to botocore. For example,
@@ -48,7 +49,7 @@ def get_interface(bucket_name=default_bucket,
 
     Returns
     -------
-    cci : cottoncandy.InterfaceObject
+    cci : cottoncandy.DefaultInterface
     """
     from cottoncandy.interfaces import DefaultInterface
 
@@ -82,10 +83,10 @@ def get_interface(bucket_name=default_bucket,
     return interface
 
 
-def get_browser(bucket_name=default_bucket,
-                ACCESS_KEY=ACCESS_KEY,
-                SECRET_KEY=SECRET_KEY,
-                endpoint_url=ENDPOINT_URL):
+def get_browser(bucket_name: str=default_bucket,
+                ACCESS_KEY: str=ACCESS_KEY,
+                SECRET_KEY: str=SECRET_KEY,
+                endpoint_url: str=ENDPOINT_URL) -> BrowserObject:
     """Browser object that allows you to tab-complete your
     way through your objects
 
@@ -131,4 +132,4 @@ def get_browser(bucket_name=default_bucket,
 
     return S3Directory('/', interface=interface)
 
-__all__ = ['get_interface', 'get_browser', 'interfaces', 'browser']
+__all__ = ['get_interface', 'get_browser', 'interfaces', 'browser', 'DefaultInterface']

cottoncandy/backend.py

@@ -1,10 +1,23 @@
 from abc import ABCMeta, abstractmethod
+from typing import Any, NamedTuple, BinaryIO, Optional, Sequence
 
 
 class FileNotFoundError(RuntimeError):
     """File not found error"""
 
 
+class CloudStream(NamedTuple):
+    """
+    A simple unified representation of an object downloaded from the cloud.
+     .content is a streaming object with a .read() function
+     .metadata is a dictionary of the custom metadata of this object
+
+    TODO: unified metadata
+    """
+    content: BinaryIO
+    metadata: dict[str, str]
+
+
 class CCBackEnd:
     """
     Interface for cottoncandy backends
@@ -14,14 +27,20 @@ class CCBackEnd:
     def __init__(self):
         pass
 
+    @property
+    @abstractmethod
+    def bucket_name(self) -> Optional[str]:
+        """Name of the current bucket or backing path."""
+        pass
+
     ## Basic File IO
     @abstractmethod
-    def check_file_exists(self, file_name, bucket_name):
+    def check_file_exists(self, cloud_name: str, bucket_name: Optional[str] = None) -> bool:
         """Checks whether a file exists on the cloud
 
         Parameters
         ----------
-        file_name : str
+        cloud_name : str
             file on cloud
         bucket_name : str
             (s3) bucket to check in
@@ -33,7 +52,7 @@ def check_file_exists(self, file_name, bucket_name):
         pass
 
     @abstractmethod
-    def upload_stream(self, stream, cloud_name, metadata, permissions, threads):
+    def upload_stream(self, stream: BinaryIO, cloud_name: str, metadata: dict[str, str], permissions: Optional[str], threads: int) -> None:
         """Uploads a stream object with a .read() function
 
         Parameters
@@ -52,12 +71,12 @@ def upload_stream(self, stream, cloud_name, metadata, permissions, threads):
 
         Returns
         -------
-        bool, upload success
+        None
         """
         pass
 
     @abstractmethod
-    def upload_file(self, file_name, cloud_name, permissions, threads):
+    def upload_file(self, file_name: str, cloud_name: Optional[str] = None, permissions: Optional[str] = None, threads: int = 1) -> None:
         """Uploads a file from disk
 
         Parameters
@@ -74,13 +93,13 @@ def upload_file(self, file_name, cloud_name, permissions, threads):
 
         Returns
         -------
-        bool, upload success
+        None
         """
         pass
 
 
     @abstractmethod
-    def download_stream(self, cloud_name, threads):
+    def download_stream(self, cloud_name: str, threads: int) -> CloudStream:
         """Downloads a object to an in-memory stream
 
         Parameters
@@ -98,7 +117,7 @@ def download_stream(self, cloud_name, threads):
         pass
 
     @abstractmethod
-    def download_to_file(self, cloud_name, file_name, threads):
+    def download_to_file(self, cloud_name: str, file_name: str, threads: int) -> None:
         """Downloads an object directly to disk
 
         Parameters
@@ -113,14 +132,14 @@ def download_to_file(self, cloud_name, file_name, threads):
 
         Returns
         -------
-        bool, download success
+        None
         """
         pass
 
     ## Basic File management
 
     @abstractmethod
-    def list_directory(self, path, limit):
+    def list_directory(self, path: str, limit: int) -> list[str]:
         """Lists the content of a directory
 
         Parameters
@@ -130,12 +149,12 @@ def list_directory(self, path, limit):
 
         Returns
         -------
-
+        list[str]
         """
         pass
 
     @abstractmethod
-    def list_objects(self):
+    def list_objects(self) -> Sequence[Any]:
         """Gets all objects contained by backend
 
         Returns
@@ -145,7 +164,7 @@ def list_objects(self):
         pass
 
     @abstractmethod
-    def copy(self, source, destination, source_bucket, destination_bucket, overwrite):
+    def copy(self, source: str, destination: str, source_bucket: Optional[str] = None, destination_bucket: Optional[str] = None, overwrite: bool = False) -> Any:
         """Copies an object
 
         Parameters
@@ -168,25 +187,30 @@ def copy(self, source, destination, source_bucket, destination_bucket, overwrite
         pass
 
     @abstractmethod
-    def move(self, source, destination, source_bucket, destination_bucket, overwrite):
+    def move(self, source: str, destination: str, source_bucket: Optional[str] = None, destination_bucket: Optional[str] = None, overwrite: bool = False) -> Any:
         """Moves an object
 
         Parameters
         ----------
-        source
-        destination
-        source_bucket
-        destination_bucket
-        overwrite
+        source: str
+            origin path
+        destination: str
+            destination path
+        source_bucket: str
+            (s3) origin bucket
+        destination_bucket: str
+            (s3) destination bucket
+        overwrite: bool
+            overwrite if destination exists?
 
         Returns
         -------
-
+        bool, move success
         """
         pass
 
     @abstractmethod
-    def delete(self, file_name, recursive=False, delete=False):
+    def delete(self, cloud_name: str, recursive: bool = False, delete: bool = False) -> bool:
         """Deletes an object
 
         Parameters
@@ -200,30 +224,17 @@ def delete(self, file_name, recursive=False, delete=False):
 
         Returns
         -------
-
+        bool, delete success
         """
         pass
 
     @property
     @abstractmethod
-    def size(self):
+    def size(self) -> int:
         """Size of stored cloud items in bytes
 
         Returns
         -------
         int
         """
         pass
-
-
-class CloudStream:
-    """
-    A simple unified representation of an object downloaded from the cloud.
-     .content is a streaming object with a .read() function
-     .metadata is a dictionary of the custom metadata of this object
-
-    TODO: unified metadata
-    """
-    def __init__(self, stream, metadata):
-        self.content = stream
-        self.metadata = metadata

cottoncandy/browser.py

@@ -165,7 +165,7 @@ def __repr__(self):
     def __len__(self):
         return len(self._subdirs)
 
-    def __dir__(self):
+    def __dir__(self): # type: ignore[no-redef]
         return list(self._subdirs.keys())
 
     def __getattr__(self, attr):

cottoncandy/gdriveclient.py

@@ -15,11 +15,11 @@
 except ImportError:
     try:
         # support >=ipython-0.11, <ipython-1.0
-        from IPython.core.ipapi import get as get_ipython
+        from IPython.core.ipapi import get as get_ipython  # type: ignore
     except ImportError:
         # support <ipython-0.11
         try:
-            from IPython.ipapi import get as get_ipython
+            from IPython.ipapi import get as get_ipython  # type: ignore
         except ImportError:
             print('Not ipython')
 
@@ -70,6 +70,11 @@ def Authenticate(secrets, credentials):
 
         return authenticator
 
+    @property
+    def bucket_name(self):
+        print('Google drive has no concept of buckets')
+        return None
+
     def __init__(self, secrets='client_secrets.json', credentials='gdrive-credentials.txt'):
         """
 
@@ -896,7 +901,7 @@ def update_metadata(self, file_name, metadata):
             return False
 
     @property
-    def size(self):
+    def size(self) -> int:
         files = self.drive.ListFile({'q': "trashed=false"}).GetList()
         sizes = [f.metadata['size'] for f in files]
         return sum(sizes)