Source code for runhouse.resources.folders.folder_factory

from pathlib import Path
from typing import Optional, Union

from runhouse.logger import get_logger

from runhouse.resources.folders.folder import Folder
from runhouse.resources.hardware.utils import _get_cluster_from

logger = get_logger(__name__)


[docs]def folder( name: Optional[str] = None, path: Optional[Union[str, Path]] = None, system: Optional[Union[str, "Cluster"]] = None, load_from_den: bool = True, dryrun: bool = False, ) -> Folder: """Creates a Runhouse folder object, which can be used to interact with the folder at the given path. Args: name (Optional[str]): Name to give the folder, to be re-used later on. path (Optional[str or Path]): Path (or path) that the folder is located at. system (Optional[str or Cluster]): File system or cluster name. If prpre-oviding a file system this must be one of: [``file``, ``s3``, ``gs``]. load_from_den (bool): Whether to try loading the Folder resource from Den. (Default: ``True``) dryrun (bool): Whether to create the Folder if it doesn't exist, or load a Folder object as a dryrun. (Default: ``False``) Returns: Folder: The resulting folder. Example: >>> import runhouse as rh >>> rh.folder(name='training_imgs', path='remote_directory/images', system='s3').save() >>> # Load folder from above >>> reloaded_folder = rh.folder(name="training_imgs") """ # TODO [DG] Include loud warning that relative paths are relative to the git root / working directory! if name and not any([path, system]): # If only the name is provided try: return Folder.from_name(name, load_from_den=load_from_den, dryrun=dryrun) except ValueError: # This is a rare instance where passing no constructor params is actually valid (anonymous folder), # so if we don't find the name, the user may still actually want to create a new folder. pass system = system or Folder.DEFAULT_FS if system == "s3": from .s3_folder import S3Folder logger.debug(f"Creating a S3 folder with name: {name}") return S3Folder( system=system, path=path, name=name, dryrun=dryrun, ) elif system == "gs": from .gcs_folder import GCSFolder logger.debug(f"Creating a GS folder with name: {name}") return GCSFolder( system=system, path=path, name=name, dryrun=dryrun, ) if system == Folder.DEFAULT_FS: logger.debug(f"Creating local folder with name: {name}") return Folder( system=system, path=path, name=name, dryrun=dryrun, ) from runhouse import Cluster cluster_system = _get_cluster_from(system, dryrun=dryrun) if isinstance(cluster_system, Cluster): logger.debug(f"Creating folder {name} for cluster: {cluster_system.name}") return Folder( system=cluster_system, path=path, name=name, dryrun=dryrun, ) raise ValueError( f"System '{system}' not currently supported. If the file system " f"is a cluster (ex: '/my-user/rh-cpu'), make sure the cluster config has been saved." )