def _get_dataset()

in azext_edge/edge/providers/rpsaas/adr/assets.py [0:0]

• 15 lines of code
• 14 McCabe index (conditional complexity)

def _get_dataset(asset: dict, dataset_name: str, create_if_none: bool = False):
    # ensure datasets will get populated if not there
    asset["properties"]["datasets"] = asset["properties"].get("datasets", [])
    datasets = asset["properties"]["datasets"]
    matched_datasets = [dset for dset in datasets if dset["name"] == dataset_name]
    # Temporary convert empty names to default
    if not matched_datasets and dataset_name == "default":
        matched_datasets = [dset for dset in datasets if dset["name"] == ""]
    # create if add or import (and no datasets yet)
    if not matched_datasets and create_if_none:
        if dataset_name != "default":
            raise InvalidArgumentValueError("Currently only one dataset with the name default is supported.")
        matched_datasets = [{}]
        datasets.extend(matched_datasets)
    elif not matched_datasets:
        raise InvalidArgumentValueError(f"Dataset {dataset_name} not found in asset {asset['name']}.")
    # note: right now we can have datasets with the same name but this will not be allowed later
    # part of the temporary convert
    matched_datasets[0]["name"] = dataset_name
    return matched_datasets[0]