From f34c962ab347de5f88aee60ae9a3b1403ad5073f Mon Sep 17 00:00:00 2001 From: machichima Date: Thu, 14 Aug 2025 17:39:22 +0800 Subject: [PATCH 1/5] fix: let downloader use current context -e Signed-off-by: machichima --- flytekit/types/file/file.py | 18 ++++++++++++------ 1 file changed, 12 insertions(+), 6 deletions(-) diff --git a/flytekit/types/file/file.py b/flytekit/types/file/file.py index 780188f9e5..39f4cbe5e8 100644 --- a/flytekit/types/file/file.py +++ b/flytekit/types/file/file.py @@ -41,17 +41,19 @@ def noop(): ... class _FlyteFileDownloader: """Downloader for FlyteFile that uses current context when called.""" - + def __init__(self, remote_path: str, local_path: str, is_multipart: bool = False): self.remote_path = remote_path - self.local_path = local_path + self.local_path = local_path self.is_multipart = is_multipart - + def __call__(self): """Download the file using current context's synced get_data method.""" current_ctx = FlyteContextManager.current_context() return current_ctx.file_access.get_data( - remote_path=self.remote_path, local_path=self.local_path, is_multipart=self.is_multipart + remote_path=self.remote_path, + local_path=self.local_path, + is_multipart=self.is_multipart ) @@ -335,7 +337,7 @@ def __init__( self._local_path = ctx.file_access.get_random_local_path(self._remote_source) self._downloader = _FlyteFileDownloader( remote_path=str(self._remote_source), # type: ignore - local_path=str(self._local_path), + local_path=self._local_path, ) def __fspath__(self): @@ -768,7 +770,11 @@ async def async_to_python_value( # For the remote case, return an FlyteFile object that can download local_path = ctx.file_access.get_random_local_path(uri) - _downloader = _FlyteFileDownloader(remote_path=uri, local_path=local_path, is_multipart=False) + _downloader = _FlyteFileDownloader( + remote_path=uri, + local_path=local_path, + is_multipart=False + ) expected_format = FlyteFilePathTransformer.get_format(expected_python_type) ff = FlyteFile.__class_getitem__(expected_format)(path=local_path, downloader=_downloader, metadata=metadata) From e52f9965d8c8149d5ad34da8462bfe83a564967a Mon Sep 17 00:00:00 2001 From: machichima Date: Fri, 15 Aug 2025 16:01:07 +0800 Subject: [PATCH 2/5] refactor: fix lint -e Signed-off-by: machichima --- flytekit/types/file/file.py | 18 ++++++------------ 1 file changed, 6 insertions(+), 12 deletions(-) diff --git a/flytekit/types/file/file.py b/flytekit/types/file/file.py index 39f4cbe5e8..780188f9e5 100644 --- a/flytekit/types/file/file.py +++ b/flytekit/types/file/file.py @@ -41,19 +41,17 @@ def noop(): ... class _FlyteFileDownloader: """Downloader for FlyteFile that uses current context when called.""" - + def __init__(self, remote_path: str, local_path: str, is_multipart: bool = False): self.remote_path = remote_path - self.local_path = local_path + self.local_path = local_path self.is_multipart = is_multipart - + def __call__(self): """Download the file using current context's synced get_data method.""" current_ctx = FlyteContextManager.current_context() return current_ctx.file_access.get_data( - remote_path=self.remote_path, - local_path=self.local_path, - is_multipart=self.is_multipart + remote_path=self.remote_path, local_path=self.local_path, is_multipart=self.is_multipart ) @@ -337,7 +335,7 @@ def __init__( self._local_path = ctx.file_access.get_random_local_path(self._remote_source) self._downloader = _FlyteFileDownloader( remote_path=str(self._remote_source), # type: ignore - local_path=self._local_path, + local_path=str(self._local_path), ) def __fspath__(self): @@ -770,11 +768,7 @@ async def async_to_python_value( # For the remote case, return an FlyteFile object that can download local_path = ctx.file_access.get_random_local_path(uri) - _downloader = _FlyteFileDownloader( - remote_path=uri, - local_path=local_path, - is_multipart=False - ) + _downloader = _FlyteFileDownloader(remote_path=uri, local_path=local_path, is_multipart=False) expected_format = FlyteFilePathTransformer.get_format(expected_python_type) ff = FlyteFile.__class_getitem__(expected_format)(path=local_path, downloader=_downloader, metadata=metadata) From 1dd019014ae7ada54c2be3668b28e14c4dd4351d Mon Sep 17 00:00:00 2001 From: machichima Date: Fri, 22 Aug 2025 11:01:00 +0800 Subject: [PATCH 3/5] fix: flyte dir downloader use current context -e Signed-off-by: machichima --- flytekit/types/directory/types.py | 25 +++++++++++++++++++++++-- 1 file changed, 23 insertions(+), 2 deletions(-) diff --git a/flytekit/types/directory/types.py b/flytekit/types/directory/types.py index 410ed5b071..ae93a7665f 100644 --- a/flytekit/types/directory/types.py +++ b/flytekit/types/directory/types.py @@ -49,6 +49,23 @@ def noop(): ... +class _FlyteDirectoryDownloader: + """Downloader for FlyteDirectory that uses current context when called.""" + + def __init__(self, remote_path: str, local_path: str, is_multipart: bool = True, batch_size: typing.Optional[int] = None): + self.remote_path = remote_path + self.local_path = local_path + self.is_multipart = is_multipart + self.batch_size = batch_size + + def __call__(self): + """Download the directory using current context's synced get_data method.""" + current_ctx = FlyteContextManager.current_context() + return current_ctx.file_access.get_data( + remote_path=self.remote_path, local_path=self.local_path, is_multipart=self.is_multipart, batch_size=self.batch_size + ) + + @dataclass class FlyteDirectory(SerializableType, DataClassJsonMixin, os.PathLike, typing.Generic[T]): path: PathType = field(default=None, metadata=config(mm_field=fields.String())) # type: ignore @@ -409,7 +426,9 @@ def listdir(cls, directory: FlyteDirectory) -> typing.List[typing.Union[FlyteDir paths.append(flyte_file) else: local_folder = file_access.get_random_local_directory() - downloader = partial(file_access.get_data, remote_path, local_folder, is_multipart=True) + downloader = _FlyteDirectoryDownloader( + remote_path=remote_path, local_path=local_folder, is_multipart=True + ) flyte_directory: FlyteDirectory = FlyteDirectory(path=local_folder, downloader=downloader) flyte_directory._remote_source = remote_path @@ -684,7 +703,9 @@ async def async_to_python_value( batch_size = get_batch_size(expected_python_type) - _downloader = partial(ctx.file_access.get_data, uri, local_folder, is_multipart=True, batch_size=batch_size) + _downloader = _FlyteDirectoryDownloader( + remote_path=uri, local_path=local_folder, is_multipart=True, batch_size=batch_size + ) expected_format = self.get_format(expected_python_type) From 510c627ebd595bbce3dafc9122baa2cd1430052b Mon Sep 17 00:00:00 2001 From: machichima Date: Fri, 22 Aug 2025 11:29:05 +0800 Subject: [PATCH 4/5] refactor: lint -e Signed-off-by: machichima --- flytekit/types/directory/types.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/flytekit/types/directory/types.py b/flytekit/types/directory/types.py index ae93a7665f..4e77341d66 100644 --- a/flytekit/types/directory/types.py +++ b/flytekit/types/directory/types.py @@ -52,7 +52,9 @@ def noop(): ... class _FlyteDirectoryDownloader: """Downloader for FlyteDirectory that uses current context when called.""" - def __init__(self, remote_path: str, local_path: str, is_multipart: bool = True, batch_size: typing.Optional[int] = None): + def __init__( + self, remote_path: str, local_path: str, is_multipart: bool = True, batch_size: typing.Optional[int] = None + ): self.remote_path = remote_path self.local_path = local_path self.is_multipart = is_multipart @@ -62,7 +64,10 @@ def __call__(self): """Download the directory using current context's synced get_data method.""" current_ctx = FlyteContextManager.current_context() return current_ctx.file_access.get_data( - remote_path=self.remote_path, local_path=self.local_path, is_multipart=self.is_multipart, batch_size=self.batch_size + remote_path=self.remote_path, + local_path=self.local_path, + is_multipart=self.is_multipart, + batch_size=self.batch_size, ) From f7791be2f2791750a037b0dcca462f13627e9ba6 Mon Sep 17 00:00:00 2001 From: machichima Date: Sat, 23 Aug 2025 08:05:59 +0800 Subject: [PATCH 5/5] refactor: fix lint error -e Signed-off-by: machichima --- flytekit/types/directory/types.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/flytekit/types/directory/types.py b/flytekit/types/directory/types.py index 4e77341d66..286cfafbf6 100644 --- a/flytekit/types/directory/types.py +++ b/flytekit/types/directory/types.py @@ -431,11 +431,11 @@ def listdir(cls, directory: FlyteDirectory) -> typing.List[typing.Union[FlyteDir paths.append(flyte_file) else: local_folder = file_access.get_random_local_directory() - downloader = _FlyteDirectoryDownloader( + dir_downloader: typing.Callable = _FlyteDirectoryDownloader( remote_path=remote_path, local_path=local_folder, is_multipart=True ) - flyte_directory: FlyteDirectory = FlyteDirectory(path=local_folder, downloader=downloader) + flyte_directory: FlyteDirectory = FlyteDirectory(path=local_folder, downloader=dir_downloader) flyte_directory._remote_source = remote_path paths.append(flyte_directory)