Skip to content

create_collection

create_collection

Service for creating collections.

Classes

CreateCollectionRunner

Bases: BaseRunner[Dict[str, Any]]

Runner for creating a Collection in MyETM.

POST /api/v3/collections

Functions
run staticmethod 
run(client, collection_data, **kwargs)

Create a new Collection in MyETM.

Parameters:

Name Type Description Default
client BaseClient

The HTTP client to use

 required
collection_data Dict[str, Any]

Dictionary with collection attributes - title (required): Collection title - scenario_ids (optional): List of ETEngine scenario IDs - saved_scenario_ids (optional): List of MyETM saved scenario IDs - area_code (optional): Area code for interpolated collections - end_year (optional): End year for interpolated collections - interpolation (optional): Whether this is an interpolated collection

 required
**kwargs Any

Additional arguments passed to the request

 {}

Returns:

Type Description
ServiceResult[Dict[str, Any]]

ServiceResult with created collection data
Example usage

result = CreateCollectionRunner.run( client=client, collection_data={ "title": "My Collection", "saved_scenario_ids": [1, 2, 3], "interpolation": False } )

Source code in src/pyetm/services/scenario_runners/create_collection.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
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
@staticmethod
def run(
    client: BaseClient,
    collection_data: Dict[str, Any],
    **kwargs: Any
) -> ServiceResult[Dict[str, Any]]:
    """
    Create a new Collection in MyETM.

    Args:
        client: The HTTP client to use
        collection_data: Dictionary with collection attributes
            - title (required): Collection title
            - scenario_ids (optional): List of ETEngine scenario IDs
            - saved_scenario_ids (optional): List of MyETM saved scenario IDs
            - area_code (optional): Area code for interpolated collections
            - end_year (optional): End year for interpolated collections
            - interpolation (optional): Whether this is an interpolated collection
        **kwargs: Additional arguments passed to the request

    Returns:
        ServiceResult with created collection data

    Example usage:
        result = CreateCollectionRunner.run(
            client=client,
            collection_data={
                "title": "My Collection",
                "saved_scenario_ids": [1, 2, 3],
                "interpolation": False
            }
        )
    """
    errors = CreateCollectionRunner._validate_required_fields(
        collection_data, CreateCollectionRunner.REQUIRED_KEYS
    )

    if errors:
        return ServiceResult.fail(errors)

    # Validate that at least one of scenario_ids or saved_scenario_ids is provided
    has_scenarios = bool(collection_data.get("scenario_ids"))
    has_saved_scenarios = bool(collection_data.get("saved_scenario_ids"))

    if not has_scenarios and not has_saved_scenarios:
        return ServiceResult.fail([
            "Collection must include at least one 'scenario_ids' or 'saved_scenario_ids'"
        ])

    all_allowed = (
        CreateCollectionRunner.REQUIRED_KEYS
        + CreateCollectionRunner.OPTIONAL_KEYS
    )
    filtered_data, warnings = CreateCollectionRunner._filter_allowed_fields(
        collection_data,
        all_allowed,
        "create collection",
    )

    payload = {"collection": filtered_data}

    result = CreateCollectionRunner._make_request(
        client=client,
        method="post",
        path="/collections",
        payload=payload,
        **kwargs,
    )

    if result.success and warnings:
        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