Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix CloudPathMeta.__call__ return type #330

Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
45 changes: 30 additions & 15 deletions cloudpathlib/cloudpath.py
Expand Up @@ -98,10 +98,12 @@
implementation_registry: Dict[str, CloudImplementation] = defaultdict(CloudImplementation)


def register_path_class(key: str) -> Callable:
T = TypeVar("T", bound=Type[CloudPath])
T = TypeVar("T")
CloudPathT = TypeVar("CloudPathT", bound="CloudPath")

def decorator(cls: Type[T]) -> Type[T]:

def register_path_class(key: str) -> Callable[[Type[CloudPathT]], Type[CloudPathT]]:
def decorator(cls: Type[CloudPathT]) -> Type[CloudPathT]:
if not issubclass(cls, CloudPath):
raise TypeError("Only subclasses of CloudPath can be registered.")
implementation_registry[key]._path_class = cls
Expand All @@ -112,34 +114,47 @@


class CloudPathMeta(abc.ABCMeta):
def __call__(cls, cloud_path, *args, **kwargs):
@overload
def __call__(cls: Type[T], cloud_path: CloudPathT, *args: Any, **kwargs: Any) -> CloudPathT:
...

Check warning on line 119 in cloudpathlib/cloudpath.py

View check run for this annotation

Codecov / codecov/patch

cloudpathlib/cloudpath.py#L119

Added line #L119 was not covered by tests

@overload
def __call__(
cls: Type[T], cloud_path: Union[str, "CloudPath"], *args: Any, **kwargs: Any
) -> T:
...

Check warning on line 125 in cloudpathlib/cloudpath.py

View check run for this annotation

Codecov / codecov/patch

cloudpathlib/cloudpath.py#L125

Added line #L125 was not covered by tests

def __call__(
cls: Type[T], cloud_path: Union[str, CloudPathT], *args: Any, **kwargs: Any
) -> Union[T, "CloudPath", CloudPathT]:
# cls is a class that is the instance of this metaclass, e.g., CloudPath
if not issubclass(cls, CloudPath):
raise TypeError(

Check warning on line 132 in cloudpathlib/cloudpath.py

View check run for this annotation

Codecov / codecov/patch

cloudpathlib/cloudpath.py#L132

Added line #L132 was not covered by tests
f"Only subclasses of {CloudPath.__name__} can be instantiated from its meta class."
)

# Dispatch to subclass if base CloudPath
if cls == CloudPath:
if cls is CloudPath:
for implementation in implementation_registry.values():
path_class = implementation._path_class
if path_class is not None and path_class.is_valid_cloudpath(
cloud_path, raise_on_error=False
):
# Instantiate path_class instance
new_obj = path_class.__new__(path_class, cloud_path, *args, **kwargs)
if isinstance(new_obj, path_class):
path_class.__init__(new_obj, cloud_path, *args, **kwargs)
new_obj = object.__new__(path_class)
path_class.__init__(new_obj, cloud_path, *args, **kwargs) # type: ignore[type-var]
return new_obj
valid = [
valid_prefixes = [
impl._path_class.cloud_prefix
for impl in implementation_registry.values()
if impl._path_class is not None
]
raise InvalidPrefixError(
f"Path {cloud_path} does not begin with a known prefix {valid}."
f"Path {cloud_path} does not begin with a known prefix {valid_prefixes}."
)

# Otherwise instantiate as normal
new_obj = cls.__new__(cls, cloud_path, *args, **kwargs)
if isinstance(new_obj, cls):
cls.__init__(new_obj, cloud_path, *args, **kwargs)
new_obj = object.__new__(cls)
cls.__init__(new_obj, cloud_path, *args, **kwargs) # type: ignore[type-var]
return new_obj

def __init__(cls, name: str, bases: Tuple[type, ...], dic: Dict[str, Any]) -> None:
Expand Down Expand Up @@ -449,7 +464,7 @@
newline: Optional[str] = None,
force_overwrite_from_cloud: bool = False, # extra kwarg not in pathlib
force_overwrite_to_cloud: bool = False, # extra kwarg not in pathlib
) -> IO:
) -> IO[Any]:
# if trying to call open on a directory that exists
if self.exists() and not self.is_file():
raise CloudPathIsADirectoryError(
Expand Down