OceanDataStore CLI

Mandatory Arguments

Long version	Short Version	Description
action		Specify the action: send_to_zarr, send_to_icechunk, update_zarr, update_icechunk or list.
--filepaths	-f	Paths to the files to send or update.
--credentials	-c	Path to the JSON file containing the credentials for the object store.
--bucket	-b	Bucket name.

Optional Arguments

Flag	Short Version	Description
--prefix	-p	Object prefix (default=None).
--append-dim	-ad	Append dimension (default=time_counter).
--variables	-v	Variables to send (default=None). Default None will send all variables.
--chunk-strategy	-cs	Chunk strategy as a JSON string (default=None). E.g., '{\"time_counter\": 1, \"x\": 100, \"y\": 100}'
--dask-configuration	-dc	Path to the JSON file defining the Dask Local Cluster configuration (default=None).
--grid-filepath	-gf	File path to model grid file containing domain information (default=None).
--update-coords	-uc	Coordinate dimensions to update as a JSON string (default=None). E.g., '{\"nav_lon\": \"glamt\", \"nav_lat\": \"gphit\"}'
--attributes	-at	Attributes to add to the dataset as a JSON string. E.g., '{\"title\": \"my_dataset\"}'
--zarr-version	-zv	Zarr version used to create the zarr store (default=3). Options are 2 (v2) or 3 (v3).
--branch	-br	Branch of Icechunk repository to commit changes to (default=main).
--commit_message	-cm	Commit message to be recorded when committing changes to Icechunk repository. (default="Add new data to my Icechunk repository").
--variable-commits	-vc	Flag to send variables to Icechunk repository using independent commits.
--icechunk-configuration	-ic	Path to the JSON file defining the Icechunk storage and repository configurations (default=None).

---

OceanDataStore Python API

OceanDataStore.cli.send_to_zarr

send_to_zarr(file, bucket, object_prefix, store_credentials_json, variables=None, append_dim='time_counter', grid_filepath=None, update_coords=None, rechunk=None, attrs=None, client=None, dask_config_kwargs=None, dask_cluster_kwargs=None, zarr_version=3)

Write data to new Zarr store in cloud object storage with option of using dask.

Parameters:

Name	Type	Description	Default
file	list[str] | str | Dataset	Regular expression or list of filepaths to netCDF file(s). Users can also pass a single xarray.Dataset directly.	required
bucket	str	Name of the bucket in the object store. Bucket names can contain only lowercase letters, numbers, dots (.), and hyphens (-).	required
object_prefix	str	Prefix to be added to the object names in the object store.	required
store_credentials_json	str	Path to the JSON file containing the object store credentials.	required
variables	Optional[list[str]]	List of variables to send. If None, all variables will be sent.	None
append_dim	str	Name of the append dimension, by default "time_counter".	'time_counter'
grid_filepath	Optional[str]	Path to file containing model grid parameter.	None
update_coords	Optional[dict]	Dictionary of coordinate variables to update.	None
rechunk	Optional[dict]	Rechunk strategy dictionary, by default None.	None
attrs	Optional[dict]	Attributes to add to the dataset.	None
client	Optional[Client]	Dask Distributed Client.	None
dask_config_kwargs	Optional[dict]	Dask configuration settings passed to dask.config.set(). Ignored if dask client is provided.	None
dask_cluster_kwargs	Optional[dict]	Dask cluster configuration settings passed to LocalCluster(). Ignored if dask client is provided.	None
zarr_version	int	Zarr version to use.	3

Source code in OceanDataStore/cli/zarr.py

def send_to_zarr(
    file: list[str] | str | xr.Dataset,
    bucket: str,
    object_prefix: str,
    store_credentials_json: str,
    variables: Optional[list[str]] = None,
    append_dim: str = "time_counter",
    grid_filepath: Optional[str] = None,
    update_coords: Optional[dict] = None,
    rechunk: Optional[dict] = None,
    attrs: Optional[dict] = None,
    client : Optional[Client] = None,
    dask_config_kwargs: Optional[dict] = None,
    dask_cluster_kwargs: Optional[dict] = None,
    zarr_version: int = 3
) -> None:
    """
    Write data to new Zarr store in cloud object storage with
    option of using dask.

    Parameters
    ----------
    file: list | str | xarray.Dataset
        Regular expression or list of filepaths to netCDF file(s).
        Users can also pass a single xarray.Dataset directly.
    bucket: str
        Name of the bucket in the object store. Bucket names can contain only
        lowercase letters, numbers, dots (.), and hyphens (-).
    object_prefix: str
        Prefix to be added to the object names in the object store.
    store_credentials_json: str
        Path to the JSON file containing the object store credentials.
    variables: list[str], optional
        List of variables to send. If None, all variables will be sent.
    append_dim: str, default="time_counter"
        Name of the append dimension, by default "time_counter".
    grid_filepath: str, optional
        Path to file containing model grid parameter.
    update_coords: dict, optional
        Dictionary of coordinate variables to update.
    rechunk: dict, optional
        Rechunk strategy dictionary, by default None.
    attrs: dict, optional
        Attributes to add to the dataset.
    client: dask.distributed.Client, optional
        Dask Distributed Client.
    dask_config_kwargs: dict[str,str], optional
        Dask configuration settings passed to dask.config.set().
        Ignored if dask client is provided.
    dask_cluster_kwargs: dict, optional
        Dask cluster configuration settings passed to LocalCluster().
        Ignored if dask client is provided.
    zarr_version: int, default=3
        Zarr version to use.
    """
    if client is not None:
        logging.info(f"Using existing Dask Cluster @ Client: {client.dashboard_link}")

        # Register plugins: capture UserWarnings when rechunking data:
        client.register_worker_plugin(CaptureWarningsPlugin())

        # Register plugins: close aiohttp.ClientSessions:
        client.register_worker_plugin(CloseClientSessionPlugin())

        _send_to_zarr(file=file,
                      bucket=bucket,
                      object_prefix=object_prefix,
                      store_credentials_json=store_credentials_json,
                      variables=variables,
                      append_dim=append_dim,
                      grid_filepath=grid_filepath,
                      update_coords=update_coords,
                      rechunk=rechunk,
                      attrs=attrs,
                      parallel=True,
                      zarr_version=zarr_version
                      )

        # --- Shutdown Dask Client --- #
        client.shutdown()
        logging.info("Existing Dask Client has been shutdown.")

    elif dask_cluster_kwargs is not None:
        # === Send to Zarr store with Dask === #
        if dask_config_kwargs is not None:
            dask.config.set(dask_config_kwargs)
            logging.info("Updated dask configuration settings.")

        # Create local dask cluster & client:
        with LocalCluster(**dask_cluster_kwargs) as cluster, Client(cluster) as client:
            logging.info(f"Created LocalCluster with {dask_cluster_kwargs['n_workers']} workers @ Client: {client.dashboard_link}")

            # Register plugins: capture UserWarnings when rechunking data:
            client.register_worker_plugin(CaptureWarningsPlugin())

            # Register plugins: close aiohttp.ClientSessions:
            client.register_worker_plugin(CloseClientSessionPlugin())

            _send_to_zarr(file=file,
                          bucket=bucket,
                          object_prefix=object_prefix,
                          store_credentials_json=store_credentials_json,
                          variables=variables,
                          append_dim=append_dim,
                          grid_filepath=grid_filepath,
                          update_coords=update_coords,
                          rechunk=rechunk,
                          attrs=attrs,
                          parallel=True,
                          zarr_version=zarr_version
                         )

            # --- Shutdown Store & Dask Cluster --- #
            cluster.close()
            client.shutdown()
            logging.info("Dask Cluster has been shutdown.")

    else:
        # === Send to Zarr store without Dask === #
        _send_to_zarr(file=file,
                      bucket=bucket,
                      object_prefix=object_prefix,
                      store_credentials_json=store_credentials_json,
                      variables=variables,
                      append_dim=append_dim,
                      grid_filepath=grid_filepath,
                      update_coords=update_coords,
                      rechunk=rechunk,
                      attrs=attrs,
                      parallel=False,
                      zarr_version=zarr_version
                      )

OceanDataStore.cli.update_zarr

update_zarr(file, bucket, object_prefix, store_credentials_json, variables=None, append_dim='time_counter', grid_filepath=None, update_coords=None, rechunk=None, attrs=None, client=None, dask_config_kwargs=None, dask_cluster_kwargs=None, zarr_version=3)

Update data in existing Zarr store in cloud object storage with option of using dask.

Parameters:

Name	Type	Description	Default
file	list[str] | str | Dataset	Regular expression or list of filepaths to netCDF file(s). Users can also pass a single xarray.Dataset directly.	required
bucket	str	Name of the bucket in the object store. Bucket names can contain only lowercase letters, numbers, dots (.), and hyphens (-).	required
object_prefix	str	Prefix to be added to the object names in the object store.	required
store_credentials_json	str	Path to the JSON file containing the object store credentials.	required
variables	Optional[list[str]]	List of variables to send to Zarr stores. If None, all variables will be sent.	None
append_dim	str	Name of the dimension to append multifile datasets.	'time_counter'
grid_filepath	Optional[str]	Path to file containing model grid parameter.	None
update_coords	Optional[dict]	Dictionary of coordinate variables to update.	None
rechunk	Optional[dict]	Rechunk strategy dictionary.	None
attrs	Optional[dict]	Attributes to add to the dataset.	None
client	Optional[Client]	Dask Distributed Client.	None
dask_config_kwargs	Optional[dict]	Dask configuration settings passed to dask.config.set(). Ignored if dask client is provided.	None
dask_cluster_kwargs	Optional[dict]	Dask cluster configuration settings passed to LocalCluster(). Ignored if dask client is provided.	None
zarr_version	int	zarr version to use.	3

Source code in OceanDataStore/cli/zarr.py

def update_zarr(
    file: list[str] | str | xr.Dataset,
    bucket: str,
    object_prefix: str,
    store_credentials_json: str,
    variables: Optional[list[str]] = None,
    append_dim: str = "time_counter",
    grid_filepath: Optional[str] = None,
    update_coords: Optional[dict] = None,
    rechunk: Optional[dict] = None,
    attrs: Optional[dict] = None,
    client : Optional[Client] = None,
    dask_config_kwargs: Optional[dict] = None,
    dask_cluster_kwargs: Optional[dict] = None,
    zarr_version: int = 3
) -> None:
    """
    Update data in existing Zarr store in cloud object
    storage with option of using dask.

    Parameters
    ----------
    file: list | str | xarray.Dataset
        Regular expression or list of filepaths to netCDF file(s).
        Users can also pass a single xarray.Dataset directly.
    bucket: str
        Name of the bucket in the object store. Bucket names can contain only
        lowercase letters, numbers, dots (.), and hyphens (-).
    object_prefix: str
        Prefix to be added to the object names in the object store.
    store_credentials_json: str
        Path to the JSON file containing the object store credentials.
    variables: list, optional
        List of variables to send to Zarr stores.
        If None, all variables will be sent.
    append_dim: str, default='time_counter'
        Name of the dimension to append multifile datasets.
    grid_filepath: str, optional
        Path to file containing model grid parameter.
    update_coords: dict, optional
        Dictionary of coordinate variables to update.
    rechunk: dict, optional
        Rechunk strategy dictionary.
    attrs: dict, optional
        Attributes to add to the dataset.
    client: dask.distributed.Client, optional
        Dask Distributed Client.
    dask_config_kwargs: Dict[str,str], optional
        Dask configuration settings passed to dask.config.set().
        Ignored if dask client is provided.
    dask_cluster_kwargs: dict, optional
        Dask cluster configuration settings passed to LocalCluster().
        Ignored if dask client is provided.
    zarr_version: int, default=3
        zarr version to use.
    """
    if client is not None:
        logging.info(f"Using existing Dask Cluster @ Client: {client.dashboard_link}")

        # Register plugins: capture UserWarnings when rechunking data:
        client.register_worker_plugin(CaptureWarningsPlugin())

        # Register plugins: close aiohttp.ClientSessions:
        client.register_worker_plugin(CloseClientSessionPlugin())

        _update_zarr(file=file,
                     bucket=bucket,
                     object_prefix=object_prefix,
                     store_credentials_json=store_credentials_json,
                     variables=variables,
                     append_dim=append_dim,
                     grid_filepath=grid_filepath,
                     update_coords=update_coords,
                     rechunk=rechunk,
                     attrs=attrs,
                     parallel=True,
                     zarr_version=zarr_version
                     )

        # --- Shutdown Dask Client --- #
        client.shutdown()
        logging.info("Existing Dask Client has been shutdown.")

    elif dask_cluster_kwargs is not None:
        # === Update Zarr store with Dask === #
        if dask_config_kwargs is not None:
            dask.config.set(dask_config_kwargs)
            logging.info("Updated dask configuration settings.")

        # Create local dask cluster & client:
        with LocalCluster(**dask_cluster_kwargs) as cluster, Client(cluster) as client:
            logging.info(f"Created LocalCluster with {dask_cluster_kwargs['n_workers']} workers @ Client: {client.dashboard_link}")

            # Register plugins: capture UserWarnings when rechunking data:
            client.register_worker_plugin(CaptureWarningsPlugin())

            # Register plugins: close aiohttp.ClientSessions:
            client.register_worker_plugin(CloseClientSessionPlugin())

            _update_zarr(file=file,
                         bucket=bucket,
                         object_prefix=object_prefix,
                         store_credentials_json=store_credentials_json,
                         variables=variables,
                         append_dim=append_dim,
                         grid_filepath=grid_filepath,
                         update_coords=update_coords,
                         rechunk=rechunk,
                         attrs=attrs,
                         parallel=True,
                         zarr_version=zarr_version
                         )

            # --- Shutdown Store & Dask Cluster --- #
            cluster.close()
            client.shutdown()
            logging.info("Dask Cluster has been shutdown.")

    else:
        # === Update Zarr store without Dask === #
        _update_zarr(file=file,
                     bucket=bucket,
                     object_prefix=object_prefix,
                     store_credentials_json=store_credentials_json,
                     variables=variables,
                     append_dim=append_dim,
                     grid_filepath=grid_filepath,
                     update_coords=update_coords,
                     rechunk=rechunk,
                     attrs=attrs,
                     parallel=False,
                     zarr_version=zarr_version
                     )

OceanDataStore.cli.send_to_icechunk

send_to_icechunk(file, bucket, object_prefix, store_credentials_json, exists=False, group=None, variables=None, append_dim='time_counter', grid_filepath=None, update_coords=None, rechunk=None, attrs=None, branch='main', commit_message='Add new data to my Icechunk repository', variable_commits=False, dask_config_kwargs=None, dask_cluster_kwargs=None, icechunk_config=None)

Write data to new Icechunk repository in cloud object storage with option of using dask.

Parameters:

Name	Type	Description	Default
file	list[str] | str | Dataset	Regular expression or list of filepaths to netCDF file(s). Users can also pass a single xarray.Dataset directly.	required
bucket	str	Name of the bucket in the object store. Bucket names can contain only lowercase letters, numbers, dots (.), and hyphens (-).	required
object_prefix	str	Prefix to be added to the object names in the object store.	required
store_credentials_json	str	Path to the JSON file containing the object store credentials.	required
exists	Optional[bool]	Whether to write to an existing Icechunk repository or create a new repository.	False
group	Optional[str]	Group in Icechunk repository to write data to.	None
variables	Optional[list[str]]	List of variables to send. If None, all variables will be sent.	None
append_dim	Optional[str]	Name of the dimension to append multifile datasets.	'time_counter'
grid_filepath	Optional[str]	Path to file containing model grid parameter.	None
update_coords	Optional[dict]	Dictionary of coordinate variables to update.	None
rechunk	Optional[dict]	Rechunk strategy dictionary, by default None.	None
attrs	Optional[dict]	Attributes to add to the dataset.	None
branch	Optional[str]	Branch on which to write data to IcechunkStore.	'main'
commit_message	Optional[str]	Commit message when updating the Icechunk repository.	'Add new data to my Icechunk repository'
variable_commits	Optional[bool]	Whether to write each variable to Icechunk repository using separate commits.	False
dask_config_kwargs	Optional[dict]	Dask configuration settings passed to dask.config.set().	None
dask_cluster_kwargs	Optional[dict]	Dask cluster configuration settings passed to LocalCluster().	None
icechunk_config	Optional[dict]	Icechunk repository configuration.	None

Source code in OceanDataStore/cli/icechunk.py

def send_to_icechunk(
    file: list[str] | str | xr.Dataset,
    bucket: str,
    object_prefix: str,
    store_credentials_json: str,
    exists: Optional[bool] = False,
    group: Optional[str] = None,
    variables: Optional[list[str]] = None,
    append_dim: Optional[str] = 'time_counter',
    grid_filepath: Optional[str] = None,
    update_coords: Optional[dict] = None,
    rechunk: Optional[dict] = None,
    attrs: Optional[dict] = None,
    branch: Optional[str] = "main",
    commit_message: Optional[str] = "Add new data to my Icechunk repository",
    variable_commits: Optional[bool] = False,
    dask_config_kwargs: Optional[dict] = None,
    dask_cluster_kwargs: Optional[dict] = None,
    icechunk_config: Optional[dict] = None,
) -> None:
    """
    Write data to new Icechunk repository in cloud object storage with
    option of using dask.

    Parameters
    ----------
    file: list | str | xarray.Dataset
        Regular expression or list of filepaths to netCDF file(s).
        Users can also pass a single xarray.Dataset directly.
    bucket: str
        Name of the bucket in the object store. Bucket names can contain only
        lowercase letters, numbers, dots (.), and hyphens (-).
    object_prefix: str
        Prefix to be added to the object names in the object store.
    store_credentials_json: str
        Path to the JSON file containing the object store credentials.
    exists: Optional[bool], default=False
        Whether to write to an existing Icechunk repository or create a new repository.
    group: Optional[str], default=None
        Group in Icechunk repository to write data to.
    variables: Optional[list[str]], default=None
        List of variables to send. If None, all variables will be sent.
    append_dim: Optional[str], default='time_counter'
        Name of the dimension to append multifile datasets.
    grid_filepath: Optional[str], default=None
        Path to file containing model grid parameter.
    update_coords: Optional[dict], default=None
        Dictionary of coordinate variables to update.
    rechunk: Optional[dict], default=None
        Rechunk strategy dictionary, by default None.
    attrs: Optional[dict], default=None
        Attributes to add to the dataset.
    branch: Optional[str], default="main"
        Branch on which to write data to IcechunkStore.
    commit_message: Optional[str], default="Initial commit"
        Commit message when updating the Icechunk repository.
    variable_commits: Optional[bool], default=False
        Whether to write each variable to Icechunk repository using
        separate commits.
    dask_config_kwargs: Optional[dict], default=None
        Dask configuration settings passed to dask.config.set().
    dask_cluster_kwargs: Optional[dict], default=None
        Dask cluster configuration settings passed to LocalCluster().
    icechunk_config: Optional[dict], default=None
        Icechunk repository configuration.
    """
    if dask_cluster_kwargs is not None:
        # === Send to Icechunk repo(s) with Dask === #
        if dask_config_kwargs is not None:
            dask.config.set(dask_config_kwargs)
            logging.info("Updated dask configuration settings.")

        # Create local dask cluster & client:
        with LocalCluster(**dask_cluster_kwargs) as cluster, Client(cluster) as client:
            logging.info(f"Created LocalCluster with {dask_cluster_kwargs['n_workers']} workers @ Client: {client.dashboard_link}")

            # Catch UserWarnings when rechunking data:
            client.register_worker_plugin(CaptureWarningsPlugin())

            _send_to_icechunk(file=file,
                              bucket=bucket,
                              object_prefix=object_prefix,
                              store_credentials_json=store_credentials_json,
                              exists=exists,
                              group=group,
                              variables=variables,
                              append_dim=append_dim,
                              grid_filepath=grid_filepath,
                              update_coords=update_coords,
                              rechunk=rechunk,
                              attrs=attrs,
                              parallel=True,
                              branch=branch,
                              commit_message=commit_message,
                              variable_commits=variable_commits,
                              icechunk_config=icechunk_config
                              )

            # --- Shutdown Store & Dask Cluster --- #
            cluster.close()
            client.shutdown()
            logging.info("Dask Cluster has been shutdown.")

    else:
        # === Send to Icechunk repo(s) without Dask === #
        _send_to_icechunk(file=file,
                          bucket=bucket,
                          object_prefix=object_prefix,
                          store_credentials_json=store_credentials_json,
                          exists=exists,
                          group=group,
                          variables=variables,
                          append_dim=append_dim,
                          grid_filepath=grid_filepath,
                          update_coords=update_coords,
                          rechunk=rechunk,
                          attrs=attrs,
                          parallel=False,
                          branch=branch,
                          commit_message=commit_message,
                          variable_commits=variable_commits,
                          icechunk_config=icechunk_config
                          )

OceanDataStore.cli.update_icechunk

update_icechunk(file, bucket, object_prefix, store_credentials_json, group=None, variables=None, append_dim='time_counter', grid_filepath=None, update_coords=None, rechunk=None, attrs=None, branch='main', commit_message='Update data in my Icechunk repository', dask_config_kwargs=None, dask_cluster_kwargs=None, icechunk_config=None)

Update data in existing Icechunk repository in cloud object storage with option of using dask.

Parameters:

Name	Type	Description	Default
file	list[str] | str | Dataset	Regular expression or list of filepaths to netCDF file(s). Users can also pass a single xarray.Dataset directly.	required
bucket	str	Name of the bucket in the object store. Bucket names can contain only lowercase letters, numbers, dots (.), and hyphens (-).	required
object_prefix	str	Prefix to be added to the object names in the object store.	required
store_credentials_json	str	Path to the JSON file containing the object store credentials.	required
group	Optional[str]	Group in Icechunk repository to write data to.	None
variables	Optional[list[str]]	List of variables to send. If None, all variables will be sent.	None
append_dim	Optional[str]	Name of the dimension to append multifile datasets.	'time_counter'
grid_filepath	Optional[str]	Path to file containing model grid parameter.	None
update_coords	Optional[dict]	Dictionary of coordinate variables to update.	None
rechunk	Optional[dict]	Rechunk strategy dictionary, by default None.	None
attrs	Optional[dict]	Attributes to add to the dataset.	None
branch	Optional[str]	Branch on which to write data to IcechunkStore.	'main'
commit_message	Optional[str]	Commit message when updating the Icechunk repository.	'Update data in my Icechunk repository'
dask_config_kwargs	Optional[dict]	Dask configuration settings passed to dask.config.set().	None
dask_cluster_kwargs	Optional[dict]	Dask cluster configuration settings passed to LocalCluster().	None
icechunk_config	Optional[dict]	Icechunk repository configuration.	None

Source code in OceanDataStore/cli/icechunk.py

def update_icechunk(
    file: list[str] | str | xr.Dataset,
    bucket: str,
    object_prefix: str,
    store_credentials_json: str,
    group: Optional[str] = None,
    variables: Optional[list[str]] = None,
    append_dim: Optional[str] = 'time_counter',
    grid_filepath: Optional[str] = None,
    update_coords: Optional[dict] = None,
    rechunk: Optional[dict] = None,
    attrs: Optional[dict] = None,
    branch: Optional[str] = "main",
    commit_message: Optional[str] = "Update data in my Icechunk repository",
    dask_config_kwargs: Optional[dict] = None,
    dask_cluster_kwargs: Optional[dict] = None,
    icechunk_config: Optional[dict] = None,
) -> None:
    """
    Update data in existing Icechunk repository in cloud object
    storage with option of using dask.

    Parameters
    ----------
    file: list | str | xarray.Dataset
        Regular expression or list of filepaths to netCDF file(s).
        Users can also pass a single xarray.Dataset directly.
    bucket: str
        Name of the bucket in the object store. Bucket names can contain only
        lowercase letters, numbers, dots (.), and hyphens (-).
    object_prefix: str
        Prefix to be added to the object names in the object store.
    store_credentials_json: str
        Path to the JSON file containing the object store credentials.
    group: Optional[str], default=None
        Group in Icechunk repository to write data to.
    variables: Optional[list[str]], default=None
        List of variables to send. If None, all variables will be sent.
    append_dim: Optional[str], default='time_counter'
        Name of the dimension to append multifile datasets.
    grid_filepath: Optional[str], default=None
        Path to file containing model grid parameter.
    update_coords: Optional[dict], default=None
        Dictionary of coordinate variables to update.
    rechunk: Optional[dict], default=None
        Rechunk strategy dictionary, by default None.
    attrs: Optional[dict], default=None
        Attributes to add to the dataset.
    branch: Optional[str], default="main"
        Branch on which to write data to IcechunkStore.
    commit_message: Optional[str], default="Initial commit"
        Commit message when updating the Icechunk repository.
    dask_config_kwargs: Optional[dict], default=None
        Dask configuration settings passed to dask.config.set().
    dask_cluster_kwargs: Optional[dict], default=None
        Dask cluster configuration settings passed to LocalCluster().
    icechunk_config: Optional[dict], default=None
        Icechunk repository configuration.
    """
    # === Update Icechunk repo(s) with Dask === #
    if dask_cluster_kwargs is not None:
        if dask_config_kwargs is not None:
            dask.config.set(dask_config_kwargs)
            logging.info("Updated dask configuration settings.")

        # Create local dask cluster & client:
        with LocalCluster(**dask_cluster_kwargs) as cluster, Client(cluster) as client:
            logging.info(f"Created LocalCluster with {dask_cluster_kwargs['n_workers']} workers @ Client: {client.dashboard_link}")

            # Catch UserWarnings when rechunking data:
            client.register_worker_plugin(CaptureWarningsPlugin())

            _update_icechunk(file=file,
                             bucket=bucket,
                             object_prefix=object_prefix,
                             store_credentials_json=store_credentials_json,
                             group=group,
                             variables=variables,
                             append_dim=append_dim,
                             grid_filepath=grid_filepath,
                             update_coords=update_coords,
                             rechunk=rechunk,
                             attrs=attrs,
                             parallel=True,
                             branch=branch,
                             commit_message=commit_message,
                             icechunk_config=icechunk_config
                             )

            # --- Shutdown Store & Dask Cluster --- #
            cluster.close()
            client.shutdown()
            logging.info("Dask Cluster has been shutdown.")

    else:
        # === Update Icechunk repo(s) without Dask === #
        _update_icechunk(file=file,
                         bucket=bucket,
                         object_prefix=object_prefix,
                         store_credentials_json=store_credentials_json,
                         group=group,
                         variables=variables,
                         append_dim=append_dim,
                         grid_filepath=grid_filepath,
                         update_coords=update_coords,
                         rechunk=rechunk,
                         attrs=attrs,
                         parallel=False,
                         branch=branch,
                         commit_message=commit_message,
                         icechunk_config=icechunk_config
                         )