ancient-wolf-19325
07/26/2024, 9:31 AMFlyteDirectory objects, you can create a task that takes both directories as inputs, copies their contents into a new directory, and returns the new directory as a FlyteDirectory. Here's an example:
import os
import shutil
from pathlib import Path
from flytekit import task, workflow
from flytekit.types.directory import FlyteDirectory
@task
def merge_directories(dir1: FlyteDirectory, dir2: FlyteDirectory) -> FlyteDirectory:
merged_dir = Path("/tmp/merged_directory")
merged_dir.mkdir(parents=True, exist_ok=True)
# Copy contents of dir1
for item in os.listdir(dir1.path):
s = os.path.join(dir1.path, item)
d = os.path.join(merged_dir, item)
if os.path.isdir(s):
shutil.copytree(s, d, dirs_exist_ok=True)
else:
shutil.copy2(s, d)
# Copy contents of dir2
for item in os.listdir(dir2.path):
s = os.path.join(dir2.path, item)
d = os.path.join(merged_dir, item)
if os.path.isdir(s):
shutil.copytree(s, d, dirs_exist_ok=True)
else:
shutil.copy2(s, d)
return FlyteDirectory(path=str(merged_dir))
@workflow
def merge_directories_workflow(dir1: FlyteDirectory, dir2: FlyteDirectory) -> FlyteDirectory:
return merge_directories(dir1=dir1, dir2=dir2)
This code defines a merge_directories task that merges the contents of two FlyteDirectory objects and a merge_directories_workflow workflow that uses this task.
Would you like more details on any specific part of this process?
Sources:
- https://discuss.flyte.org/t/10062332/i-am-trying-something-a-bit-unconventional-given-that-severa/
- https://discuss.flyte.org/t/16753623/what-is-the-best-way-for-a-task-to-pass-a-bunch-of-images-to/
Note: I wasn’t able to find highly relevant data sources, but above are a few potentially relevant links.
Slack Link
https://flyte-org.slack.com/archives/CP2HDHKE1/p1721986262.795649 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.