anemoi.utils.remote package

anemoi.utils.remote.robust(call: callable, *args, maximum_tries: int = 60, retry_after: int = 60, **kwargs) → callable: Forwards the arguments to the multiurl robust function. with default retry_after=60 and maximum_tries=60.

class anemoi.utils.remote.Loader

Bases: object

transfer_folder(*, source: str, target: str, overwrite: bool = False, resume: bool = False, verbosity: int = 1, threads: int = 1, progress: callable = None) → None

Transfer a folder from the source to the target location.

Parameters:

source (str) – The source folder path.
target (str) – The target folder path.
overwrite (bool, optional) – Whether to overwrite the target if it exists, by default False.
resume (bool, optional) – Whether to resume the transfer if possible, by default False.
verbosity (int, optional) – The verbosity level, by default 1.
threads (int, optional) – The number of threads to use, by default 1.
progress (callable, optional) – A callable for progress reporting, by default None.

transfer_file(source: str, target: str, overwrite: bool, resume: bool, verbosity: int, threads: int = 1, progress: callable = None, config: dict = None) → int

Transfer a file from the source to the target location.

Parameters:

source (str) – The source file path.
target (str) – The target file path.
overwrite (bool) – Whether to overwrite the target if it exists.
resume (bool) – Whether to resume the transfer if possible.
verbosity (int) – The verbosity level.
threads (int, optional) – The number of threads to use, by default 1.
progress (callable, optional) – A callable for progress reporting, by default None.
config (dict, optional) – Additional configuration options, by default None.

Returns:

The size of the transferred file.

Return type:

int

Raises:

Exception – If an error occurs during the transfer.

abstractmethod list_source(source: str) → Iterable

List the files in the source location.

Parameters:: source (str) – The source location.
Returns:: An iterable of files in the source location.
Return type:: Iterable

abstractmethod source_path(local_path: str, source: str) → str

Get the source path for a local file.

Parameters:

local_path (str) – The local file path.
source (str) – The source location.

Returns:

The source path for the local file.

Return type:

str

abstractmethod target_path(source_path: str, source: str, target: str) → str

Get the target path for a source file.

Parameters:

source_path (str) – The source file path.
source (str) – The source location.
target (str) – The target location.

Returns:

The target path for the source file.

Return type:

str

abstractmethod source_size(local_path: str) → int

Get the size of a local file.

Parameters:: local_path (str) – The local file path.
Returns:: The size of the local file.
Return type:: int

abstractmethod copy(source: str, target: str, **kwargs) → None

Copy a file or folder from the source to the target location.

Parameters:

source (str) – The source location.
target (str) – The target location.
kwargs (dict) – Additional arguments for the transfer.

abstractmethod get_temporary_target(target: str, pattern: str) → str

Get a temporary target path based on the given pattern.

Parameters:

target (str) – The original target path.
pattern (str) – The pattern to format the temporary path.

Returns:

The temporary target path.

Return type:

str

abstractmethod rename_target(target: str, temporary_target: str) → None

Rename the target to a new target path.

Parameters:

target (str) – The original target path.
temporary_target (str) – The new target path.

class anemoi.utils.remote.BaseDownload

Bases: Loader

action = 'Downloading'

abstractmethod copy(source: str, target: str, **kwargs) → None

Copy a file or folder from the source to the target location.

Parameters:

source (str) – The source location.
target (str) – The target location.
kwargs (dict) – Additional arguments for the transfer.

get_temporary_target(target: str, pattern: str) → str

Get a temporary target path based on the given pattern.

Parameters:

target (str) – The original target path.
pattern (str) – The pattern to format the temporary path.

Returns:

The temporary target path.

Return type:

str

rename_target(target: str, new_target: str) → None

Rename the target to a new target path.

Parameters:

target (str) – The original target path.
new_target (str) – The new target path.

delete_target(target: str) → None

Delete the target if it exists.

Parameters:: target (str) – The target path.

class anemoi.utils.remote.BaseUpload

Bases: Loader

action = 'Uploading'

copy(source: str, target: str, **kwargs) → None

Copy a file or folder from the source to the target location.

Parameters:

source (str) – The source location.
target (str) – The target location.
kwargs (dict) – Additional arguments for the transfer.

list_source(source: str) → Iterable

List the files in the source location.

Parameters:: source (str) – The source location.
Returns:: An iterable of files in the source location.
Return type:: Iterable

source_path(local_path: str, source: str) → str

Get the source path for a local file.

Parameters:

local_path (str) – The local file path.
source (str) – The source location.

Returns:

The source path for the local file.

Return type:

str

target_path(source_path: str, source: str, target: str) → str

Get the target path for a source file.

Parameters:

source_path (str) – The source file path.
source (str) – The source location.
target (str) – The target location.

Returns:

The target path for the source file.

Return type:

str

source_size(local_path: str) → int

Get the size of a local file.

Parameters:: local_path (str) – The local file path.
Returns:: The size of the local file.
Return type:: int

exception anemoi.utils.remote.TransferMethodNotImplementedError: Bases: NotImplementedError

class anemoi.utils.remote.Transfer(*, source: str, target: str, overwrite: bool = False, resume: bool = False, verbosity: int = 1, threads: int = 1, progress: callable = None, temporary_target: bool = False)

Bases: object

This is the internal API and should not be used directly. Use the transfer function instead.

exception TransferMethodNotImplementedError: Bases: NotImplementedError

run() → Transfer

Execute the transfer process.

Returns:: The Transfer instance.
Return type:: Transfer

rename_target(target: str, new_target: str) → None

Rename the target to a new target path.

Parameters:

target (str) – The original target path.
new_target (str) – The new target path.

delete_target(target: str) → None

Delete the target if it exists.

Parameters:: target (str) – The target path.

anemoi.utils.remote.transfer(source, target, *, overwrite=False, resume=False, verbosity=1, progress=None, threads=1, temporary_target=False) → Loader

Transfer files or folders from the source to the target location.

Parameters:

source (str) – A path to a local file or folder or a URL to a file or a folder on S3 or Azure Blob Storage. The url should start with ‘s3://’ or ‘abfs://’, respectively.
target (str) – A path to a local file or folder or a URL to a file or a folder on S3, Azure Blob Storage, or a remote folder. The url should start with ‘s3://’, ‘abfs://’, or ‘ssh://’, respectively.
overwrite (bool, optional) – If the data is already on in the target location it will be overwritten. By default False
resume (bool, optional) – If the data is already on S3 or Azure Blob Storage it will not be uploaded, unless the remote file has a different size. Ignored if the target is an SSH remote folder (ssh://). By default False
verbosity (int, optional) – The level of verbosity, by default 1
progress (callable, optional) – A callable that will be called with the number of files, the total size of the files, the total size transferred and a boolean indicating if the transfer has started. By default None
threads (int, optional) – The number of threads to use when uploading a directory, by default 1
temporary_target (bool, optional) – Experimental feature If True and if the target location supports it, the data will be uploaded to a temporary location then renamed to the final location. Supported by SSH and local targets, not supported by S3 or Azure Blob Storage. By default False.

Returns:

The Loader instance.

Return type:

Loader

Submodules

anemoi.utils.remote.az module

Management of objects in Azure Blob Storage.

Provides functions to upload, download, list, and delete files and folders, including hierarchical namespace (Azure Data Lake Storage Gen 2).

Only abfs:// scheme URLs are accepted in one of two shapes, including optional query strings for SAS tokens (e.g. suffixed ?sv=…):

abfs://<container>/<path>: account name comes from settings or environment variables
abfs://<container>@<account>.dfs.core.windows.net/<path>?: account name in the URL (.blob.core.windows.net also

accepted)

Credentials can be supplied via ~/.config/anemoi/settings.toml (or settings.secrets.toml) under the object-storage section:

```toml

[object-storage] endpoint_url = “…” # for Azure, only needed for sovereign clouds / Azurite account_name = “…” # storage account name account_key = “…” # storage account key sas_token = “?sv=…” # SAS token instead of an account key skip_signature = true # for public / anonymous containers

# container-keyed overrides, if source and destination are in different accounts or have different credentials [“object-storage.training-data”] account_name = “…” sas_token = “?sv=…”

```

Alternatively, obstore attempts to read standard Azure environment variables directly if no credentials are configured as above. For a list of supported environment variables, see [obstore documentation](https://developmentseed.org/obstore/latest/api/store/azure/#obstore.store.AzureConfig).

class anemoi.utils.remote.az.AzureObject(url: str)

Bases: object

Parsed representation of an Azure Blob Storage URL.

Return a cached obstore client for the container referenced by obj.

The client is keyed by account and container so that requests to different accounts sharing a container name do not collide in the cache. Credential precedence, highest to lowest:

SAS token in the URL query
SAS token in settings
Account key in settings
Anonymous access (skip_signature).

If none are set, obstore falls back to the standard AZURE_STORAGE_* environment variables.

Parameters:: obj (str | AzureObject) – Azure Blob Storage URL or a pre-parsed AzureObject.
Returns:: Ready-to-use client for the account and container referenced by obj.
Return type:: obstore.store.ObjectStore

anemoi.utils.remote.az.list_folder(folder: str) → Iterable[ObjectMeta]

List blobs under the URL prefix.

Parameters:: folder (str) – Azure Blob Storage URL used as a prefix.
Returns:: One item per blob under the prefix.
Return type:: Iterable[ObjectMeta]

anemoi.utils.remote.az.delete_folder(target: str) → None

Delete all blobs under the URL prefix.

Parameters:: target (str) – Azure Blob Storage URL used as a prefix.

anemoi.utils.remote.az.delete_file(target: str) → None

Delete a single blob.

If no blob exists at the URL, a warning is logged and no error is raised.

Parameters:: target (str) – Azure Blob Storage URL for the blob to delete.

anemoi.utils.remote.az.delete(target: str) → None

Delete a single blob or all blobs under a prefix.

Dispatches to delete_folder if target ends with ‘/’, otherwise delete_file.

Parameters:: target (str) – Azure Blob Storage URL. Treated as a prefix if it ends with ‘/’, otherwise as a single blob.

anemoi.utils.remote.az.object_info(target: str | AzureObject) → ObjectMeta

Fetch metadata for a single blob.

Parameters:: target (str | AzureObject) – Azure Blob Storage URL or a pre-parsed AzureObject.
Returns:: Metadata for the blob, including size, path, and last-modified time.
Return type:: ObjectMeta
Raises:: FileNotFoundError – If no blob exists at the URL.

anemoi.utils.remote.az.object_exists(target: str | AzureObject) → bool

Check whether a blob exists at the URL.

Parameters:: target (str | AzureObject) – Azure Blob Storage URL or a pre-parsed AzureObject.
Returns:: True if the blob exists, False otherwise.
Return type:: bool

anemoi.utils.remote.az.get_object(target: str) → Bytes

Fetch a blob and return its contents.

Parameters:: target (str) – Azure Blob Storage URL.
Returns:: The blob’s contents.
Return type:: obstore.Bytes

anemoi.utils.remote.az.get_objects_parallel(targets: list[str]) → list[Bytes]

Fetch multiple blobs concurrently.

Each fetch is retried up to three times on errors other than FileNotFoundError, which is raised immediately without retry.

Parameters:

targets (list[str]) – Azure Blob Storage URLs to fetch.

Returns:

Blob contents, in the same order as targets.

Return type:

list[obstore.Bytes]

Raises:

FileNotFoundError – If any blob does not exist.
OSError – If any fetch fails three times consecutively for reasons other than the blob being missing.

anemoi.utils.remote.az.upload(source: str, target: str, *args: bool | int, **kwargs: bool | int) → Loader

Upload a local file or directory to Azure Blob Storage via transfer.

Parameters:

source (str) – Local file or directory path.
target (str) – Azure Blob Storage URL for the destination. Ends with ‘/’ for a directory upload.
*args (bool | int) – Additional positional arguments passed through to transfer.
**kwargs (bool | int) – Additional keyword arguments passed through to transfer.

Returns:

Loader that performs and reports on the transfer when iterated.

Return type:

Loader

anemoi.utils.remote.az.upload_file(source: str, target: str, overwrite: bool, resume: bool, verbosity: int) → int

Upload a single local file to Azure Blob Storage.

Parameters:

source (str) – Local file path to upload.
target (str) – Azure Blob Storage URL for the destination blob.
overwrite (bool) – If True, replace any existing blob at the destination.
resume (bool) – If True, skip the upload when a blob of the expected size already exists at the destination.
verbosity (int) – Verbosity level: 0 is silent, higher values show more progress.

Returns:

Number of bytes transferred.

Return type:

int

Raises:

ValueError – If a blob already exists at the destination and neither overwrite nor resume is set.

class anemoi.utils.remote.az.AzureUpload

Bases: BaseUpload

Transfer implementation for uploading local files and directories to Azure Blob Storage.

get_temporary_target(target: str, _: str) → str

Return the final target unchanged.

Azure Blob Storage uploads write directly to the destination.

Parameters:

target (str) – Azure Blob Storage URL for the destination.
_ (str) – Ignored, kept for signature compatibility with the base class.

Returns:

The target URL, unchanged.

Return type:

str

rename_target(target: str, temporary_target: str) → None

No-op, Azure Blob Storage uploads write directly to the destination.

Parameters:

target (str) – Final Azure Blob Storage URL. Unused.
temporary_target (str) – Temporary Azure Blob Storage URL. Unused.

delete_target(target: str) → None

No-op, Azure Blob Storage uploads write directly to the destination.

Parameters:: target (str) – Azure Blob Storage URL. Unused.

anemoi.utils.remote.az.download(source: str, target: str, *args: bool | int, **kwargs: bool | int) → Loader

Download a blob or prefix from Azure Blob Storage to a local path via transfer.

Parameters:

source (str) – Azure Blob Storage URL. Ends with ‘/’ for a prefix download.
target (str) – Local file or directory path.
*args (bool | int) – Additional positional arguments passed through to transfer.
**kwargs (bool | int) – Additional keyword arguments passed through to transfer.

Returns:

Loader that performs and reports on the transfer when iterated.

Return type:

Loader

anemoi.utils.remote.az.download_file(source: str, target: str, overwrite: bool, resume: bool, verbosity: int) → int

Download a single blob to a local file path.

The download is retried up to three times on errors other than FileNotFoundError, which is raised immediately without retry.

Parameters:

source (str) – Azure Blob Storage URL for the blob to download.
target (str) – Local file path to write.
overwrite (bool) – If True, replace any existing file at the target path.
resume (bool) – If True, skip the download when a file of the expected size already exists at the target.
verbosity (int) – Verbosity level: 0 is silent, higher values show more progress.

Returns:

Number of bytes transferred.

Return type:

int

Raises:

ValueError – If a file already exists at the target and neither overwrite nor resume is set.
FileNotFoundError – If no blob exists at the source URL.
OSError – If the download fails three times consecutively for reasons other than the blob being missing.

class anemoi.utils.remote.az.AzureDownload

Bases: BaseDownload

Transfer implementation for downloading blobs and prefixes from Azure Blob Storage.

copy(source: str, target: str, **kwargs: Any) → None

Copy a single blob or all blobs under a prefix to a local path.

Dispatches to transfer_folder if source ends with ‘/’, otherwise transfer_file.

Parameters:

source (str) – Azure Blob Storage URL. Treated as a prefix if it ends with ‘/’, otherwise as a single blob.
target (str) – Local file or directory path.
**kwargs – Additional keyword arguments forwarded to the underlying transfer.

list_source(source: str) → Iterable[ObjectMeta]

List blobs under the source URL prefix.

Parameters:: source (str) – Azure Blob Storage URL used as a prefix.
Yields:: ObjectMeta – One item per blob under the prefix.

source_path(az_object: ObjectMeta, source: str) → str

Build the Azure Blob Storage URL for a single blob within source.

Parameters:

az_object (ObjectMeta) – Metadata for a single blob returned by list_source.
source (str) – Azure Blob Storage URL used as a prefix.

Returns:

Full abfs:// URL for the blob, in container/path form.

Return type:

str

target_path(az_object: ObjectMeta, source: str, target: str) → str

Build the local file path for a single blob, creating parent directories as needed.

Parameters:

az_object (ObjectMeta) – Metadata for a single blob returned by list_source.
source (str) – Azure Blob Storage URL used as a prefix.
target (str) – Local directory into which the blob should be written.

Returns:

Local path at which the blob will be written.

Return type:

str

source_size(az_object: ObjectMeta) → int

Return the size in bytes of a single blob.

Parameters:: az_object (ObjectMeta) – Metadata for a single blob returned by list_source.
Returns:: Size of the blob in bytes.
Return type:: int

anemoi.utils.remote.s3 module

This module provides functions to upload, download, list and delete files and folders on S3. The functions of this package expect that the AWS credentials are set up in the environment typicaly by setting the AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY environment variables or by creating a ~/.aws/credentials file. It is also possible to set the endpoint_url in the same file to use a different S3 compatible service:

[default]
endpoint_url = https://some-storage.somewhere.world
aws_access_key_id = xxxxxxxxxxxxxxxxxxxxxxxx
aws_secret_access_key = xxxxxxxxxxxxxxxxxxxxxxxx

Alternatively, the endpoint_url, and keys can be set in one of the ~/.config/anemoi/settings.toml or ~/.config/anemoi/settings.secrets.toml files.

class anemoi.utils.remote.s3.S3Object(url: str): Bases: object

anemoi.utils.remote.s3.s3_client(obj: str | S3Object) → Any

Return a cached S3 client for the given URL.

Parameters:: obj (str or S3Object) – S3 URL or S3Object instance.
Returns:: S3 client instance.
Return type:: Any

anemoi.utils.remote.s3.upload_file(source: str, target: str, overwrite: bool, resume: bool, verbosity: int) → int

Upload a file to S3.

Parameters:

source (str) – Local file path to upload.
target (str) – S3 target URL.
overwrite (bool) – Overwrite existing file if True.
resume (bool) – Resume upload if True.
verbosity (int) – Verbosity level.

Returns:

Number of bytes uploaded.

Return type:

int

anemoi.utils.remote.s3.download_file(source: str, target: str, overwrite: bool, resume: bool, verbosity: int) → int

Download a file from S3.

Parameters:

source (str) – S3 source URL.
target (str) – Local file path to save.
overwrite (bool) – Overwrite existing file if True.
resume (bool) – Resume download if True.
verbosity (int) – Verbosity level.

Returns:

Number of bytes downloaded.

Return type:

int

anemoi.utils.remote.s3.delete_folder(target: str) → None

anemoi.utils.remote.s3.delete_file(target: str) → None

anemoi.utils.remote.s3.delete(target: str) → None

Delete a file or folder from S3.

Parameters:: target (str) – S3 URL (file or folder).

anemoi.utils.remote.s3.list_folder(folder: str) → Iterable[dict]

List objects in an S3 folder.

Parameters:: folder (str) – S3 folder URL.
Returns:: Iterable of objects.
Return type:: Iterable

anemoi.utils.remote.s3.object_info(target: str) → dict

Get information about an S3 object.

Parameters:: target (str) – S3 object URL.
Returns:: Object metadata.
Return type:: dict

anemoi.utils.remote.s3.object_exists(target: str) → bool

Check if an S3 object exists.

Parameters:: target (str) – S3 object URL.
Returns:: True if object exists, False otherwise.
Return type:: bool

anemoi.utils.remote.s3.get_object(target: str) → bytes

Fetch an S3 object and return its contents as bytes.

Parameters:: target (str) – S3 object URL.
Returns:: Object contents.
Return type:: bytes

anemoi.utils.remote.s3.get_objects_parallel(targets: list[str]) → list[bytes]

Fetch multiple S3 objects concurrently and return their contents.

Parameters:: targets (list[str]) – List of S3 URLs to fetch in parallel.
Returns:: Object contents in the same order as targets.
Return type:: list[bytes]

anemoi.utils.remote.s3.download(source: str, target: str, *args, **kwargs) → None

Download from S3 using transfer utility.

Parameters:

source (str) – S3 source URL.
target (str) – Local target path.
*args – Additional arguments.
**kwargs – Additional keyword arguments.

anemoi.utils.remote.s3.upload(source: str, target: str, *args, **kwargs) → None

Upload to S3 using transfer utility.

Parameters:

source (str) – Local source path.
target (str) – S3 target URL.
*args – Additional arguments.
**kwargs – Additional keyword arguments.

class anemoi.utils.remote.s3.S3Upload

Bases: BaseUpload

get_temporary_target(target: str, pattern: str) → str

Get temporary target path for upload.

Parameters:

target (str) – S3 target URL.
pattern (str) – Pattern for temporary naming.

Returns:

Temporary target path.

Return type:

str

rename_target(target: str, temporary_target: str) → None

Rename temporary target to final target.

Parameters:

target (str) – Final target path.
temporary_target (str) – Temporary target path.

delete_target(target: str) → None

Delete target from S3.

Parameters:: target (str) – S3 target URL.

class anemoi.utils.remote.s3.S3Download

Bases: BaseDownload

copy(source: str, target: str, **kwargs) → None

Copy file or folder from S3.

Parameters:

source (str) – S3 source URL.
target (str) – Local target path.
**kwargs – Additional keyword arguments.

list_source(source: str) → Iterable[dict]

List objects in S3 source folder.

Parameters:: source (str) – S3 source folder URL.
Returns:: Iterable of objects.
Return type:: Iterable

source_path(s3_object: dict, source: str) → str

Get S3 path for a source object.

Parameters:

s3_object (dict) – S3 object metadata.
source (str) – S3 source folder URL.

Returns:

S3 object path.

Return type:

str

target_path(s3_object: dict, source: str, target: str) → str

Get local target path for an S3 object.

Parameters:

s3_object (dict) – S3 object metadata.
source (str) – S3 source folder URL.
target (str) – Local target folder.

Returns:

Local target path.

Return type:

str

source_size(s3_object: dict) → int

Get size of S3 object.

Parameters:: s3_object (dict) – S3 object metadata.
Returns:: Size in bytes.
Return type:: int

anemoi.utils.remote.ssh module

anemoi.utils.remote.ssh.call_process(*args: str) → str

Execute a subprocess with the given arguments and return its output.

Parameters:: args (str) – The command and its arguments to execute.
Returns:: The standard output of the command.
Return type:: str
Raises:: RuntimeError – If the command returns a non-zero exit code.

class anemoi.utils.remote.ssh.SshBaseUpload

Bases: BaseUpload

get_temporary_target(target: str, pattern: str) → str

Get a temporary target path based on the given pattern.

Parameters:

target (str) – The original target path.
pattern (str) – The pattern to format the temporary path.

Returns:

The temporary target path.

Return type:

str

rename_target(target: str, new_target: str) → None

Rename the target to a new target path.

Parameters:

target (str) – The original target path.
new_target (str) – The new target path.

delete_target(target: str) → None

Delete the target path.

Parameters:: target (str) – The target path to delete.

class anemoi.utils.remote.ssh.RsyncUpload: Bases: SshBaseUpload

class anemoi.utils.remote.ssh.ScpUpload: Bases: SshBaseUpload

anemoi.utils.remote.ssh.upload(source: str, target: str, **kwargs) → None

Upload a file or folder to the target location using rsync.

Parameters:

source (str) – The source file or folder path.
target (str) – The target path.
kwargs (dict) – Additional arguments for the transfer.