Plugin support (#32)

A plugin should define a new subclass of CloudPath and reexport it as
`ExportedCloudPath`. It should also register new dispatches for all the
functions it implements.

---------

Co-authored-by: Shantanu <[email protected]>
Co-authored-by: Shantanu <[email protected]>

Author: taras-openai

boostedblob/__init__.py

@@ -35,3 +35,9 @@
 from .path import isfile as isfile
 from .path import stat as stat
 from .syncing import sync as sync
+
+# isort: off
+from .path_registry import _register_plugins
+
+_register_plugins()
+# isort: on

boostedblob/azure_auth.py

@@ -155,7 +155,9 @@ async def get_access_token(cache_key: tuple[str, str | None]) -> tuple[Any, floa
     # This enables the use of Managed Identity, Workload Identity, and other auth methods not implemented here
     if creds["_azure_auth"] == "azure-identity":
         try:
-            from azure.identity.aio import DefaultAzureCredential  # type: ignore
+            from azure.identity.aio import (  # type: ignore[import, unused-ignore]
+                DefaultAzureCredential,
+            )
         except ImportError as e:
             raise RuntimeError(
                 "When setting AZURE_USE_IDENTITY=1, you must also install the azure-identity package"

boostedblob/path.py

@@ -22,28 +22,28 @@
 class BasePath:
     @staticmethod
     def from_str(path: str) -> BasePath:
+        from .path_registry import try_get_cloud_path_type
+
         url = urllib.parse.urlparse(path)
-        if url.scheme == "gs":
-            return GooglePath.from_str(path)
-        if url.scheme == "az" or (
-            url.scheme == "https" and url.netloc.endswith(".blob.core.windows.net")
-        ):
-            return AzurePath.from_str(path)
+        cloud_path_type = try_get_cloud_path_type(url)
+        if cloud_path_type:
+            return cloud_path_type.from_str(path)
+
         if url.scheme:
             raise ValueError(f"Invalid path '{path}'")
         return LocalPath(path)
 
     @property
     def name(self) -> str:
         """Returns the name of path, normalised to exclude any trailing slash."""
-        raise NotImplementedError
+        raise NotImplementedError(f"Name not implemented for {type(self)}")
 
     @property
     def parent(self: T) -> T:
-        raise NotImplementedError
+        raise NotImplementedError(f"Parent not implemented for {type(self)}")
 
     def relative_to(self: T, other: T) -> str:
-        raise NotImplementedError
+        raise NotImplementedError(f"relative_to not implemented for {type(self)}")
 
     def is_relative_to(self: T, other: T) -> bool:
         try:
@@ -53,13 +53,13 @@ def is_relative_to(self: T, other: T) -> bool:
             return False
 
     def is_directory_like(self) -> bool:
-        raise NotImplementedError
+        raise NotImplementedError(f"is_directory_like not implemented for {type(self)}")
 
     def ensure_directory_like(self: T) -> T:
-        raise NotImplementedError
+        raise NotImplementedError(f"ensure_directory_like not implemented for {type(self)}")
 
     def __truediv__(self: T, relative_path: str) -> T:
-        raise NotImplementedError
+        raise NotImplementedError(f"__truediv__ not implemented for {type(self)}")
 
 
 @dataclass(frozen=True)
@@ -115,7 +115,12 @@ def __fspath__(self) -> str:
 
 
 class CloudPath(BasePath):
-    pass
+    @staticmethod
+    def is_cloud_path(url: urllib.parse.ParseResult) -> bool:
+        """
+        Returns True if the URL is a cloud path for this cloud provider
+        """
+        raise NotImplementedError
 
 
 @dataclass(frozen=True)
@@ -124,6 +129,12 @@ class AzurePath(CloudPath):
     container: str
     blob: str
 
+    @staticmethod
+    def is_cloud_path(url: urllib.parse.ParseResult) -> bool:
+        return url.scheme == "az" or (
+            url.scheme == "https" and url.netloc.endswith(".blob.core.windows.net")
+        )
+
     @staticmethod
     def from_str(url: str) -> AzurePath:
         parsed_url = urllib.parse.urlparse(url)
@@ -196,6 +207,10 @@ class GooglePath(CloudPath):
     bucket: str
     blob: str
 
+    @staticmethod
+    def is_cloud_path(url: urllib.parse.ParseResult) -> bool:
+        return url.scheme == "gs"
+
     @staticmethod
     def from_str(url: str) -> GooglePath:
         parsed_url = urllib.parse.urlparse(url)

boostedblob/path_registry.py

@@ -0,0 +1,47 @@
+import functools
+import importlib
+import pkgutil
+import urllib.parse
+from typing import Optional, Sequence
+
+from .path import CloudPath
+
+plugins: list[type[CloudPath]] = []
+
+
+def _available_plugin_modules() -> Sequence[str]:
+    # boostedblob_ext is a namespace package
+    # submodules inside boostedblob_ext will be inspected for ExportedCloudPath attributes
+    # - we use namespace package pattern so `pkgutil.iter_modules` is fast
+    # - it's a separate top-level package because namespace subpackages of non-namespace
+    #   packages don't quite do what you want with editable installs
+    try:
+        import boostedblob_ext  # type: ignore[import, unused-ignore]
+    except ImportError:
+        return []
+
+    mods = []
+    plugin_mods = pkgutil.iter_modules(boostedblob_ext.__path__, boostedblob_ext.__name__ + ".")
+    for _, mod_name, _ in plugin_mods:
+        mods.append(mod_name)
+    return mods
+
+
+@functools.cache
+def _register_plugins() -> None:
+    from .path import AzurePath, GooglePath
+
+    plugins.append(AzurePath)
+    plugins.append(GooglePath)
+
+    for mod_name in _available_plugin_modules():
+        mod = importlib.import_module(mod_name)
+        if hasattr(mod, "ExportedCloudPath"):
+            plugins.append(mod.ExportedCloudPath)
+
+
+def try_get_cloud_path_type(url: urllib.parse.ParseResult) -> Optional[type[CloudPath]]:
+    for plugin in plugins:
+        if plugin.is_cloud_path(url):
+            return plugin
+    return None