Skip to content

update_metadata

update_metadata

Service for updating metadata on a scenario.

Classes

UpdateMetadataRunner

Bases: BaseRunner[Dict[str, Any]]

Runner for updating metadata fields on a scenario through the main scenario endpoint.

PUT /api/v3/scenarios/{scenario_id}

Functions
run staticmethod 
run(client, scenario, metadata, **kwargs)

Update metadata for a scenario.

Fields in META_KEYS are set directly on the scenario. Other fields are automatically merged and nested under the 'metadata' field.

Parameters:

Name Type Description Default
client BaseClient

The HTTP client to use

 required
scenario Any

The scenario object (must have an 'id' attribute)

 required
metadata Dict[str, Any]

Dictionary of metadata updates to apply

 required
**kwargs Any

Additional arguments passed to the request

 {}
Example usage

result = UpdateMetadataRunner.run( client=client, scenario=scenario, metadata={ "end_year": 2050, "private": True, "area_code": "nl", } )

Source code in src/pyetm/services/scenario_runners/update_metadata.py 
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@staticmethod
def run(
    client: BaseClient, scenario: Any, metadata: Dict[str, Any], **kwargs: Any
) -> ServiceResult[Dict[str, Any]]:
    """
    Update metadata for a scenario.

    Fields in META_KEYS are set directly on the scenario.
    Other fields are automatically merged and nested under the 'metadata' field.

    Args:
        client: The HTTP client to use
        scenario: The scenario object (must have an 'id' attribute)
        metadata: Dictionary of metadata updates to apply
        **kwargs: Additional arguments passed to the request

    Example usage:
        result = UpdateMetadataRunner.run(
            client=client,
            scenario=scenario,
            metadata={
                "end_year": 2050,
                "private": True,
                "area_code": "nl",
            }
        )
    """
    direct_fields = {}
    nested_metadata = {}
    warnings = []

    for key, value in metadata.items():
        if key in UpdateMetadataRunner.META_KEYS:
            direct_fields[key] = value
        elif key in UpdateMetadataRunner.UNSETTABLE_META_KEYS:
            # Field exists on scenario but cannot be updated - add to nested metadata with warning
            nested_metadata[key] = value
            warnings.append(
                f"Field '{key}' cannot be updated directly and has been added to nested metadata instead"
            )
        else:
            nested_metadata[key] = value

    existing_metadata = {}
    if hasattr(scenario, "metadata") and isinstance(scenario.metadata, dict):
        existing_metadata = scenario.metadata.copy()

    legacy_title = None
    if isinstance(existing_metadata, dict):
        legacy_title = existing_metadata.pop("title", None)
    if "metadata" in direct_fields and isinstance(direct_fields["metadata"], dict):
        direct_fields["metadata"].pop("title", None)
    nested_metadata.pop("title", None)
    if legacy_title is not None and "title" not in direct_fields:
        direct_fields["title"] = legacy_title

    final_metadata = existing_metadata.copy()
    if nested_metadata:
        final_metadata.update(nested_metadata)

    # If user provided a direct "metadata" field, merge it in last (highest priority)
    if "metadata" in direct_fields:
        if isinstance(direct_fields["metadata"], dict):
            final_metadata.update(direct_fields["metadata"])

    if final_metadata or nested_metadata or "metadata" in direct_fields:
        direct_fields["metadata"] = final_metadata

    if "template_id" in direct_fields:
        direct_fields["preset_scenario_id"] = direct_fields.pop("template_id")

    # Transform metadata to the API format
    payload = {"scenario": direct_fields}

    result = UpdateMetadataRunner._make_request(
        client=client,
        method="put",
        path=f"/scenarios/{scenario.id}",
        payload=payload,
    )

    return result