Feat: Add copy-existing-asset tab to asset creation flow

documentation/changelog.rst

@@ -27,6 +27,7 @@ New features
 * New ``GET /api/v3_0/sources`` endpoint to list accessible data sources and defined types, with ``only_latest=true`` by default to return only the most recent version per source [see `PR #2126 <https://www.github.com/FlexMeasures/flexmeasures/pull/2126>`_]
 * Add support for filtering sensor data GET requests by ``source-type`` on ``/api/v3_0/sensors/<id>/data`` [see `PR #2127 <https://www.github.com/FlexMeasures/flexmeasures/pull/2127>`_]
 * Making monitoring alerts more flexible: allow ``flexmeasures monitor`` alerts to target one or more user IDs or email addresses with ``--recipient``; ``flexmeasures monitor last-seen`` can now narrow monitored users to one or more accounts with ``--account`` or to client accounts with ``--consultancy`` [see `PR #2158 <https://www.github.com/FlexMeasures/flexmeasures/pull/2158>`_]
+* Support for creating new assets by using another asset as a template from the UI. [see `PR #2195 <https://www.github.com/FlexMeasures/flexmeasures/pull/2195>`_]
 
 Infrastructure / Support
 ----------------------

flexmeasures/api/common/schemas/assets.py

@@ -92,6 +92,15 @@ class AssetAPIQuerySchema(PaginationSchema):
             example=False,
         ),
     )
+    asset_type = fields.Int(
+        data_key="asset_type",
+        required=False,
+        validate=validate.Range(min=1),
+        metadata=dict(
+            description="Filter assets by generic asset type ID.",
+            example=2,
+        ),
+    )
 
 
 class PublicAssetAPISchema(Schema):

flexmeasures/api/common/utils/api_utils.py

@@ -640,6 +640,9 @@ def copy_asset(
                 "Invalid copy target parent: cannot copy an asset to itself or any of its descendants."
             )
 
+        # set external_id to None to avoid conflicts with unique constraint on (account_id, external_id)
+        asset.external_id = None
+
         copied_root, sensor_id_map = _copy_asset_subtree(
             source_asset=asset,
             destination_account_id=target_account_id,

flexmeasures/api/v3_0/assets.py

@@ -267,6 +267,7 @@ def index(
         fields_in_response: list[str] | None,
         all_accessible: bool,
         include_public: bool,
+        asset_type: int | None = None,
         account: Account | None = None,
         root_asset: GenericAsset | None = None,
         max_depth: int | None = None,
@@ -287,6 +288,7 @@ def index(
               - The `account_id` query parameter can be used to list assets from any account (if the user is allowed to read them). Per default, the user's account is used.
               - Alternatively, the `all_accessible` query parameter can be used to list assets from all accounts the current_user has read-access to, plus all public assets. Defaults to `false`.
               - The `include_public` query parameter can be used to include public assets in the response. Defaults to `false`.
+              - The `asset_type` query parameter can be used to filter by generic asset type ID.
               - The `root` query parameter can be used to list only descendants of a given root asset (including the root itself).
               - The `depth` query parameter can be used to search only a max number of descendant generations from the root.
 
@@ -363,6 +365,10 @@ def index(
         filter_statement = GenericAsset.account_id.in_(account_ids)
         if include_public:
             filter_statement = filter_statement | GenericAsset.account_id.is_(None)
+        if asset_type is not None:
+            filter_statement = filter_statement & (
+                GenericAsset.generic_asset_type_id == asset_type
+            )
 
         query = query_assets_by_search_terms(
             search_terms=filter,

flexmeasures/api/v3_0/tests/test_assets_api.py

@@ -170,6 +170,35 @@ def test_get_assets(
         assert turbine["account_id"] == setup_accounts["Supplier"].id
 
 
+@pytest.mark.parametrize("requesting_user", ["test_admin_user@seita.nl"], indirect=True)
+def test_get_assets_filtered_by_asset_type(client, setup_accounts, requesting_user):
+    supplier_account = setup_accounts["Supplier"]
+    supplier_account_id = supplier_account.id
+    supplier_assets = supplier_account.generic_assets
+
+    requested_type_id = supplier_assets[0].generic_asset_type_id
+    expected_assets = [
+        asset
+        for asset in supplier_assets
+        if asset.generic_asset_type_id == requested_type_id
+    ]
+
+    response = client.get(
+        url_for("AssetAPI:index"),
+        query_string={
+            "account_id": supplier_account_id,
+            "asset_type": requested_type_id,
+        },
+    )
+
+    assert response.status_code == 200
+    assert len(response.json) == len(expected_assets)
+    assert all(
+        asset["generic_asset_type"]["id"] == requested_type_id
+        for asset in response.json
+    )
+
+
 @pytest.mark.parametrize(
     "requesting_user, sort_by, sort_dir, expected_name_of_first_sensor",
     [