Switch from kwargs to options object

lib/galaxy/files/sources/__init__.py

@@ -13,7 +13,6 @@ from typing import (
 from typing_extensions import (
     NotRequired,
     TypedDict,
-    Unpack,
 )
 
 from galaxy.exceptions import (
@@ -44,9 +43,8 @@ class FilesSourceProperties(TypedDict):
     requires_groups: NotRequired[List[str]]
 
 
-class FilesSourceOptions(TypedDict):
-    # Overridees for file source initialization properties
-    extra_props: NotRequired[FilesSourceProperties]
+class FilesSourceOptions:
+    extra_props: Optional[FilesSourceProperties]
 
 
 class SingleFileSource(metaclass=abc.ABCMeta):
@@ -59,11 +57,15 @@ class SingleFileSource(metaclass=abc.ABCMeta):
         """Return a boolean indicating if the user can access the FileSource."""
 
     @abc.abstractmethod
-    def realize_to(self, source_path: str, native_path: str, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def realize_to(
+        self, source_path: str, native_path: str, user_context=None, opts: Optional[FilesSourceOptions] = None
+    ):
         """Realize source path (relative to uri root) to local file system path."""
 
     @abc.abstractmethod
-    def write_from(self, target_path: str, native_path: str, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def write_from(
+        self, target_path: str, native_path: str, user_context=None, opts: Optional[FilesSourceOptions] = None
+    ):
         """Write file at native path to target_path (relative to uri root)."""
 
     @abc.abstractmethod
@@ -112,7 +114,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, **kwargs: Unpack[FilesSourceOptions]):
+    def list(self, path="/", recursive=False, user_context=None, opts: Optional[FilesSourceOptions] = None):
         """Return dictionary of 'Directory's and 'File's."""
 
 
@@ -220,29 +222,29 @@ class BaseFilesSource(FilesSource):
         Used in to_dict method if for_serialization is True.
         """
 
-    def list(self, path="/", recursive=False, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def list(self, path="/", recursive=False, user_context=None, opts: Optional[FilesSourceOptions] = None):
         self._check_user_access(user_context)
-        return self._list(path, recursive, user_context, **kwargs)
+        return self._list(path, recursive, user_context, opts)
 
-    def _list(self, path="/", recursive=False, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def _list(self, path="/", recursive=False, user_context=None, opts: Optional[FilesSourceOptions] = None):
         pass
 
-    def write_from(self, target_path, native_path, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def write_from(self, target_path, native_path, user_context=None, opts: Optional[FilesSourceOptions] = None):
         if not self.get_writable():
             raise Exception("Cannot write to a non-writable file source.")
         self._check_user_access(user_context)
-        self._write_from(target_path, native_path, user_context=user_context, **kwargs)
+        self._write_from(target_path, native_path, user_context=user_context, opts=opts)
 
     @abc.abstractmethod
-    def _write_from(self, target_path, native_path, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def _write_from(self, target_path, native_path, user_context=None, opts: Optional[FilesSourceOptions] = None):
         pass
 
-    def realize_to(self, source_path, native_path, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def realize_to(self, source_path, native_path, user_context=None, opts: Optional[FilesSourceOptions] = None):
         self._check_user_access(user_context)
-        self._realize_to(source_path, native_path, user_context, **kwargs)
+        self._realize_to(source_path, native_path, user_context, opts=opts)
 
     @abc.abstractmethod
-    def _realize_to(self, source_path, native_path, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def _realize_to(self, source_path, native_path, user_context=None, opts: Optional[FilesSourceOptions] = None):
         pass
 
     def _check_user_access(self, user_context):

lib/galaxy/files/sources/_pyfilesystem2.py

@@ -13,7 +13,6 @@ from typing import (
 
 import fs
 from fs.base import FS
-from typing_extensions import Unpack
 
 from . import (
     BaseFilesSource,
@@ -36,13 +35,13 @@ class PyFilesystem2FilesSource(BaseFilesSource):
         self._props = props
 
     @abc.abstractmethod
-    def _open_fs(self, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def _open_fs(self, user_context=None, opts: Optional[FilesSourceOptions] = None):
         """Subclasses must instantiate a PyFilesystem2 handle for this file system."""
 
-    def _list(self, path="/", recursive=False, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def _list(self, path="/", recursive=False, user_context=None, opts: Optional[FilesSourceOptions] = None):
         """Return dictionary of 'Directory's and 'File's."""
 
-        with self._open_fs(user_context=user_context, **kwargs) as h:
+        with self._open_fs(user_context=user_context, opts=opts) as h:
             if recursive:
                 res: List[Dict[str, Any]] = []
                 for p, dirs, files in h.walk(path):
@@ -55,13 +54,13 @@ class PyFilesystem2FilesSource(BaseFilesSource):
                 to_dict = functools.partial(self._resource_info_to_dict, path)
                 return list(map(to_dict, res))
 
-    def _realize_to(self, source_path, native_path, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def _realize_to(self, source_path, native_path, user_context=None, opts: Optional[FilesSourceOptions] = None):
         with open(native_path, "wb") as write_file:
-            self._open_fs(user_context=user_context, **kwargs).download(source_path, write_file)
+            self._open_fs(user_context=user_context, opts=opts).download(source_path, write_file)
 
-    def _write_from(self, target_path, native_path, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def _write_from(self, target_path, native_path, user_context=None, opts: Optional[FilesSourceOptions] = None):
         with open(native_path, "rb") as read_file:
-            openfs = self._open_fs(user_context=user_context, **kwargs)
+            openfs = self._open_fs(user_context=user_context, opts=opts)
             dirname = fs.path.dirname(target_path)
             if not openfs.isdir(dirname):
                 openfs.makedirs(dirname)

lib/galaxy/files/sources/anvil.py

@@ -2,9 +2,10 @@ try:
     from anvilfs.anvilfs import AnVILFS
 except ImportError:
     AnVILFS = None
-from typing import Union
-
-from typing_extensions import Unpack
+from typing import (
+    Optional,
+    Union,
+)
 
 from . import (
     FilesSourceOptions,
@@ -18,9 +19,9 @@ class AnVILFilesSource(PyFilesystem2FilesSource):
     required_module = AnVILFS
     required_package = "fs.anvilfs"
 
-    def _open_fs(self, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def _open_fs(self, user_context=None, opts: Optional[FilesSourceOptions] = None):
         props = self._serialization_props(user_context)
-        extra_props: Union[FilesSourceProperties, dict] = kwargs.get("extra_props") or {}
+        extra_props: Union[FilesSourceProperties, dict] = opts.extra_props or {} if opts else {}
         handle = AnVILFS(**{**props, **extra_props})
         return handle
 

lib/galaxy/files/sources/base64.py

@@ -1,7 +1,6 @@
 import base64
 import logging
-
-from typing_extensions import Unpack
+from typing import Optional
 
 from . import (
     BaseFilesSource,
@@ -25,15 +24,19 @@ class Base64FilesSource(BaseFilesSource):
         props = self._parse_common_config_opts(kwds)
         self._props = props
 
-    def _realize_to(self, source_path: str, native_path: str, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def _realize_to(
+        self, source_path: str, native_path: str, user_context=None, opts: Optional[FilesSourceOptions] = None
+    ):
         with open(native_path, "wb") as temp:
             temp.write(base64.b64decode(source_path[len("base64://") :]))
             temp.flush()
 
-    def _write_from(self, target_path: str, native_path: str, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def _write_from(
+        self, target_path: str, native_path: str, user_context=None, opts: Optional[FilesSourceOptions] = None
+    ):
         raise NotImplementedError()
 
-    def score_url_match(self, url: str, **kwargs: Unpack[FilesSourceOptions]):
+    def score_url_match(self, url: str):
         if url.startswith("base64://"):
             return len("base64://")
         else:

lib/galaxy/files/sources/basespace.py

@@ -3,9 +3,10 @@ try:
 except ImportError:
     BASESPACEFS = None
 
-from typing import Union
-
-from typing_extensions import Unpack
+from typing import (
+    Optional,
+    Union,
+)
 
 from . import (
     FilesSourceOptions,
@@ -19,9 +20,9 @@ class BaseSpaceFilesSource(PyFilesystem2FilesSource):
     required_module = BASESPACEFS
     required_package = "fs-basespace"
 
-    def _open_fs(self, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def _open_fs(self, user_context=None, opts: Optional[FilesSourceOptions] = None):
         props = self._serialization_props(user_context)
-        extra_props: Union[FilesSourceProperties, dict] = kwargs.get("extra_props") or {}
+        extra_props: Union[FilesSourceProperties, dict] = opts.extra_props or {} if opts else {}
         handle = BASESPACEFS(**{**props, **extra_props})
         return handle
 

lib/galaxy/files/sources/dropbox.py

@@ -3,9 +3,10 @@ try:
 except ImportError:
     DropboxFS = None
 
-from typing import Union
-
-from typing_extensions import Unpack
+from typing import (
+    Optional,
+    Union,
+)
 
 from . import (
     FilesSourceOptions,
@@ -19,9 +20,9 @@ class DropboxFilesSource(PyFilesystem2FilesSource):
     required_module = DropboxFS
     required_package = "fs.dropboxfs"
 
-    def _open_fs(self, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def _open_fs(self, user_context=None, opts: Optional[FilesSourceOptions] = None):
         props = self._serialization_props(user_context)
-        extra_props: Union[FilesSourceProperties, dict] = kwargs.get("extra_props") or {}
+        extra_props: Union[FilesSourceProperties, dict] = opts.extra_props or {} if opts else {}
         handle = DropboxFS(**{**props, **extra_props})
         return handle
 

lib/galaxy/files/sources/drs.py

@@ -1,6 +1,5 @@
 import logging
-
-from typing_extensions import Unpack
+from typing import Optional
 
 from galaxy.util.drs import fetch_drs_to_file
 from . import (
@@ -25,15 +24,15 @@ class DRSFilesSource(BaseFilesSource):
         props = self._parse_common_config_opts(kwds)
         self._props = props
 
-    def _realize_to(self, source_path, native_path, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def _realize_to(self, source_path, native_path, user_context=None, opts: Optional[FilesSourceOptions] = None):
         props = self._serialization_props(user_context)
         headers = props.pop("http_headers", {}) or {}
         fetch_drs_to_file(source_path, native_path, user_context, headers=headers)
 
-    def _write_from(self, target_path, native_path, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def _write_from(self, target_path, native_path, user_context=None, opts: Optional[FilesSourceOptions] = None):
         raise NotImplementedError()
 
-    def score_url_match(self, url: str, **kwargs: Unpack[FilesSourceOptions]):
+    def score_url_match(self, url: str):
         if url.startswith("drs://"):
             return len("drs://")
         else:

lib/galaxy/files/sources/ftp.py

@@ -7,11 +7,10 @@ except ImportError:
 
 from typing import (
     cast,
+    Optional,
     Tuple,
 )
 
-from typing_extensions import Unpack
-
 from . import (
     FilesSourceOptions,
     FilesSourceProperties,
@@ -31,21 +30,35 @@ class FtpFilesSource(PyFilesystem2FilesSource):
     required_module = FTPFS
     required_package = "fs.ftpfs"
 
-    def _open_fs(self, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def _open_fs(self, user_context=None, opts: Optional[FilesSourceOptions] = None):
         props = self._serialization_props(user_context)
-        extra_props: FTPFilesSourceProperties = cast(FTPFilesSourceProperties, kwargs.get("extra_props") or {})
+        extra_props: FTPFilesSourceProperties = cast(FTPFilesSourceProperties, opts.extra_props or {} if opts else {})
         handle = FTPFS(**{**props, **extra_props})
         return handle
 
-    def _realize_to(self, source_path: str, native_path: str, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
-        extra_props: FTPFilesSourceProperties = cast(FTPFilesSourceProperties, kwargs.get("extra_props") or {})
-        path, kwargs["extra_props"] = self._get_props_and_rel_path(extra_props, source_path)
-        super()._realize_to(path, native_path, user_context=user_context, **kwargs)
+    def _realize_to(
+        self, source_path: str, native_path: str, user_context=None, opts: Optional[FilesSourceOptions] = None
+    ):
+        extra_props: FTPFilesSourceProperties
+        if opts and opts.extra_props:
+            extra_props = cast(FTPFilesSourceProperties, opts.extra_props)
+        else:
+            opts = FilesSourceOptions()
+            extra_props = {}
+        path, opts.extra_props = self._get_props_and_rel_path(extra_props, source_path)
+        super()._realize_to(path, native_path, user_context=user_context, opts=opts)
 
-    def _write_from(self, target_path: str, native_path: str, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
-        extra_props: FTPFilesSourceProperties = cast(FTPFilesSourceProperties, kwargs.get("extra_props") or {})
-        path, kwargs["extra_props"] = self._get_props_and_rel_path(extra_props, target_path)
-        super()._write_from(path, native_path, user_context=user_context, **kwargs)
+    def _write_from(
+        self, target_path: str, native_path: str, user_context=None, opts: Optional[FilesSourceOptions] = None
+    ):
+        extra_props: FTPFilesSourceProperties
+        if opts and opts.extra_props:
+            extra_props = cast(FTPFilesSourceProperties, opts.extra_props)
+        else:
+            opts = FilesSourceOptions()
+            extra_props = {}
+        path, opts.extra_props = self._get_props_and_rel_path(extra_props, target_path)
+        super()._write_from(path, native_path, user_context=user_context, opts=opts)
 
     def _get_props_and_rel_path(
         self, extra_props: FTPFilesSourceProperties, url: str

lib/galaxy/files/sources/googlecloudstorage.py

@@ -7,10 +7,9 @@ except ImportError:
 
 from typing import (
     cast,
+    Optional,
 )
 
-from typing_extensions import Unpack
-
 from . import (
     FilesSourceOptions,
     FilesSourceProperties,
@@ -30,10 +29,10 @@ class GoogleCloudStorageFilesSource(PyFilesystem2FilesSource):
     required_module = GCSFS
     required_package = "fs-gcsfs"
 
-    def _open_fs(self, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def _open_fs(self, user_context=None, opts: Optional[FilesSourceOptions] = None):
         props = self._serialization_props(user_context)
         extra_props: GoogleCloudStorageFilesSourceProperties = cast(
-            GoogleCloudStorageFilesSourceProperties, kwargs.get("extra_props") or {}
+            GoogleCloudStorageFilesSourceProperties, opts.extra_props or {} if opts else {}
         )
         bucket_name = props.pop("bucket_name", None)
         root_path = props.pop("root_path", None)

lib/galaxy/files/sources/googledrive.py

@@ -4,11 +4,9 @@ try:
 except ImportError:
     GoogleDriveFS = None
 
-from typing_extensions import Unpack
+from typing import Optional
 
-from . import (
-    FilesSourceOptions,
-)
+from . import FilesSourceOptions
 from ._pyfilesystem2 import PyFilesystem2FilesSource
 
 
@@ -17,7 +15,7 @@ class GoogleDriveFilesSource(PyFilesystem2FilesSource):
     required_module = GoogleDriveFS
     required_package = "fs.googledrivefs"
 
-    def _open_fs(self, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def _open_fs(self, user_context=None, opts: Optional[FilesSourceOptions] = None):
         props = self._serialization_props(user_context)
         credentials = Credentials(**props)
         handle = GoogleDriveFS(credentials)

lib/galaxy/files/sources/http.py

@@ -4,10 +4,9 @@ import urllib.request
 from typing import (
     cast,
     Dict,
+    Optional,
 )
 
-from typing_extensions import Unpack
-
 from galaxy.util import (
     DEFAULT_SOCKET_TIMEOUT,
     get_charset_from_http_headers,
@@ -30,23 +29,25 @@ class HTTPFilesSourceProperties(FilesSourceProperties, total=False):
 class HTTPFilesSource(BaseFilesSource):
     plugin_type = "http"
 
-    def __init__(self, label="HTTP File", doc="Default HTTP file handler", **kwd):
+    def __init__(self, **kwd):
         kwds = dict(
             id="_http",
-            label=label,
-            doc=doc,
+            label="HTTP File",
+            doc="Default HTTP file handler",
             writable=False,
         )
         kwds.update(kwd)
-        props: HTTPFilesSourceProperties = cast(HTTPFilesSourceProperties, self._parse_common_config_opts(kwd))
+        props: HTTPFilesSourceProperties = cast(HTTPFilesSourceProperties, self._parse_common_config_opts(kwds))
         self._url_regex_str = props.pop("url_regex", r"^https?://|^ftp://")
         assert self._url_regex_str
         self._url_regex = re.compile(self._url_regex_str)
         self._props = props
 
-    def _realize_to(self, source_path: str, native_path: str, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def _realize_to(
+        self, source_path: str, native_path: str, user_context=None, opts: Optional[FilesSourceOptions] = None
+    ):
         props = self._serialization_props(user_context)
-        extra_props: HTTPFilesSourceProperties = cast(HTTPFilesSourceProperties, kwargs.get("extra_props") or {})
+        extra_props: HTTPFilesSourceProperties = cast(HTTPFilesSourceProperties, opts.extra_props or {} if opts else {})
         headers = props.pop("http_headers", {}) or {}
         headers.update(extra_props.get("http_headers") or {})
 
@@ -58,7 +59,9 @@ class HTTPFilesSource(BaseFilesSource):
                 page, f.fileno(), native_path, source_encoding=get_charset_from_http_headers(page.headers)
             )
 
-    def _write_from(self, target_path: str, native_path: str, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def _write_from(
+        self, target_path: str, native_path: str, user_context=None, opts: Optional[FilesSourceOptions] = None
+    ):
         raise NotImplementedError()
 
     def _serialization_props(self, user_context=None) -> HTTPFilesSourceProperties:
@@ -68,7 +71,7 @@ class HTTPFilesSource(BaseFilesSource):
         effective_props["url_regex"] = self._url_regex_str
         return cast(HTTPFilesSourceProperties, effective_props)
 
-    def score_url_match(self, url: str, **kwargs):
+    def score_url_match(self, url: str):
         match = self._url_regex.match(url)
         if match:
             return match.span()[1]

lib/galaxy/files/sources/onedata.py

@@ -3,9 +3,10 @@ try:
 except ImportError:
     OnedataFS = None
 
-from typing import Union
-
-from typing_extensions import Unpack
+from typing import (
+    Optional,
+    Union,
+)
 
 from . import (
     FilesSourceOptions,
@@ -19,9 +20,9 @@ class OneDataFilesSource(PyFilesystem2FilesSource):
     required_module = OnedataFS
     required_package = "fs-onedatafs"
 
-    def _open_fs(self, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def _open_fs(self, user_context=None, opts: Optional[FilesSourceOptions] = None):
         props = self._serialization_props(user_context)
-        extra_props: Union[FilesSourceProperties, dict] = kwargs.get("extra_props") or {}
+        extra_props: Union[FilesSourceProperties, dict] = opts.extra_props or {} if opts else {}
         handle = OnedataFS(**{**props, **extra_props})
         return handle
 

lib/galaxy/files/sources/posix.py

@@ -5,11 +5,9 @@ from typing import (
     Any,
     Dict,
     List,
-    Union,
+    Optional,
 )
 
-from typing_extensions import Unpack
-
 from galaxy import exceptions
 from galaxy.util.path import (
     safe_contains,
@@ -51,7 +49,7 @@ class PosixFilesSource(BaseFilesSource):
         self.delete_on_realize = props.get("delete_on_realize", DEFAULT_DELETE_ON_REALIZE)
         self.allow_subdir_creation = props.get("allow_subdir_creation", DEFAULT_ALLOW_SUBDIR_CREATION)
 
-    def _list(self, path="/", recursive=True, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def _list(self, path="/", recursive=True, user_context=None, opts: Optional[FilesSourceOptions] = None):
         dir_path = self._to_native_path(path, user_context=user_context)
         if not self._safe_directory(dir_path):
             raise exceptions.ObjectNotFound(f"The specified directory does not exist [{dir_path}].")
@@ -69,7 +67,9 @@ class PosixFilesSource(BaseFilesSource):
             to_dict = functools.partial(self._resource_info_to_dict, path, user_context=user_context)
             return list(map(to_dict, res))
 
-    def _realize_to(self, source_path: str, native_path: str, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def _realize_to(
+        self, source_path: str, native_path: str, user_context=None, opts: Optional[FilesSourceOptions] = None
+    ):
         effective_root = self._effective_root(user_context)
         source_native_path = self._to_native_path(source_path, user_context=user_context)
         if self.enforce_symlink_security:
@@ -84,7 +84,9 @@ class PosixFilesSource(BaseFilesSource):
         else:
             shutil.move(source_native_path, native_path)
 
-    def _write_from(self, target_path: str, native_path: str, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def _write_from(
+        self, target_path: str, native_path: str, user_context=None, opts: Optional[FilesSourceOptions] = None
+    ):
         effective_root = self._effective_root(user_context)
         target_native_path = self._to_native_path(target_path, user_context=user_context)
         if self.enforce_symlink_security:

lib/galaxy/files/sources/s3fs.py

@@ -6,10 +6,9 @@ from typing import (
     cast,
     Dict,
     List,
+    Optional,
 )
 
-from typing_extensions import Unpack
-
 from . import (
     FilesSourceOptions,
     FilesSourceProperties,
@@ -49,8 +48,8 @@ class S3FsFilesSource(BaseFilesSource):
         if self._endpoint_url:
             self._props.update({"client_kwargs": {"endpoint_url": self._endpoint_url}})
 
-    def _list(self, path="/", recursive=True, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
-        fs = self._open_fs(user_context=user_context, **kwargs)
+    def _list(self, path="/", recursive=True, user_context=None, opts: Optional[FilesSourceOptions] = None):
+        fs = self._open_fs(user_context=user_context, opts=opts)
         if recursive:
             res: List[Dict[str, Any]] = []
             bucket_path = self._bucket_path(path)
@@ -65,11 +64,11 @@ class S3FsFilesSource(BaseFilesSource):
             to_dict = functools.partial(self._resource_info_to_dict, path)
             return list(map(to_dict, res))
 
-    def _realize_to(self, source_path, native_path, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def _realize_to(self, source_path, native_path, user_context=None, opts: Optional[FilesSourceOptions] = None):
         bucket_path = self._bucket_path(source_path)
-        self._open_fs(user_context=user_context, **kwargs).download(bucket_path, native_path)
+        self._open_fs(user_context=user_context, opts=opts).download(bucket_path, native_path)
 
-    def _write_from(self, target_path, native_path, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def _write_from(self, target_path, native_path, user_context=None, opts: Optional[FilesSourceOptions] = None):
         raise NotImplementedError()
 
     def _bucket_path(self, path: str):
@@ -79,8 +78,8 @@ class S3FsFilesSource(BaseFilesSource):
             path = f"/{path}"
         return f"{self._bucket}{path}"
 
-    def _open_fs(self, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
-        extra_props = kwargs.get("extra_props") or {}
+    def _open_fs(self, user_context=None, opts: Optional[FilesSourceOptions] = None):
+        extra_props = opts.extra_props or {} if opts else {}
         fs = s3fs.S3FileSystem(**{**self._props, **extra_props})
         return fs
 

lib/galaxy/files/sources/ssh.py

@@ -3,9 +3,10 @@ try:
 except ImportError:
     SSHFS = None
 
-from typing import Union
-
-from typing_extensions import Unpack
+from typing import (
+    Optional,
+    Union,
+)
 
 from . import (
     FilesSourceOptions,
@@ -19,9 +20,9 @@ class SshFilesSource(PyFilesystem2FilesSource):
     required_module = SSHFS
     required_package = "fs.sshfs"
 
-    def _open_fs(self, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def _open_fs(self, user_context=None, opts: Optional[FilesSourceOptions] = None):
         props = self._serialization_props(user_context)
-        extra_props: Union[FilesSourceProperties, dict] = kwargs.get("extra_props") or {}
+        extra_props: Union[FilesSourceProperties, dict] = opts.extra_props or {} if opts else {}
         path = props.pop("path")
         handle = SSHFS(**{**props, **extra_props})
         if path:

lib/galaxy/files/sources/webdav.py

@@ -3,9 +3,10 @@ try:
 except ImportError:
     WebDAVFS = None
 
-from typing import Union
-
-from typing_extensions import Unpack
+from typing import (
+    Optional,
+    Union,
+)
 
 from . import (
     FilesSourceOptions,
@@ -19,9 +20,9 @@ class WebDavFilesSource(PyFilesystem2FilesSource):
     required_module = WebDAVFS
     required_package = "fs.webdavfs"
 
-    def _open_fs(self, user_context=None, **kwargs: Unpack[FilesSourceOptions]):
+    def _open_fs(self, user_context=None, opts: Optional[FilesSourceOptions] = None):
         props = self._serialization_props(user_context)
-        extra_props: Union[FilesSourceProperties, dict] = kwargs.get("extra_props") or {}
+        extra_props: Union[FilesSourceProperties, dict] = opts.extra_props or {} if opts else {}
         handle = WebDAVFS(**{**props, **extra_props})
         return handle
 

lib/galaxy/files/uris.py

@@ -18,6 +18,7 @@ from galaxy.files import (
     ConfiguredFileSources,
     NoMatchingFileSource,
 )
+from galaxy.files.sources import FilesSourceOptions
 from galaxy.util import (
     stream_to_open_named_file,
     unicodify,
@@ -44,7 +45,7 @@ def stream_url_to_file(
     dir: Optional[str] = None,
     user_context=None,
     target_path: Optional[str] = None,
-    **kwargs: dict,
+    file_source_opts: Optional[FilesSourceOptions] = None,
 ) -> str:
     if file_sources is None:
         file_sources = ConfiguredFileSources.from_dict(None, load_stock_plugins=True)
@@ -53,7 +54,7 @@ def stream_url_to_file(
         if not target_path:
             with tempfile.NamedTemporaryFile(prefix=prefix, delete=False, dir=dir) as temp:
                 target_path = temp.name
-        file_source.realize_to(rel_path, target_path, user_context=user_context, **kwargs)
+        file_source.realize_to(rel_path, target_path, user_context=user_context, opts=file_source_opts)
         return target_path
     else:
         raise NoMatchingFileSource(f"Could not find a matching handler for: {url}")

lib/galaxy/util/drs.py

@@ -10,6 +10,8 @@ import requests
 
 from galaxy import exceptions
 from galaxy.files import FileSourceDictifiable
+from galaxy.files.sources import FilesSourceOptions
+from galaxy.files.sources.http import HTTPFilesSourceProperties
 from galaxy.files.uris import stream_url_to_file
 from galaxy.util import DEFAULT_SOCKET_TIMEOUT
 
@@ -100,16 +102,19 @@ def fetch_drs_to_file(
     downloaded = False
     for access_method in access_methods:
         access_url, access_headers = _get_access_info(get_url, access_method, headers=headers)
-        extra_props = {}
+        opts = FilesSourceOptions()
         if access_method["type"] == "https":
-            extra_props["http_headers"] = access_headers or {}
+            extra_props: HTTPFilesSourceProperties = {"http_headers": access_headers or {}}
+            opts.extra_props = extra_props
+        else:
+            opts.extra_props = {}
         try:
             stream_url_to_file(
                 access_url,
                 target_path=str(target_path),
                 file_sources=user_context.file_sources,
                 user_context=user_context,
-                extra_props=extra_props,
+                file_source_opts=opts,
             )
             downloaded = True
             break