create_scenario

create_scenario

Service for create scenario operations.

Classes

CreateScenarioRunner

Bases: BaseRunner[Dict[str, Any]]

Runner for creating a new scenario.

POST /api/v3/scenarios

Functions

run staticmethod

run(client, scenario_data, **kwargs)

Create a new scenario.

Parameters:

Name	Type	Description	Default
client	BaseClient	The HTTP client to use	required
scenario_data	Dict[str, Any]	Dictionary of scenario attributes for creation	required
**kwargs	Any	Additional arguments passed to the request	{}

Example usage

result = CreateScenarioRunner.run( client=client, scenario_data={ "area_code": "nl", "end_year": 2050, "private": True } )

Source code in src/pyetm/services/scenario_runners/create_scenario.py

@staticmethod
def run(
    client: BaseClient, scenario_data: Dict[str, Any], **kwargs: Any
) -> ServiceResult[Dict[str, Any]]:
    """
    Create a new scenario.

    Args:
        client: The HTTP client to use
        scenario_data: Dictionary of scenario attributes for creation
        **kwargs: Additional arguments passed to the request

    Example usage:
        result = CreateScenarioRunner.run(
            client=client,
            scenario_data={
                "area_code": "nl",
                "end_year": 2050,
                "private": True
            }
        )
    """
    # Validate required fields
    has_template = "template_id" in scenario_data

    missing_required = []
    for key in CreateScenarioRunner.REQUIRED_KEYS:
        if key not in scenario_data and not has_template:
            missing_required.append(key)

    if missing_required:
        return ServiceResult.fail(
            [f"Missing required fields: {', '.join(missing_required)}"]
        )

    # Filter to only allowed fields
    all_allowed = (
        CreateScenarioRunner.REQUIRED_KEYS + CreateScenarioRunner.OPTIONAL_KEYS
    )
    filtered_data = {
        key: value for key, value in scenario_data.items() if key in all_allowed
    }

    warnings = []
    filtered_keys = set(scenario_data.keys()) - set(filtered_data.keys())
    for key in filtered_keys:
        warnings.append(f"Ignoring invalid field for scenario creation: {key!r}")

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

    payload = {"scenario": filtered_data}

    result = CreateScenarioRunner._make_request(
        client=client,
        method="post",
        path="/scenarios",
        payload=payload,
    )

    if result.success and warnings:
        # Merge our warnings with any from the API call
        combined_errors = list(result.errors) + warnings
        assert result.data is not None, "Success result must have data"
        return ServiceResult.ok(data=result.data, errors=combined_errors)

    return result