Add type annotations to initial keyword args

This commit is contained in:
nuwang
2023-02-09 18:14:18 +05:30
parent 9b7257cb43
commit edc130ed32
7 changed files with 39 additions and 21 deletions
+12 -8
View File
@@ -36,11 +36,15 @@ class FilesSourceProperties(TypedDict):
file_sources_config: NotRequired["ConfiguredFileSourcesConfig"]
id: NotRequired[str]
label: NotRequired[str]
doc: NotRequired[str]
doc: NotRequired[Optional[str]]
scheme: NotRequired[str]
writable: NotRequired[bool]
required_roles: NotRequired[List[str]]
requires_groups: NotRequired[List[str]]
requires_roles: NotRequired[Optional[str]]
requires_groups: NotRequired[Optional[str]]
# API helper values
uri_root: NotRequired[str]
type: NotRequired[str]
browsable: NotRequired[bool]
class FilesSourceOptions:
@@ -100,7 +104,7 @@ class SingleFileSource(metaclass=abc.ABCMeta):
returned unchanged."""
@abc.abstractmethod
def to_dict(self, for_serialization=False, user_context=None):
def to_dict(self, for_serialization=False, user_context=None) -> FilesSourceProperties:
"""Return a dictified representation of this FileSource instance.
If ``user_context`` is supplied, properties should be written so user
@@ -114,7 +118,7 @@ class SupportsBrowsing(metaclass=abc.ABCMeta):
"""Return a prefix for the root (e.g. gxfiles://prefix/)."""
@abc.abstractmethod
def list(self, path="/", recursive=False, user_context=None, opts: Optional[FilesSourceOptions] = None):
def list(self, path="/", recursive=False, user_context=None, opts: Optional[FilesSourceOptions] = None) -> dict:
"""Return dictionary of 'Directory's and 'File's."""
@@ -175,7 +179,7 @@ class BaseFilesSource(FilesSource):
uri_root = self.get_uri_root()
return uri_join(uri_root, path)
def _parse_common_config_opts(self, kwd: dict):
def _parse_common_config_opts(self, kwd: FilesSourceProperties):
self._file_sources_config = kwd.pop("file_sources_config")
self.id = kwd.pop("id")
self.label = kwd.pop("label", None) or self.id
@@ -191,8 +195,8 @@ class BaseFilesSource(FilesSource):
kwd.pop("browsable", None)
return kwd
def to_dict(self, for_serialization=False, user_context=None):
rval = {
def to_dict(self, for_serialization=False, user_context=None) -> FilesSourceProperties:
rval: FilesSourceProperties = {
"id": self.id,
"type": self.plugin_type,
"label": self.label,
+3 -1
View File
@@ -13,10 +13,12 @@ from typing import (
import fs
from fs.base import FS
from typing_extensions import Unpack
from . import (
BaseFilesSource,
FilesSourceOptions,
FilesSourceProperties,
)
log = logging.getLogger(__name__)
@@ -28,7 +30,7 @@ class PyFilesystem2FilesSource(BaseFilesSource):
required_module: ClassVar[Optional[Type[FS]]]
required_package: ClassVar[str]
def __init__(self, **kwd):
def __init__(self, **kwd: Unpack[FilesSourceProperties]):
if self.required_module is None:
raise Exception(PACKAGE_MESSAGE % self.required_package)
props = self._parse_common_config_opts(kwd)
+7 -4
View File
@@ -2,9 +2,12 @@ import base64
import logging
from typing import Optional
from typing_extensions import Unpack
from . import (
BaseFilesSource,
FilesSourceOptions,
FilesSourceProperties,
)
log = logging.getLogger(__name__)
@@ -13,11 +16,11 @@ log = logging.getLogger(__name__)
class Base64FilesSource(BaseFilesSource):
plugin_type = "base64"
def __init__(self, label="Base64 encoded string", doc="Base64 string handler", **kwd):
kwds = dict(
def __init__(self, **kwd: Unpack[FilesSourceProperties]):
kwds: FilesSourceProperties = dict(
id="_base64",
label=label,
doc=doc,
label="Base64 encoded string",
doc="Base64 string handler",
writable=False,
)
kwds.update(kwd)
+7 -4
View File
@@ -1,10 +1,13 @@
import logging
from typing import Optional
from typing_extensions import Unpack
from galaxy.util.drs import fetch_drs_to_file
from . import (
BaseFilesSource,
FilesSourceOptions,
FilesSourceProperties,
)
log = logging.getLogger(__name__)
@@ -13,11 +16,11 @@ log = logging.getLogger(__name__)
class DRSFilesSource(BaseFilesSource):
plugin_type = "drs"
def __init__(self, label="DRS file", doc="DRS file handler", **kwd):
kwds = dict(
def __init__(self, **kwd: Unpack[FilesSourceProperties]):
kwds: FilesSourceProperties = dict(
id="_drs",
label=label,
doc=doc,
label="DRS file",
doc="DRS file handler",
writable=False,
)
kwds.update(kwd)
+4 -2
View File
@@ -7,6 +7,8 @@ from typing import (
Optional,
)
from typing_extensions import Unpack
from galaxy.util import (
DEFAULT_SOCKET_TIMEOUT,
get_charset_from_http_headers,
@@ -29,8 +31,8 @@ class HTTPFilesSourceProperties(FilesSourceProperties, total=False):
class HTTPFilesSource(BaseFilesSource):
plugin_type = "http"
def __init__(self, **kwd):
kwds = dict(
def __init__(self, **kwd: Unpack[FilesSourceProperties]):
kwds: FilesSourceProperties = dict(
id="_http",
label="HTTP File",
doc="Default HTTP file handler",
+3 -1
View File
@@ -8,6 +8,8 @@ from typing import (
Optional,
)
from typing_extensions import Unpack
from galaxy import exceptions
from galaxy.util.path import (
safe_contains,
@@ -42,7 +44,7 @@ class PosixFilesSource(BaseFilesSource):
# handle = OSFS(**self._props)
# return handle
def __init__(self, **kwd):
def __init__(self, **kwd: Unpack[PosixFilesSourceProperties]):
props = self._parse_common_config_opts(kwd)
self.root = props["root"]
self.enforce_symlink_security = props.get("enforce_symlink_security", DEFAULT_ENFORCE_SYMLINK_SECURITY)
+3 -1
View File
@@ -9,6 +9,8 @@ from typing import (
Optional,
)
from typing_extensions import Unpack
from . import (
FilesSourceOptions,
FilesSourceProperties,
@@ -38,7 +40,7 @@ class S3FsFilesSourceProperties(FilesSourceProperties, total=False):
class S3FsFilesSource(BaseFilesSource):
plugin_type = "s3fs"
def __init__(self, **kwd):
def __init__(self, **kwd: Unpack[S3FsFilesSourceProperties]):
if s3fs is None:
raise Exception("Package s3fs unavailable but required for this file source plugin.")
props: S3FsFilesSourceProperties = cast(S3FsFilesSourceProperties, self._parse_common_config_opts(kwd))