flyte.storage

Class	Description
`ABFS`	Any Azure Blob Storage specific configuration.
`GCS`	Any GCS specific configuration.
`S3`	S3 specific configuration.
`Storage`	Data storage configuration that applies across any provider.

Method	Description
`exists()`	Check if a path exists.
`exists_sync()`
`get()`
`get_configured_fsspec_kwargs()`
`get_random_local_directory()`	pathlib.Path: a random directory.
`get_random_local_path()`	Use file_path_or_file_name, when you want a random directory, but want to preserve the leaf file name.
`get_stream()`	Get a stream of data from a remote location.
`get_underlying_filesystem()`
`is_remote()`	Let’s find a replacement.
`join()`	Join multiple paths together.
`open()`	Asynchronously open a file and return an async context manager.
`put()`	Upload a local file or directory to a remote location.
`put_stream()`	Put a stream of data to a remote location.

Methods

exists()

        
    
def exists(
    path: str,
    **kwargs,
) -> bool

Check if a path exists.

Parameter	Type	Description
`path`	`str`	Path to be checked.
`**kwargs`		Additional arguments to be passed to the underlying filesystem.

Returns: True if the path exists, False otherwise.

exists_sync()

        
    
def exists_sync(
    path: str,
    **kwargs,
) -> bool

Parameter	Type	Description
`path`	`str`
`**kwargs`

get()

        
    
def get(
    from_path: str,
    to_path: Optional[str | pathlib.Path] = None,
    recursive: bool = False,
    **kwargs,
) -> str

Parameter	Type	Description
`from_path`	`str`
`to_path`	`Optional[str \| pathlib.Path]`
`recursive`	`bool`
`**kwargs`

get_configured_fsspec_kwargs()

        
    
def get_configured_fsspec_kwargs(
    protocol: typing.Optional[str] = None,
    anonymous: bool = False,
) -> typing.Dict[str, typing.Any]

Parameter	Type	Description
`protocol`	`typing.Optional[str]`
`anonymous`	`bool`

get_random_local_path()

        
    
def get_random_local_path(
    file_path_or_file_name: pathlib.Path | str | None = None,
) -> pathlib.Path

Use file_path_or_file_name, when you want a random directory, but want to preserve the leaf file name

Parameter	Type	Description
`file_path_or_file_name`	`pathlib.Path \| str \| None`

get_stream()

        
    
def get_stream(
    path: str,
    chunk_size = 10485760,
    **kwargs,
) -> AsyncGenerator[bytes, None]

Get a stream of data from a remote location. This is useful for downloading streaming data from a remote location. Example usage:

        
    
import flyte.storage as storage
async for chunk in storage.get_stream(path="s3://my_bucket/my_file.txt"):
    process(chunk)

Parameter	Type	Description
`path`	`str`	Path to the remote location where the data will be downloaded.
`chunk_size`		Size of each chunk to be read from the file.
`**kwargs`		Additional arguments to be passed to the underlying filesystem.

Returns: An async iterator that yields chunks of bytes.

get_underlying_filesystem()

        
    
def get_underlying_filesystem(
    protocol: typing.Optional[str] = None,
    anonymous: bool = False,
    path: typing.Optional[str] = None,
    **kwargs,
) -> fsspec.AbstractFileSystem

Parameter	Type	Description
`protocol`	`typing.Optional[str]`
`anonymous`	`bool`
`path`	`typing.Optional[str]`
`**kwargs`

is_remote()

        
    
def is_remote(
    path: typing.Union[pathlib.Path | str],
) -> bool

Let’s find a replacement

Parameter	Type	Description
`path`	`typing.Union[pathlib.Path \| str]`

join()

        
    
def join(
    *paths: str,
) -> str

Join multiple paths together. This is a wrapper around os.path.join.

TODO replace with proper join with fsspec root etc

Parameter	Type	Description
`*paths`	`str`	Paths to be joined.

open()

        
    
def open(
    path: str,
    mode: str = 'rb',
    **kwargs,
) -> AsyncReadableFile | AsyncWritableFile

Asynchronously open a file and return an async context manager. This function checks if the underlying filesystem supports obstore bypass. If it does, it uses obstore to open the file. Otherwise, it falls back to the standard _open function which uses AsyncFileSystem.

It will raise NotImplementedError if neither obstore nor AsyncFileSystem is supported.

Parameter	Type	Description
`path`	`str`
`mode`	`str`
`**kwargs`

put()

        
    
def put(
    from_path: str,
    to_path: Optional[str] = None,
    recursive: bool = False,
    batch_size: Optional[int] = None,
    chunk_size: Optional[int] = None,
    max_concurrency: Optional[int] = None,
    **kwargs,
) -> str

Upload a local file or directory to a remote location.

Parameter	Type	Description
`from_path`	`str`	Local path to upload.
`to_path`	`Optional[str]`	Remote destination. If not provided, a random path under the raw data prefix is used.
`recursive`	`bool`	Upload a directory tree rather than a single file.
`batch_size`	`Optional[int]`	Number of files uploaded at once for recursive uploads. Defaults to the FLYTE_IO_BATCH_SIZE environment variable (default: 32).
`chunk_size`	`Optional[int]`	Minimum multipart part size in bytes. Defaults to the FLYTE_IO_UPLOAD_CHUNK_SIZE environment variable (default: 5 MiB). Larger files still scale the part size above this to stay under the 10,000-part limit; values below 5 MiB are raised to 5 MiB. Applies to S3, GCS and Azure; ignored by other filesystems.
`max_concurrency`	`Optional[int]`	Number of parts of a single file uploaded at once. Defaults to the FLYTE_IO_UPLOAD_MAX_CONCURRENCY environment variable (default: 12). Applies to S3, GCS and Azure; ignored by other filesystems.
`**kwargs`		Additional arguments passed to the underlying filesystem (non-obstore filesystems only).

Returns: str: The remote path the data was uploaded to.

put_stream()

        
    
def put_stream(
    data_iterable: typing.AsyncIterable[bytes] | bytes,
    name: str | None = None,
    to_path: str | None = None,
    size_hint: int | None = None,
    chunk_size: int | None = None,
    max_concurrency: int | None = None,
    **kwargs,
) -> str

Put a stream of data to a remote location. This is useful for streaming data to a remote location. Example usage:

        
    
import flyte.storage as storage
storage.put_stream(iter([b'hello']), name="my_file.txt")
OR
storage.put_stream(iter([b'hello']), to_path="s3://my_bucket/my_file.txt")

Parameter	Type	Description
`data_iterable`	`typing.AsyncIterable[bytes] \| bytes`	Iterable of bytes to be streamed.
`name`	`str \| None`	Name of the file to be created. If not provided, a random name will be generated.
`to_path`	`str \| None`	Path to the remote location where the data will be stored.
`size_hint`	`int \| None`	Total number of bytes the stream is expected to yield, if known. On obstore-backed filesystems the writer uploads one multipart part per buffer, so a fixed buffer caps a single object at ~97.6 GiB (10,000 x 10 MiB). When the hint says the stream is larger than that, the part size is scaled up to stay under the limit; smaller streams are unaffected. If FLYTE_IO_UPLOAD_CHUNK_SIZE is set above the default buffer, it is used as the part size.
`chunk_size`	`int \| None`	Minimum multipart part size in bytes, overriding FLYTE_IO_UPLOAD_CHUNK_SIZE for this call. Scaled up when `size_hint` requires it; values below 5 MiB are raised to 5 MiB. Applies to S3, GCS and Azure; ignored by other filesystems.
`max_concurrency`	`int \| None`	Number of parts uploaded at once, overriding FLYTE_IO_UPLOAD_MAX_CONCURRENCY for this call. Applies to S3, GCS and Azure; ignored by other filesystems.
`**kwargs`		Additional arguments to be passed to the underlying filesystem.

Returns: str: The path to the remote location where the data was stored.

flyte.storage

Directory

Classes

Methods

Methods

exists()

exists_sync()

get()

get_configured_fsspec_kwargs()

get_random_local_directory()

get_random_local_path()

get_stream()

get_underlying_filesystem()

is_remote()

join()

TODO replace with proper join with fsspec root etc

open()

put()

put_stream()