From a886a663965afe02c7e55636dfe62eef023d592d Mon Sep 17 00:00:00 2001
From: joshuaunity <oghenerobojosh01@gmail.com>
Date: Thu, 5 Mar 2026 16:25:24 +0100
Subject: [PATCH 1/9] chore: initialize repo

Signed-off-by: joshuaunity <oghenerobojosh01@gmail.com>
---
 flexmeasures/api/v3_0/assets.py           | 64 +++++++++++++++++++++++
 flexmeasures/ui/static/openapi-specs.json |  9 +---
 2 files changed, 65 insertions(+), 8 deletions(-)

diff --git a/flexmeasures/api/v3_0/assets.py b/flexmeasures/api/v3_0/assets.py
index 4a01ea311..f6ac6bf76 100644
--- a/flexmeasures/api/v3_0/assets.py
+++ b/flexmeasures/api/v3_0/assets.py
@@ -159,6 +159,69 @@ class KPIKwargsSchema(Schema):
     event_ends_before = AwareDateTimeField(format="iso", required=False)
 
 
+def fetch_all_assets_in_account(
+    account_id: int, target_account_id: int
+) -> list[GenericAsset]:
+    try:
+        # order from oldest to newest to help with parent/child dependencies
+        assets = db.session.scalars(
+            select(GenericAsset)
+            .filter(GenericAsset.account_id == account_id)
+            .order_by(GenericAsset.id)
+        ).all()
+
+        asset_mapping = {}
+        new_assets = []
+
+        for old_asset in assets:
+            asset_kwargs = {}
+            for column in old_asset.__table__.columns:
+                if column.name not in ["id"]:
+                    asset_kwargs[column.name] = getattr(old_asset, column.name)
+
+            # Avoid name collisions
+            asset_kwargs["name"] = f"{asset_kwargs['name']} (Copy)"
+            # Assign to the target account
+            asset_kwargs["account_id"] = target_account_id
+
+            # Assign the new parent if applicable, otherwise keep the old (or None)
+            if old_asset.parent_asset_id in asset_mapping:
+                asset_kwargs["parent_asset_id"] = asset_mapping[
+                    old_asset.parent_asset_id
+                ].id
+
+            new_asset = GenericAsset(**asset_kwargs)
+            db.session.add(new_asset)
+            db.session.flush()
+
+            asset_mapping[old_asset.id] = new_asset
+            new_assets.append(new_asset)
+
+            sensors = db.session.scalars(
+                select(Sensor).filter(Sensor.generic_asset_id == old_asset.id)
+            ).all()
+
+            for old_sensor in sensors:
+                sensor_kwargs = {}
+                for column in old_sensor.__table__.columns:
+                    if column.name not in ["id", "generic_asset_id"]:
+                        sensor_kwargs[column.name] = getattr(old_sensor, column.name)
+
+                sensor_kwargs["generic_asset_id"] = new_asset.id
+
+                # Check for any external_id collision or similar unique constraints on sensors?
+                # Usually sensors are unique per asset name, or timezone, etc.
+
+                new_sensor = Sensor(**sensor_kwargs)
+                db.session.add(new_sensor)
+
+        db.session.commit()
+        return new_assets
+    except Exception as e:
+        db.session.rollback()
+        raise e
+
+
 class AssetTypesAPI(FlaskView):
     """
     This API view exposes generic asset types.
@@ -295,6 +358,7 @@ def index(
           tags:
             - Assets
         """
+        fetch_all_assets_in_account(account.id, 2) if account else []
 
         # Find out which accounts are relevant
         if account is not None:
diff --git a/flexmeasures/ui/static/openapi-specs.json b/flexmeasures/ui/static/openapi-specs.json
index 1383d75eb..4e00a90d9 100644
--- a/flexmeasures/ui/static/openapi-specs.json
+++ b/flexmeasures/ui/static/openapi-specs.json
@@ -3888,14 +3888,7 @@
     "schemas": {
       "Quantity": {
         "type": "string",
-        "description": "Quantity string describing a fixed quantity.",
-        "examples": [
-          "130 EUR/MWh",
-          "12 V",
-          "4.5 m/s",
-          "20 \u00b0C",
-          "3 * 230V * 16A"
-        ]
+        "description": "Quantity string describing a fixed quantity."
       },
       "SensorReference": {
         "type": "object",

From 5fc6e42598f4ba61a41585056a8885f7ad7f91d7 Mon Sep 17 00:00:00 2001
From: joshuaunity <oghenerobojosh01@gmail.com>
Date: Mon, 9 Mar 2026 11:35:01 +0100
Subject: [PATCH 2/9] chore: work in progress

Signed-off-by: joshuaunity <oghenerobojosh01@gmail.com>
---
 flexmeasures/api/v3_0/assets.py           | 91 ++++++++++++++++-------
 flexmeasures/ui/static/openapi-specs.json |  1 +
 test_sensor_knowledge_horizon.py          | 35 +++++++++
 3 files changed, 100 insertions(+), 27 deletions(-)
 create mode 100644 test_sensor_knowledge_horizon.py

diff --git a/flexmeasures/api/v3_0/assets.py b/flexmeasures/api/v3_0/assets.py
index f6ac6bf76..8e504cb46 100644
--- a/flexmeasures/api/v3_0/assets.py
+++ b/flexmeasures/api/v3_0/assets.py
@@ -83,6 +83,29 @@
 sensors_schema = SensorSchema(many=True)
 
 
+def convert_asset_json_fields(asset_kwargs):
+    """
+    Convert string fields in asset_kwargs to JSON where needed.
+    """
+    if "attributes" in asset_kwargs and isinstance(asset_kwargs["attributes"], str):
+        asset_kwargs["attributes"] = json.loads(asset_kwargs["attributes"])
+    if "sensors_to_show" in asset_kwargs and isinstance(
+        asset_kwargs["sensors_to_show"], str
+    ):
+        asset_kwargs["sensors_to_show"] = json.loads(asset_kwargs["sensors_to_show"])
+    if "flex_context" in asset_kwargs and isinstance(asset_kwargs["flex_context"], str):
+        asset_kwargs["flex_context"] = json.loads(asset_kwargs["flex_context"])
+    if "flex_model" in asset_kwargs and isinstance(asset_kwargs["flex_model"], str):
+        asset_kwargs["flex_model"] = json.loads(asset_kwargs["flex_model"])
+    if "sensors_to_show_as_kpis" in asset_kwargs and isinstance(
+        asset_kwargs["sensors_to_show_as_kpis"], str
+    ):
+        asset_kwargs["sensors_to_show_as_kpis"] = json.loads(
+            asset_kwargs["sensors_to_show_as_kpis"]
+        )
+    return asset_kwargs
+
+
 class AssetTriggerOpenAPISchema(AssetTriggerSchema):
 
     def __init__(self, *args, **kwargs):
@@ -170,25 +193,30 @@ def fetch_all_assets_in_account(
             .order_by(GenericAsset.id)
         ).all()
 
+        if len(assets) == 0:
+            raise ValueError(f"No assets found for account {account_id}.")
+
         asset_mapping = {}
+        parent_mapping = {}
         new_assets = []
 
         for old_asset in assets:
-            asset_kwargs = {}
-            for column in old_asset.__table__.columns:
-                if column.name not in ["id"]:
-                    asset_kwargs[column.name] = getattr(old_asset, column.name)
+            asset_kwargs = asset_schema.dump(old_asset)
+
+            # Remove dump_only and read-only fields
+            for key in ["id", "owner", "generic_asset_type", "child_assets", "sensors"]:
+                asset_kwargs.pop(key, None)
 
             # Avoid name collisions
             asset_kwargs["name"] = f"{asset_kwargs['name']} (Copy)"
             # Assign to the target account
             asset_kwargs["account_id"] = target_account_id
+            asset_kwargs = convert_asset_json_fields(asset_kwargs)
 
-            # Assign the new parent if applicable, otherwise keep the old (or None)
-            if old_asset.parent_asset_id in asset_mapping:
-                asset_kwargs["parent_asset_id"] = asset_mapping[
-                    old_asset.parent_asset_id
-                ].id
+            # Keep track of parent_asset_id to reconnect later
+            if asset_kwargs.get("parent_asset_id"):
+                parent_mapping[old_asset.id] = asset_kwargs["parent_asset_id"]
+            asset_kwargs["parent_asset_id"] = None
 
             new_asset = GenericAsset(**asset_kwargs)
             db.session.add(new_asset)
@@ -197,23 +225,10 @@ def fetch_all_assets_in_account(
             asset_mapping[old_asset.id] = new_asset
             new_assets.append(new_asset)
 
-            sensors = db.session.scalars(
-                select(Sensor).filter(Sensor.generic_asset_id == old_asset.id)
-            ).all()
-
-            for old_sensor in sensors:
-                sensor_kwargs = {}
-                for column in old_sensor.__table__.columns:
-                    if column.name not in ["id", "generic_asset_id"]:
-                        sensor_kwargs[column.name] = getattr(old_sensor, column.name)
-
-                sensor_kwargs["generic_asset_id"] = new_asset.id
-
-                # Check for any external_id collision or similar unique constraints on sensors?
-                # Usually sensors are unique per asset name, or timezone, etc.
-
-                new_sensor = Sensor(**sensor_kwargs)
-                db.session.add(new_sensor)
+        # Second loop to set the proper parent
+        for old_id, old_parent_id in parent_mapping.items():
+            if old_parent_id in asset_mapping:
+                asset_mapping[old_id].parent_asset_id = asset_mapping[old_parent_id].id
 
         db.session.commit()
         return new_assets
@@ -358,7 +373,6 @@ def index(
           tags:
             - Assets
         """
-        fetch_all_assets_in_account(account.id, 2) if account else []
 
         # Find out which accounts are relevant
         if account is not None:
@@ -1606,3 +1620,26 @@ def get_kpis(self, id: int, asset: GenericAsset, start, end):
             }
             kpis.append(kpi_dict)
         return dict(data=kpis), 200
+
+    @route("/copy", methods=["POST"])
+    @use_kwargs(
+        {
+            "account_id": fields.Int(required=True),
+            "target_account_id": fields.Int(required=True),
+        },
+        location="query",
+    )
+    @as_json
+    def copy_assets(self, account_id: int, target_account_id: int):
+        """
+        .. :quickref: Assets; Copy all assets from one account to another.
+        """
+
+        # Ensure the user has administrative privileges or access to both accounts, if needed,
+        # but for now we just call the function as requested.
+        new_assets = fetch_all_assets_in_account(account_id, target_account_id)
+
+        return {
+            "message": f"Successfully copied {len(new_assets)} assets to account {target_account_id}.",
+            "assets": [a.id for a in new_assets],
+        }, 200
diff --git a/flexmeasures/ui/static/openapi-specs.json b/flexmeasures/ui/static/openapi-specs.json
index 4e00a90d9..f119a8e85 100644
--- a/flexmeasures/ui/static/openapi-specs.json
+++ b/flexmeasures/ui/static/openapi-specs.json
@@ -2638,6 +2638,7 @@
         ]
       }
     },
+    "/api/v3_0/assets/copy": {},
     "/api/v3_0/assets/{id}": {
       "delete": {
         "summary": "Delete an asset.",
diff --git a/test_sensor_knowledge_horizon.py b/test_sensor_knowledge_horizon.py
new file mode 100644
index 000000000..5f9b9cf00
--- /dev/null
+++ b/test_sensor_knowledge_horizon.py
@@ -0,0 +1,35 @@
+from flexmeasures.app import create_app
+from flexmeasures.data.models.time_series import Sensor
+
+# Create the Flask application instance
+app = create_app()
+
+# Run within the app context to access the database
+with app.app_context():
+    # Query the first available sensor
+    sensor = Sensor.query.first()
+
+    if sensor:
+        print(f"--- Sensor {sensor.id}: {sensor.name} ---")
+
+        # Access specifically the knowledge horizon attribute
+        if hasattr(sensor, "knowledge_horizon"):
+            print(f"Knowledge Horizon default: {sensor.knowledge_horizon}")
+        else:
+            print(
+                "Sensor object does not have a 'knowledge_horizon' attribute directly."
+            )
+
+        print("\nAll properties in __dict__:")
+        import pprint
+
+        pprint.pprint(sensor.__dict__)
+
+        # Knowledge horizon is sometimes stored in generic_asset or as a separate function.
+        # Let's also check methods related to knowledge horizon if present:
+        knowledge_methods = [m for m in dir(sensor) if "knowledge" in m.lower()]
+        print(f"\nMethods/Attributes related to knowledge: {knowledge_methods}")
+    else:
+        print(
+            "No sensors found in the database. Please ensure you have test data loaded."
+        )

From 09e879fd9f1899d2220707be01daeed6c1bd00a9 Mon Sep 17 00:00:00 2001
From: joshuaunity <oghenerobojosh01@gmail.com>
Date: Wed, 11 Mar 2026 12:20:26 +0100
Subject: [PATCH 3/9] tests: added test to test for cpy asset feature

Signed-off-by: joshuaunity <oghenerobojosh01@gmail.com>
---
 flexmeasures/api/v3_0/assets.py               |  4 +-
 .../api/v3_0/tests/test_assets_api.py         | 47 +++++++++++++++++++
 2 files changed, 49 insertions(+), 2 deletions(-)

diff --git a/flexmeasures/api/v3_0/assets.py b/flexmeasures/api/v3_0/assets.py
index 8e504cb46..6c224431e 100644
--- a/flexmeasures/api/v3_0/assets.py
+++ b/flexmeasures/api/v3_0/assets.py
@@ -182,7 +182,7 @@ class KPIKwargsSchema(Schema):
     event_ends_before = AwareDateTimeField(format="iso", required=False)
 
 
-def fetch_all_assets_in_account(
+def fetch_and_copy_all_assets_in_account(
     account_id: int, target_account_id: int
 ) -> list[GenericAsset]:
     try:
@@ -1637,7 +1637,7 @@ def copy_assets(self, account_id: int, target_account_id: int):
 
         # Ensure the user has administrative privileges or access to both accounts, if needed,
         # but for now we just call the function as requested.
-        new_assets = fetch_all_assets_in_account(account_id, target_account_id)
+        new_assets = fetch_and_copy_all_assets_in_account(account_id, target_account_id)
 
         return {
             "message": f"Successfully copied {len(new_assets)} assets to account {target_account_id}.",
diff --git a/flexmeasures/api/v3_0/tests/test_assets_api.py b/flexmeasures/api/v3_0/tests/test_assets_api.py
index b762cd21b..959aadac8 100644
--- a/flexmeasures/api/v3_0/tests/test_assets_api.py
+++ b/flexmeasures/api/v3_0/tests/test_assets_api.py
@@ -9,6 +9,7 @@
 from flexmeasures.data.services.users import find_user_by_email
 from flexmeasures.api.tests.utils import get_auth_token, UserContext, AccountContext
 from flexmeasures.api.v3_0.tests.utils import get_asset_post_data, check_audit_log_event
+from flexmeasures.api.v3_0.assets import fetch_and_copy_all_assets_in_account
 from flexmeasures.utils.unit_utils import is_valid_unit
 
 
@@ -685,3 +686,49 @@ def test_consultant_get_asset(
     print("Server responded with:\n%s" % get_asset_response.json)
     assert get_asset_response.status_code == 200
     assert get_asset_response.json["name"] == "Test ConsultancyClient Asset"
+
+
+def test_fetch_and_copy_all_assets_in_account(setup_api_test_data, setup_accounts, db):
+
+    base_account = setup_accounts["Prosumer"]
+    target_account = setup_accounts["Empty"]
+
+    # Get the original assets in the base account
+    original_assets = db.session.scalars(
+        select(GenericAsset).filter_by(account_id=base_account.id)
+    ).all()
+    original_asset_count = len(original_assets)
+
+    assert (
+        original_asset_count > 0
+    ), "Base account should have at least one asset to test properly"
+
+    # Count assets in the target account before the operation
+    target_assets_before = db.session.scalars(
+        select(GenericAsset).filter_by(account_id=target_account.id)
+    ).all()
+    target_asset_count_before = len(target_assets_before)
+
+    assert (
+        target_asset_count_before == 0
+    ), "Empty account should have exactly 0 assets initially"
+
+    # Call the copy function
+    new_assets = fetch_and_copy_all_assets_in_account(
+        base_account.id, target_account.id
+    )
+
+    # 1. Check if the amount of assets copied are complete
+    target_assets_after = db.session.scalars(
+        select(GenericAsset).filter_by(account_id=target_account.id)
+    ).all()
+
+    assert len(target_assets_after) == target_asset_count_before + original_asset_count
+    assert len(new_assets) == original_asset_count
+
+    # 2. Using the name of the original asset, search if it exists in the new account
+    new_asset_names = [a.name for a in target_assets_after]
+
+    for old_asset in original_assets:
+        expected_new_name = f"{old_asset.name} (Copy)"
+        assert expected_new_name in new_asset_names

From 3f97255064e77e4ecc80704d06ae450db8ad9cbb Mon Sep 17 00:00:00 2001
From: joshuaunity <oghenerobojosh01@gmail.com>
Date: Wed, 11 Mar 2026 12:28:15 +0100
Subject: [PATCH 4/9] chore: removed unused file

---
 test_sensor_knowledge_horizon.py | 35 --------------------------------
 1 file changed, 35 deletions(-)
 delete mode 100644 test_sensor_knowledge_horizon.py

diff --git a/test_sensor_knowledge_horizon.py b/test_sensor_knowledge_horizon.py
deleted file mode 100644
index 5f9b9cf00..000000000
--- a/test_sensor_knowledge_horizon.py
+++ /dev/null
@@ -1,35 +0,0 @@
-from flexmeasures.app import create_app
-from flexmeasures.data.models.time_series import Sensor
-
-# Create the Flask application instance
-app = create_app()
-
-# Run within the app context to access the database
-with app.app_context():
-    # Query the first available sensor
-    sensor = Sensor.query.first()
-
-    if sensor:
-        print(f"--- Sensor {sensor.id}: {sensor.name} ---")
-
-        # Access specifically the knowledge horizon attribute
-        if hasattr(sensor, "knowledge_horizon"):
-            print(f"Knowledge Horizon default: {sensor.knowledge_horizon}")
-        else:
-            print(
-                "Sensor object does not have a 'knowledge_horizon' attribute directly."
-            )
-
-        print("\nAll properties in __dict__:")
-        import pprint
-
-        pprint.pprint(sensor.__dict__)
-
-        # Knowledge horizon is sometimes stored in generic_asset or as a separate function.
-        # Let's also check methods related to knowledge horizon if present:
-        knowledge_methods = [m for m in dir(sensor) if "knowledge" in m.lower()]
-        print(f"\nMethods/Attributes related to knowledge: {knowledge_methods}")
-    else:
-        print(
-            "No sensors found in the database. Please ensure you have test data loaded."
-        )

From d609e37924ade02c3e44b4aee849300f838b6230 Mon Sep 17 00:00:00 2001
From: joshuaunity <oghenerobojosh01@gmail.com>
Date: Wed, 11 Mar 2026 12:39:31 +0100
Subject: [PATCH 5/9] refactor: relocate util function

Signed-off-by: joshuaunity <oghenerobojosh01@gmail.com>
---
 flexmeasures/api/common/utils/api_utils.py    | 83 +++++++++++++++++++
 flexmeasures/api/v3_0/assets.py               | 83 +------------------
 .../api/v3_0/tests/test_assets_api.py         |  6 +-
 3 files changed, 88 insertions(+), 84 deletions(-)

diff --git a/flexmeasures/api/common/utils/api_utils.py b/flexmeasures/api/common/utils/api_utils.py
index 5b70551ef..be9e195c8 100644
--- a/flexmeasures/api/common/utils/api_utils.py
+++ b/flexmeasures/api/common/utils/api_utils.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 from timely_beliefs.beliefs.classes import BeliefsDataFrame
 from typing import Sequence
 from datetime import timedelta
@@ -14,6 +15,7 @@
 
 from flexmeasures.data import db
 from flexmeasures.data.models.user import Account
+from flexmeasures.data.models.generic_assets import GenericAsset
 from flexmeasures.data.utils import save_to_db
 from flexmeasures.auth.policy import check_access
 from flexmeasures.api.common.responses import (
@@ -22,6 +24,7 @@
     request_processed,
     already_received_and_successfully_processed,
 )
+from flexmeasures.data.schemas.generic_assets import GenericAssetSchema as AssetSchema
 from flexmeasures.utils.error_utils import error_handling_router
 from flexmeasures.utils.flexmeasures_inflection import capitalize
 
@@ -182,3 +185,83 @@ def get_accessible_accounts() -> list[Account]:
             pass
 
     return accounts
+
+
+def convert_asset_json_fields(asset_kwargs):
+    """
+    Convert string fields in asset_kwargs to JSON where needed.
+    """
+    if "attributes" in asset_kwargs and isinstance(asset_kwargs["attributes"], str):
+        asset_kwargs["attributes"] = json.loads(asset_kwargs["attributes"])
+    if "sensors_to_show" in asset_kwargs and isinstance(
+        asset_kwargs["sensors_to_show"], str
+    ):
+        asset_kwargs["sensors_to_show"] = json.loads(asset_kwargs["sensors_to_show"])
+    if "flex_context" in asset_kwargs and isinstance(asset_kwargs["flex_context"], str):
+        asset_kwargs["flex_context"] = json.loads(asset_kwargs["flex_context"])
+    if "flex_model" in asset_kwargs and isinstance(asset_kwargs["flex_model"], str):
+        asset_kwargs["flex_model"] = json.loads(asset_kwargs["flex_model"])
+    if "sensors_to_show_as_kpis" in asset_kwargs and isinstance(
+        asset_kwargs["sensors_to_show_as_kpis"], str
+    ):
+        asset_kwargs["sensors_to_show_as_kpis"] = json.loads(
+            asset_kwargs["sensors_to_show_as_kpis"]
+        )
+    return asset_kwargs
+
+
+def fetch_and_copy_all_assets_in_account(
+    account_id: int, target_account_id: int
+) -> list[GenericAsset]:
+    try:
+        asset_schema = AssetSchema()
+
+        # order from oldest to newest to help with parent/child dependencies
+        assets = db.session.scalars(
+            select(GenericAsset)
+            .filter(GenericAsset.account_id == account_id)
+            .order_by(GenericAsset.id)
+        ).all()
+
+        if len(assets) == 0:
+            raise ValueError(f"No assets found for account {account_id}.")
+
+        asset_mapping = {}
+        parent_mapping = {}
+        new_assets = []
+
+        for old_asset in assets:
+            asset_kwargs = asset_schema.dump(old_asset)
+
+            # Remove dump_only and read-only fields
+            for key in ["id", "owner", "generic_asset_type", "child_assets", "sensors"]:
+                asset_kwargs.pop(key, None)
+
+            # Avoid name collisions
+            asset_kwargs["name"] = f"{asset_kwargs['name']} (Copy)"
+            # Assign to the target account
+            asset_kwargs["account_id"] = target_account_id
+            asset_kwargs = convert_asset_json_fields(asset_kwargs)
+
+            # Keep track of parent_asset_id to reconnect later
+            if asset_kwargs.get("parent_asset_id"):
+                parent_mapping[old_asset.id] = asset_kwargs["parent_asset_id"]
+            asset_kwargs["parent_asset_id"] = None
+
+            new_asset = GenericAsset(**asset_kwargs)
+            db.session.add(new_asset)
+            db.session.flush()
+
+            asset_mapping[old_asset.id] = new_asset
+            new_assets.append(new_asset)
+
+        # Second loop to set the proper parent
+        for old_id, old_parent_id in parent_mapping.items():
+            if old_parent_id in asset_mapping:
+                asset_mapping[old_id].parent_asset_id = asset_mapping[old_parent_id].id
+
+        db.session.commit()
+        return new_assets
+    except Exception as e:
+        db.session.rollback()
+        raise e
diff --git a/flexmeasures/api/v3_0/assets.py b/flexmeasures/api/v3_0/assets.py
index 6c224431e..6608b9955 100644
--- a/flexmeasures/api/v3_0/assets.py
+++ b/flexmeasures/api/v3_0/assets.py
@@ -56,7 +56,10 @@
     create_sequential_scheduling_job,
     create_simultaneous_scheduling_job,
 )
-from flexmeasures.api.common.utils.api_utils import get_accessible_accounts
+from flexmeasures.api.common.utils.api_utils import (
+    get_accessible_accounts,
+    fetch_and_copy_all_assets_in_account,
+)
 from flexmeasures.api.common.responses import (
     unprocessable_entity,
     request_processed,
@@ -83,29 +86,6 @@
 sensors_schema = SensorSchema(many=True)
 
 
-def convert_asset_json_fields(asset_kwargs):
-    """
-    Convert string fields in asset_kwargs to JSON where needed.
-    """
-    if "attributes" in asset_kwargs and isinstance(asset_kwargs["attributes"], str):
-        asset_kwargs["attributes"] = json.loads(asset_kwargs["attributes"])
-    if "sensors_to_show" in asset_kwargs and isinstance(
-        asset_kwargs["sensors_to_show"], str
-    ):
-        asset_kwargs["sensors_to_show"] = json.loads(asset_kwargs["sensors_to_show"])
-    if "flex_context" in asset_kwargs and isinstance(asset_kwargs["flex_context"], str):
-        asset_kwargs["flex_context"] = json.loads(asset_kwargs["flex_context"])
-    if "flex_model" in asset_kwargs and isinstance(asset_kwargs["flex_model"], str):
-        asset_kwargs["flex_model"] = json.loads(asset_kwargs["flex_model"])
-    if "sensors_to_show_as_kpis" in asset_kwargs and isinstance(
-        asset_kwargs["sensors_to_show_as_kpis"], str
-    ):
-        asset_kwargs["sensors_to_show_as_kpis"] = json.loads(
-            asset_kwargs["sensors_to_show_as_kpis"]
-        )
-    return asset_kwargs
-
-
 class AssetTriggerOpenAPISchema(AssetTriggerSchema):
 
     def __init__(self, *args, **kwargs):
@@ -182,61 +162,6 @@ class KPIKwargsSchema(Schema):
     event_ends_before = AwareDateTimeField(format="iso", required=False)
 
 
-def fetch_and_copy_all_assets_in_account(
-    account_id: int, target_account_id: int
-) -> list[GenericAsset]:
-    try:
-        # order from oldest to newest to help with parent/child dependencies
-        assets = db.session.scalars(
-            select(GenericAsset)
-            .filter(GenericAsset.account_id == account_id)
-            .order_by(GenericAsset.id)
-        ).all()
-
-        if len(assets) == 0:
-            raise ValueError(f"No assets found for account {account_id}.")
-
-        asset_mapping = {}
-        parent_mapping = {}
-        new_assets = []
-
-        for old_asset in assets:
-            asset_kwargs = asset_schema.dump(old_asset)
-
-            # Remove dump_only and read-only fields
-            for key in ["id", "owner", "generic_asset_type", "child_assets", "sensors"]:
-                asset_kwargs.pop(key, None)
-
-            # Avoid name collisions
-            asset_kwargs["name"] = f"{asset_kwargs['name']} (Copy)"
-            # Assign to the target account
-            asset_kwargs["account_id"] = target_account_id
-            asset_kwargs = convert_asset_json_fields(asset_kwargs)
-
-            # Keep track of parent_asset_id to reconnect later
-            if asset_kwargs.get("parent_asset_id"):
-                parent_mapping[old_asset.id] = asset_kwargs["parent_asset_id"]
-            asset_kwargs["parent_asset_id"] = None
-
-            new_asset = GenericAsset(**asset_kwargs)
-            db.session.add(new_asset)
-            db.session.flush()
-
-            asset_mapping[old_asset.id] = new_asset
-            new_assets.append(new_asset)
-
-        # Second loop to set the proper parent
-        for old_id, old_parent_id in parent_mapping.items():
-            if old_parent_id in asset_mapping:
-                asset_mapping[old_id].parent_asset_id = asset_mapping[old_parent_id].id
-
-        db.session.commit()
-        return new_assets
-    except Exception as e:
-        db.session.rollback()
-        raise e
-
-
 class AssetTypesAPI(FlaskView):
     """
     This API view exposes generic asset types.
diff --git a/flexmeasures/api/v3_0/tests/test_assets_api.py b/flexmeasures/api/v3_0/tests/test_assets_api.py
index 959aadac8..17f89ebf0 100644
--- a/flexmeasures/api/v3_0/tests/test_assets_api.py
+++ b/flexmeasures/api/v3_0/tests/test_assets_api.py
@@ -9,7 +9,7 @@
 from flexmeasures.data.services.users import find_user_by_email
 from flexmeasures.api.tests.utils import get_auth_token, UserContext, AccountContext
 from flexmeasures.api.v3_0.tests.utils import get_asset_post_data, check_audit_log_event
-from flexmeasures.api.v3_0.assets import fetch_and_copy_all_assets_in_account
+from flexmeasures.api.common.utils.api_utils import fetch_and_copy_all_assets_in_account
 from flexmeasures.utils.unit_utils import is_valid_unit
 
 
@@ -709,10 +709,6 @@ def test_fetch_and_copy_all_assets_in_account(setup_api_test_data, setup_account
     ).all()
     target_asset_count_before = len(target_assets_before)
 
-    assert (
-        target_asset_count_before == 0
-    ), "Empty account should have exactly 0 assets initially"
-
     # Call the copy function
     new_assets = fetch_and_copy_all_assets_in_account(
         base_account.id, target_account.id

From abd590bb5acc805cbe6bca96f98bdf81f724f692 Mon Sep 17 00:00:00 2001
From: Joshua Edward <oghenerobojosh01@gmail.com>
Date: Fri, 13 Mar 2026 11:57:27 +0100
Subject: [PATCH 6/9] refacto: Refactored util for copying asset

Signed-off-by: Joshua Edward <oghenerobojosh01@gmail.com>
---
 flexmeasures/api/common/utils/api_utils.py    |  95 +++++++------
 flexmeasures/api/v3_0/assets.py               |  86 +++++++++---
 .../api/v3_0/tests/test_assets_api.py         | 131 +++++++++++++-----
 3 files changed, 215 insertions(+), 97 deletions(-)

diff --git a/flexmeasures/api/common/utils/api_utils.py b/flexmeasures/api/common/utils/api_utils.py
index be9e195c8..1e56c6cf3 100644
--- a/flexmeasures/api/common/utils/api_utils.py
+++ b/flexmeasures/api/common/utils/api_utils.py
@@ -210,58 +210,61 @@ def convert_asset_json_fields(asset_kwargs):
     return asset_kwargs
 
 
-def fetch_and_copy_all_assets_in_account(
-    account_id: int, target_account_id: int
-) -> list[GenericAsset]:
+def copy_asset(
+    asset: GenericAsset,
+    account=None,
+    parent_asset=None,
+) -> GenericAsset:
+    """
+    Copy a single asset to a target account and/or under a target parent asset.
+
+    Resolution rules:
+
+    - If neither ``account`` nor ``parent_asset`` is given, the copy is placed in
+      the same account and under the same parent as the original (i.e. a sibling).
+    - If ``account`` is given but ``parent_asset`` is not, the copy becomes a
+      top-level asset (no parent) in the given account.
+    - If ``parent_asset`` is given but ``account`` is not, the copy is placed under
+      the given parent and inherits that parent's account.
+    - If both are given, the copy belongs to the given account and is placed under
+      the given parent. This allows creating a copy that belongs to a different
+      account than its parent.
+    """
     try:
         asset_schema = AssetSchema()
-
-        # order from oldest to newest to help with parent/child dependencies
-        assets = db.session.scalars(
-            select(GenericAsset)
-            .filter(GenericAsset.account_id == account_id)
-            .order_by(GenericAsset.id)
-        ).all()
-
-        if len(assets) == 0:
-            raise ValueError(f"No assets found for account {account_id}.")
-
-        asset_mapping = {}
-        parent_mapping = {}
-        new_assets = []
-
-        for old_asset in assets:
-            asset_kwargs = asset_schema.dump(old_asset)
-
-            # Remove dump_only and read-only fields
-            for key in ["id", "owner", "generic_asset_type", "child_assets", "sensors"]:
-                asset_kwargs.pop(key, None)
-
-            # Avoid name collisions
-            asset_kwargs["name"] = f"{asset_kwargs['name']} (Copy)"
-            # Assign to the target account
-            asset_kwargs["account_id"] = target_account_id
-            asset_kwargs = convert_asset_json_fields(asset_kwargs)
-
-            # Keep track of parent_asset_id to reconnect later
-            if asset_kwargs.get("parent_asset_id"):
-                parent_mapping[old_asset.id] = asset_kwargs["parent_asset_id"]
+        asset_kwargs = asset_schema.dump(asset)
+
+        # Remove dump_only and read-only fields
+        for key in ["id", "owner", "generic_asset_type", "child_assets", "sensors"]:
+            asset_kwargs.pop(key, None)
+
+        # Avoid name collision
+        asset_kwargs["name"] = f"{asset_kwargs['name']} (Copy)"
+
+        # Resolve target account and parent
+        if account is None and parent_asset is None:
+            # Neither given: preserve the original asset's placement
+            asset_kwargs["account_id"] = asset.account_id
+            asset_kwargs["parent_asset_id"] = asset.parent_asset_id
+        elif account is not None and parent_asset is None:
+            # Only account given: top-level asset in the target account
+            asset_kwargs["account_id"] = account.id
             asset_kwargs["parent_asset_id"] = None
+        elif account is None and parent_asset is not None:
+            # Only parent given: inherit the parent's account
+            asset_kwargs["account_id"] = parent_asset.account_id
+            asset_kwargs["parent_asset_id"] = parent_asset.id
+        else:
+            # Both given: explicit placement, possibly cross-account
+            asset_kwargs["account_id"] = account.id
+            asset_kwargs["parent_asset_id"] = parent_asset.id
 
-            new_asset = GenericAsset(**asset_kwargs)
-            db.session.add(new_asset)
-            db.session.flush()
-
-            asset_mapping[old_asset.id] = new_asset
-            new_assets.append(new_asset)
-
-        # Second loop to set the proper parent
-        for old_id, old_parent_id in parent_mapping.items():
-            if old_parent_id in asset_mapping:
-                asset_mapping[old_id].parent_asset_id = asset_mapping[old_parent_id].id
+        asset_kwargs = convert_asset_json_fields(asset_kwargs)
 
+        new_asset = GenericAsset(**asset_kwargs)
+        db.session.add(new_asset)
         db.session.commit()
-        return new_assets
+        return new_asset
     except Exception as e:
         db.session.rollback()
         raise e
diff --git a/flexmeasures/api/v3_0/assets.py b/flexmeasures/api/v3_0/assets.py
index 6608b9955..9944cbabd 100644
--- a/flexmeasures/api/v3_0/assets.py
+++ b/flexmeasures/api/v3_0/assets.py
@@ -11,7 +11,7 @@
 from flask_json import as_json
 from flask_sqlalchemy.pagination import SelectPagination
 
-from marshmallow import fields, ValidationError, Schema, validate
+from marshmallow import fields, post_load, ValidationError, Schema, validate
 
 from webargs.flaskparser import use_kwargs, use_args
 from sqlalchemy import select, func, or_
@@ -58,7 +58,7 @@
 )
 from flexmeasures.api.common.utils.api_utils import (
     get_accessible_accounts,
-    fetch_and_copy_all_assets_in_account,
+    copy_asset,
 )
 from flexmeasures.api.common.responses import (
     unprocessable_entity,
@@ -162,6 +162,48 @@ class KPIKwargsSchema(Schema):
     event_ends_before = AwareDateTimeField(format="iso", required=False)
 
 
+class CopyAssetSchema(Schema):
+    account = AccountIdField(
+        data_key="account_id",
+        required=False,
+        metadata=dict(
+            description="Target account to copy the asset to.",
+            example=67,
+        ),
+    )
+    parent_asset = AssetIdField(
+        data_key="parent_id",
+        required=False,
+        metadata=dict(
+            description="Target parent asset to copy the asset under.",
+            example=482,
+        ),
+    )
+
+    @post_load
+    def resolve_account_and_parent(self, data, **kwargs):
+        """
+        Resolve the account/parent relationship after loading:
+
+        - If ``account`` is explicitly given but ``parent_asset`` is not, the copy
+          becomes a top-level asset (no parent) in the given account.
+        - If ``parent_asset`` is explicitly given but ``account`` is not, the copy
+          inherits the account of the parent asset.
+        - If both are explicitly given, the copy can belong to a different account
+          than its parent, which is a valid cross-account parent relationship.
+        - If neither is given, the original asset's account and parent are preserved.
+        """
+        account_given = "account" in data
+        parent_given = "parent_asset" in data
+
+        if account_given and not parent_given:
+            data["parent_asset"] = None
+        elif parent_given and not account_given:
+            data["account"] = data["parent_asset"].owner
+
+        return data
+
+
 class AssetTypesAPI(FlaskView):
     """
     This API view exposes generic asset types.
@@ -1546,25 +1588,37 @@ def get_kpis(self, id: int, asset: GenericAsset, start, end):
             kpis.append(kpi_dict)
         return dict(data=kpis), 200
 
-    @route("/copy", methods=["POST"])
+    @route("/<id>/copy", methods=["POST"])
     @use_kwargs(
-        {
-            "account_id": fields.Int(required=True),
-            "target_account_id": fields.Int(required=True),
-        },
-        location="query",
+        {"asset": AssetIdField(data_key="id", status_if_not_found=HTTPStatus.NOT_FOUND)},
+        location="path",
     )
+    @use_kwargs(CopyAssetSchema, location="query")
     @as_json
-    def copy_assets(self, account_id: int, target_account_id: int):
+    def copy_assets(self, id, asset: GenericAsset, account=None, parent_asset=None):
         """
-        .. :quickref: Assets; Copy all assets from one account to another.
+        .. :quickref: Assets; Copy an asset to a target account and/or parent.
         """
+        # Resolve the target account and parent for permission checking.
+        # When neither is given (account is None), the copy is placed in the
+        # same account and under the same parent as the original.
+        if account is None:
+            resolved_account = asset.owner
+            resolved_parent = asset.parent_asset
+        else:
+            resolved_account = account
+            resolved_parent = parent_asset
+
+        # Check create-children permission on the target account.
+        check_access(resolved_account, "create-children")
 
-        # Ensure the user has administrative privileges or access to both accounts, if needed,
-        # but for now we just call the function as requested.
-        new_assets = fetch_and_copy_all_assets_in_account(account_id, target_account_id)
+        # Also check create-children permission on the target parent (if any).
+        if resolved_parent is not None:
+            check_access(resolved_parent, "create-children")
+
+        new_asset = copy_asset(asset, account=account, parent_asset=parent_asset)
 
         return {
-            "message": f"Successfully copied {len(new_assets)} assets to account {target_account_id}.",
-            "assets": [a.id for a in new_assets],
-        }, 200
+            "message": f"Successfully copied asset {asset.id} to account {new_asset.account_id}.",
+            "asset": new_asset.id,
+        }, 201
diff --git a/flexmeasures/api/v3_0/tests/test_assets_api.py b/flexmeasures/api/v3_0/tests/test_assets_api.py
index 17f89ebf0..5b458956d 100644
--- a/flexmeasures/api/v3_0/tests/test_assets_api.py
+++ b/flexmeasures/api/v3_0/tests/test_assets_api.py
@@ -9,7 +9,7 @@
 from flexmeasures.data.services.users import find_user_by_email
 from flexmeasures.api.tests.utils import get_auth_token, UserContext, AccountContext
 from flexmeasures.api.v3_0.tests.utils import get_asset_post_data, check_audit_log_event
-from flexmeasures.api.common.utils.api_utils import fetch_and_copy_all_assets_in_account
+from flexmeasures.api.common.utils.api_utils import copy_asset
 from flexmeasures.utils.unit_utils import is_valid_unit
 
 
@@ -688,43 +688,104 @@ def test_consultant_get_asset(
     assert get_asset_response.json["name"] == "Test ConsultancyClient Asset"
 
 
-def test_fetch_and_copy_all_assets_in_account(setup_api_test_data, setup_accounts, db):
-
-    base_account = setup_accounts["Prosumer"]
-    target_account = setup_accounts["Empty"]
-
-    # Get the original assets in the base account
-    original_assets = db.session.scalars(
-        select(GenericAsset).filter_by(account_id=base_account.id)
-    ).all()
-    original_asset_count = len(original_assets)
+def test_copy_asset(setup_api_test_data, setup_accounts, db):
+    """
+    Test all four placement use cases for copy_asset:
 
-    assert (
-        original_asset_count > 0
-    ), "Base account should have at least one asset to test properly"
-
-    # Count assets in the target account before the operation
-    target_assets_before = db.session.scalars(
-        select(GenericAsset).filter_by(account_id=target_account.id)
-    ).all()
-    target_asset_count_before = len(target_assets_before)
-
-    # Call the copy function
-    new_assets = fetch_and_copy_all_assets_in_account(
-        base_account.id, target_account.id
+    1. Neither account nor parent given  → same account, same parent (sibling copy).
+    2. Only account given                → top-level asset in the given account.
+    3. Only parent given                 → under the parent, inheriting its account.
+    4. Both account and parent given     → under the parent, in the given account
+                                           (cross-account parent relationship allowed).
+    """
+    prosumer_account = setup_accounts["Prosumer"]
+    supplier_account = setup_accounts["Supplier"]
+
+    # Source assets created by setup_generic_assets (via setup_api_test_data dependency)
+    battery = db.session.scalars(
+        select(GenericAsset).filter_by(
+            account_id=prosumer_account.id,
+            name="Test grid connected battery storage",
+        )
+    ).first()
+    turbine = db.session.scalars(
+        select(GenericAsset).filter_by(name="Test wind turbine")
+    ).first()
+
+    assert battery is not None, "Battery asset must exist in Prosumer account"
+    assert turbine is not None, "Wind turbine asset must exist in Supplier account"
+
+    # Create a parent asset in the Supplier account for use cases 3 and 4.
+    parent = GenericAsset(
+        name="Test parent for copy",
+        generic_asset_type=battery.generic_asset_type,
+        owner=supplier_account,
     )
+    db.session.add(parent)
+    db.session.flush()
+
+    # 1. Neither given → sibling copy (same account, same parent)
+    copy1 = copy_asset(battery)
+    assert copy1.name == f"{battery.name} (Copy)"
+    assert copy1.account_id == battery.account_id
+    assert copy1.parent_asset_id == battery.parent_asset_id  # None
+
+    # 2. Only account given → top-level in target account
+    # Use the turbine so the name doesn't clash with copy1 (parent_asset_id is None for both).
+    copy2 = copy_asset(turbine, account=prosumer_account)
+    assert copy2.name == f"{turbine.name} (Copy)"
+    assert copy2.account_id == prosumer_account.id
+    assert copy2.parent_asset_id is None
+
+    # 3. Only parent given → under parent, inherits parent's account
+    copy3 = copy_asset(battery, parent_asset=parent)
+    assert copy3.name == f"{battery.name} (Copy)"
+    assert copy3.account_id == parent.account_id  # Supplier account
+    assert copy3.parent_asset_id == parent.id
+
+    # 4. Both given → under parent, in explicitly given account (cross-account)
+    copy4 = copy_asset(turbine, account=prosumer_account, parent_asset=parent)
+    assert copy4.name == f"{turbine.name} (Copy)"
+    assert copy4.account_id == prosumer_account.id
+    assert copy4.parent_asset_id == parent.id
+
+
+def test_copy_asset_fails_on_duplicate_name_under_same_parent(
+    setup_api_test_data, setup_accounts, db
+):
+    """
+    Copying the same asset twice under the same parent raises an IntegrityError.
 
-    # 1. Check if the amount of assets copied are complete
-    target_assets_after = db.session.scalars(
-        select(GenericAsset).filter_by(account_id=target_account.id)
-    ).all()
+    The DB enforces UNIQUE(name, parent_asset_id). The first copy succeeds
+    producing e.g. 'Battery (Copy)' under the given parent. The second copy
+    tries to insert another row with the exact same (name, parent_asset_id)
+    pair, which violates the constraint.
+    """
+    from sqlalchemy.exc import IntegrityError
 
-    assert len(target_assets_after) == target_asset_count_before + original_asset_count
-    assert len(new_assets) == original_asset_count
+    prosumer_account = setup_accounts["Prosumer"]
+    battery = db.session.scalars(
+        select(GenericAsset).filter_by(
+            account_id=prosumer_account.id,
+            name="Test grid connected battery storage",
+        )
+    ).first()
+    assert battery is not None
+
+    # Create a dedicated parent so this test is independent of others.
+    parent = GenericAsset(
+        name="Test parent for duplicate-name failure",
+        generic_asset_type=battery.generic_asset_type,
+        owner=prosumer_account,
+    )
+    db.session.add(parent)
+    db.session.flush()
 
-    # 2. Using the name of the original asset, search if it exists in the new account
-    new_asset_names = [a.name for a in target_assets_after]
+    # First copy under the parent succeeds.
+    first_copy = copy_asset(battery, parent_asset=parent)
+    assert first_copy.parent_asset_id == parent.id
 
-    for old_asset in original_assets:
-        expected_new_name = f"{old_asset.name} (Copy)"
-        assert expected_new_name in new_asset_names
+    # Second copy under the same parent fails: UNIQUE(name, parent_asset_id) is violated
+    # because parent_asset_id is non-NULL (PostgreSQL only treats NULLs as distinct).
+    with pytest.raises(IntegrityError):
+        copy_asset(battery, parent_asset=parent)

From fb39eaca55dd8b1a7a251bd4b949dc0a82d89c6f Mon Sep 17 00:00:00 2001
From: Joshua Edward <oghenerobojosh01@gmail.com>
Date: Fri, 13 Mar 2026 17:23:06 +0100
Subject: [PATCH 7/9] tests: and new test case

Signed-off-by: Joshua Edward <oghenerobojosh01@gmail.com>
---
 .../api/v3_0/tests/test_assets_api.py         | 108 +++++++++++++++++-
 1 file changed, 107 insertions(+), 1 deletion(-)

diff --git a/flexmeasures/api/v3_0/tests/test_assets_api.py b/flexmeasures/api/v3_0/tests/test_assets_api.py
index 5b458956d..127a6e8c4 100644
--- a/flexmeasures/api/v3_0/tests/test_assets_api.py
+++ b/flexmeasures/api/v3_0/tests/test_assets_api.py
@@ -1,11 +1,13 @@
 import json
+from datetime import timedelta
 
 from flask import url_for
 import pytest
 from sqlalchemy import select, func
 
 from flexmeasures.data.models.audit_log import AssetAuditLog
-from flexmeasures.data.models.generic_assets import GenericAsset
+from flexmeasures.data.models.generic_assets import GenericAsset, GenericAssetType
+from flexmeasures.data.models.time_series import Sensor
 from flexmeasures.data.services.users import find_user_by_email
 from flexmeasures.api.tests.utils import get_auth_token, UserContext, AccountContext
 from flexmeasures.api.v3_0.tests.utils import get_asset_post_data, check_audit_log_event
@@ -789,3 +791,107 @@ def test_copy_asset_fails_on_duplicate_name_under_same_parent(
     # because parent_asset_id is non-NULL (PostgreSQL only treats NULLs as distinct).
     with pytest.raises(IntegrityError):
         copy_asset(battery, parent_asset=parent)
+
+
+def test_copy_asset_to_another_account_preserves_config(
+    setup_api_test_data, setup_accounts, setup_markets, setup_generic_asset_types, db
+):
+    """
+    Copy a richly configured asset from one account to another and verify everything
+    is preserved correctly.
+
+    Source asset layout (Prosumer account):
+
+    House  (EMS)
+    ├── flex_context:
+    │     - consumption-price  → sensor on the public epex asset (no account)
+    │     - site-power-capacity → sensor on the House itself (kW capacity)
+    ├── EV charger 1  (child)
+    │     - flex_model: { "power-capacity": "7.4 kW", "soc-unit": "kWh" }
+    │     - sensors: power (kW), energy (kWh)
+    └── EV charger 2  (child)
+          - flex_model: { "power-capacity": "7.4 kW", "soc-unit": "kWh" }
+          - sensors: power (kW), energy (kWh)
+
+    Assertions after copying House to the Supplier account:
+    1. The copy lands in the Supplier account with the expected name.
+    2. The copy is a top-level asset (no parent).
+    3. flex_context is preserved verbatim (sensor IDs are unchanged).
+    4. copy_asset is a *shallow* copy: the original child assets are not duplicated.
+    """
+    prosumer_account = setup_accounts["Prosumer"]
+    supplier_account = setup_accounts["Supplier"]
+
+    # The epex_da sensor lives on the public "epex" asset (account_id=None).
+    price_sensor = setup_markets["epex_da"]
+    assert price_sensor.generic_asset.account_id is None, "epex must be a public asset"
+
+    asset_type = setup_generic_asset_types["battery"]
+    charger_type = setup_generic_asset_types["wind"]
+
+    # Build the source house asset.
+    house = GenericAsset(
+        name="Test house for rich copy",
+        generic_asset_type=asset_type,
+        owner=prosumer_account,
+    )
+    db.session.add(house)
+    db.session.flush()  # obtain house.id before adding sensors
+
+    # A kW sensor on the house itself, referenced as the site-power-capacity.
+    site_capacity_sensor = Sensor(
+        name="site capacity",
+        generic_asset=house,
+        event_resolution=timedelta(minutes=15),
+        unit="kW",
+    )
+    db.session.add(site_capacity_sensor)
+    db.session.flush()
+
+    house.flex_context = {
+        "consumption-price": {"sensor": price_sensor.id},
+        "site-power-capacity": {"sensor": site_capacity_sensor.id},
+    }
+
+    # Two child assets, each with two sensors and a two-setting flex_model.
+    for i in range(1, 3):
+        charger = GenericAsset(
+            name=f"EV charger {i}",
+            generic_asset_type=charger_type,
+            owner=prosumer_account,
+            parent_asset_id=house.id,
+            flex_model={"power-capacity": "7.4 kW", "soc-unit": "kWh"},
+        )
+        db.session.add(charger)
+        db.session.flush()
+        for j, unit in enumerate(["kW", "kWh"], start=1):
+            db.session.add(
+                Sensor(
+                    name=f"charger {i} sensor {j}",
+                    generic_asset=charger,
+                    event_resolution=timedelta(minutes=15),
+                    unit=unit,
+                )
+            )
+    db.session.flush()
+
+    original_flex_context = house.flex_context.copy()
+
+    # --- Act ---
+    house_copy = copy_asset(house, account=supplier_account)
+
+    # 1. Correct account and name.
+    assert house_copy.account_id == supplier_account.id
+    assert house_copy.name == f"{house.name} (Copy)"
+
+    # 2. Top-level in the target account (no parent given → parent_asset_id = None).
+    assert house_copy.parent_asset_id is None
+
+    # 3. flex_context is preserved verbatim.
+    assert house_copy.flex_context == original_flex_context
+
+    # 4. Shallow copy: the original children have *not* been duplicated.
+    children_of_copy = db.session.scalars(
+        select(GenericAsset).filter_by(parent_asset_id=house_copy.id)
+    ).all()
+    assert len(children_of_copy) == 0, "copy_asset should not recursively copy children"

From 5371a4e7184f3b64fecb9f5db7fb4f6aafaea53a Mon Sep 17 00:00:00 2001
From: joshuaunity <oghenerobojosh01@gmail.com>
Date: Mon, 16 Mar 2026 15:34:14 +0100
Subject: [PATCH 8/9] chore: move logic into @post_load

Signed-off-by: joshuaunity <oghenerobojosh01@gmail.com>
---
 flexmeasures/api/v3_0/assets.py           | 59 ++++++++++++++++-------
 flexmeasures/ui/static/openapi-specs.json |  2 +-
 2 files changed, 42 insertions(+), 19 deletions(-)

diff --git a/flexmeasures/api/v3_0/assets.py b/flexmeasures/api/v3_0/assets.py
index 9944cbabd..12833749d 100644
--- a/flexmeasures/api/v3_0/assets.py
+++ b/flexmeasures/api/v3_0/assets.py
@@ -1,4 +1,6 @@
 from __future__ import annotations
+
+from typing import Any
 import json
 from datetime import datetime, timedelta
 from http import HTTPStatus
@@ -201,6 +203,17 @@ def resolve_account_and_parent(self, data, **kwargs):
         elif parent_given and not account_given:
             data["account"] = data["parent_asset"].owner
 
+        # Resolve effective targets for permission checks and fallback behavior.
+        # If neither target is given, use the source asset's account/parent.
+        source_asset: GenericAsset | None = self.context.get("asset")
+        if source_asset is not None:
+            if data.get("account") is None:
+                data["resolved_account"] = source_asset.owner
+                data["resolved_parent"] = source_asset.parent_asset
+            else:
+                data["resolved_account"] = data["account"]
+                data["resolved_parent"] = data.get("parent_asset")
+
         return data
 
 
@@ -346,13 +359,14 @@ def index(
             check_access(account, "read")
             account_ids = [account.id]
         else:
-            use_all_accounts = all_accessible or root_asset
-            include_public = all_accessible or include_public or root_asset
-            account_ids = (
-                [a.id for a in get_accessible_accounts()]
-                if use_all_accounts
-                else [current_user.account.id]
+            use_all_accounts = all_accessible or (root_asset is not None)
+            include_public = (
+                all_accessible or include_public or (root_asset is not None)
             )
+            if use_all_accounts:
+                account_ids = [a.id for a in get_accessible_accounts()]
+            else:
+                account_ids = [current_user.account.id]
         filter_statement = GenericAsset.account_id.in_(account_ids)
         if include_public:
             filter_statement = filter_statement | GenericAsset.account_id.is_(None)
@@ -486,6 +500,9 @@ def asset_sensors(
           tags:
             - Assets
         """
+        if asset is None:
+            return unprocessable_entity("No asset found for the given id.")
+
         query_statement = Sensor.generic_asset_id == asset.id
 
         query = select(Sensor).filter(query_statement)
@@ -1590,24 +1607,30 @@ def get_kpis(self, id: int, asset: GenericAsset, start, end):
 
     @route("/<id>/copy", methods=["POST"])
     @use_kwargs(
-        {"asset": AssetIdField(data_key="id", status_if_not_found=HTTPStatus.NOT_FOUND)},
+        {
+            "asset": AssetIdField(
+                data_key="id", status_if_not_found=HTTPStatus.NOT_FOUND
+            )
+        },
         location="path",
     )
-    @use_kwargs(CopyAssetSchema, location="query")
     @as_json
-    def copy_assets(self, id, asset: GenericAsset, account=None, parent_asset=None):
+    def copy_assets(self, id, asset: GenericAsset):
         """
         .. :quickref: Assets; Copy an asset to a target account and/or parent.
         """
-        # Resolve the target account and parent for permission checking.
-        # When neither is given (account is None), the copy is placed in the
-        # same account and under the same parent as the original.
-        if account is None:
-            resolved_account = asset.owner
-            resolved_parent = asset.parent_asset
-        else:
-            resolved_account = account
-            resolved_parent = parent_asset
+        copy_asset_schema: Any = CopyAssetSchema()
+        copy_asset_schema.context["asset"] = asset
+
+        try:
+            copy_data = copy_asset_schema.load(request.args)
+        except ValidationError as e:
+            return unprocessable_entity(str(e.messages))
+
+        account = copy_data.get("account")
+        parent_asset = copy_data.get("parent_asset")
+        resolved_account = copy_data["resolved_account"]
+        resolved_parent = copy_data["resolved_parent"]
 
         # Check create-children permission on the target account.
         check_access(resolved_account, "create-children")
diff --git a/flexmeasures/ui/static/openapi-specs.json b/flexmeasures/ui/static/openapi-specs.json
index 31f1750f6..cb6ce0349 100644
--- a/flexmeasures/ui/static/openapi-specs.json
+++ b/flexmeasures/ui/static/openapi-specs.json
@@ -2648,7 +2648,7 @@
         ]
       }
     },
-    "/api/v3_0/assets/copy": {},
+    "/api/v3_0/assets/{id}/copy": {},
     "/api/v3_0/assets/{id}": {
       "delete": {
         "summary": "Delete an asset.",

From e84899e7bfa9cf30455c63c1e51fcf355f6b0384 Mon Sep 17 00:00:00 2001
From: joshuaunity <oghenerobojosh01@gmail.com>
Date: Tue, 17 Mar 2026 12:39:08 +0100
Subject: [PATCH 9/9] feat: implement deep copy of asset subtree including
 direct sensors

Signed-off-by: joshuaunity <oghenerobojosh01@gmail.com>
---
 flexmeasures/api/common/utils/api_utils.py    | 115 +++++++++++++-----
 .../api/v3_0/tests/test_assets_api.py         |  54 ++++++--
 2 files changed, 134 insertions(+), 35 deletions(-)

diff --git a/flexmeasures/api/common/utils/api_utils.py b/flexmeasures/api/common/utils/api_utils.py
index 1e56c6cf3..3b8ced436 100644
--- a/flexmeasures/api/common/utils/api_utils.py
+++ b/flexmeasures/api/common/utils/api_utils.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from copy import deepcopy
 import json
 from timely_beliefs.beliefs.classes import BeliefsDataFrame
 from typing import Sequence
@@ -16,6 +17,7 @@
 from flexmeasures.data import db
 from flexmeasures.data.models.user import Account
 from flexmeasures.data.models.generic_assets import GenericAsset
+from flexmeasures.data.models.time_series import Sensor
 from flexmeasures.data.utils import save_to_db
 from flexmeasures.auth.policy import check_access
 from flexmeasures.api.common.responses import (
@@ -210,13 +212,81 @@ def convert_asset_json_fields(asset_kwargs):
     return asset_kwargs
 
 
+def _copy_direct_sensors(
+    source_asset: GenericAsset, copied_asset: GenericAsset
+) -> None:
+    """Copy sensors directly attached to one asset."""
+    source_sensors = db.session.scalars(
+        select(Sensor).filter(Sensor.generic_asset_id == source_asset.id)
+    ).all()
+    for source_sensor in source_sensors:
+        sensor_kwargs = {}
+        for column in source_sensor.__table__.columns:
+            if column.name in [
+                "id",
+                "generic_asset_id",
+                "knowledge_horizon_fnc",
+                "knowledge_horizon_par",
+            ]:
+                continue
+            sensor_kwargs[column.name] = deepcopy(getattr(source_sensor, column.name))
+
+        sensor_kwargs["generic_asset_id"] = copied_asset.id
+
+        db.session.add(Sensor(**sensor_kwargs))
+
+
+def _copy_asset_subtree(
+    source_asset: GenericAsset,
+    destination_account_id: int,
+    destination_parent_asset_id: int | None,
+    asset_schema: AssetSchema,
+) -> GenericAsset:
+    """Recursively copy one asset and all descendants."""
+    asset_kwargs = asset_schema.dump(source_asset)
+
+    for key in ["id", "owner", "generic_asset_type", "child_assets", "sensors"]:
+        asset_kwargs.pop(key, None)
+
+    asset_kwargs["name"] = f"{asset_kwargs['name']} (Copy)"
+    asset_kwargs["account_id"] = destination_account_id
+    asset_kwargs["parent_asset_id"] = destination_parent_asset_id
+    asset_kwargs = convert_asset_json_fields(asset_kwargs)
+
+    copied_asset = GenericAsset(**asset_kwargs)
+    db.session.add(copied_asset)
+    db.session.flush()
+
+    _copy_direct_sensors(source_asset, copied_asset)
+
+    source_children = db.session.scalars(
+        select(GenericAsset)
+        .filter(GenericAsset.parent_asset_id == source_asset.id)
+        .order_by(GenericAsset.id)
+    ).all()
+    for source_child in source_children:
+        _copy_asset_subtree(
+            source_asset=source_child,
+            destination_account_id=destination_account_id,
+            destination_parent_asset_id=copied_asset.id,
+            asset_schema=asset_schema,
+        )
+
+    return copied_asset
+
+
 def copy_asset(
     asset: GenericAsset,
     account=None,
     parent_asset=None,
 ) -> GenericAsset:
     """
-    Copy a single asset to a target account and/or under a target parent asset.
+    Copy an asset subtree to a target account and/or under a target parent asset.
+
+    The copied subtree includes:
+    - the selected asset
+    - all descendant child assets (recursively)
+    - all sensors directly attached to each copied asset
 
     Resolution rules:
 
@@ -232,39 +302,28 @@ def copy_asset(
     """
     try:
         asset_schema = AssetSchema()
-        asset_kwargs = asset_schema.dump(asset)
 
-        # Remove dump_only and read-only fields
-        for key in ["id", "owner", "generic_asset_type", "child_assets", "sensors"]:
-            asset_kwargs.pop(key, None)
-
-        # Avoid name collision
-        asset_kwargs["name"] = f"{asset_kwargs['name']} (Copy)"
-
-        # Resolve target account and parent
         if account is None and parent_asset is None:
-            # Neither given: preserve the original asset's placement
-            asset_kwargs["account_id"] = asset.account_id
-            asset_kwargs["parent_asset_id"] = asset.parent_asset_id
+            target_account_id = int(asset.account_id)
+            target_parent_asset_id = asset.parent_asset_id
         elif account is not None and parent_asset is None:
-            # Only account given: top-level asset in the target account
-            asset_kwargs["account_id"] = account.id
-            asset_kwargs["parent_asset_id"] = None
+            target_account_id = int(account.id)
+            target_parent_asset_id = None
         elif account is None and parent_asset is not None:
-            # Only parent given: inherit the parent's account
-            asset_kwargs["account_id"] = parent_asset.account_id
-            asset_kwargs["parent_asset_id"] = parent_asset.id
+            target_account_id = int(parent_asset.account_id)
+            target_parent_asset_id = int(parent_asset.id)
         else:
-            # Both given: explicit placement, possibly cross-account
-            asset_kwargs["account_id"] = account.id
-            asset_kwargs["parent_asset_id"] = parent_asset.id
-
-        asset_kwargs = convert_asset_json_fields(asset_kwargs)
-
-        new_asset = GenericAsset(**asset_kwargs)
-        db.session.add(new_asset)
+            target_account_id = int(account.id)
+            target_parent_asset_id = int(parent_asset.id)
+
+        copied_root = _copy_asset_subtree(
+            source_asset=asset,
+            destination_account_id=target_account_id,
+            destination_parent_asset_id=target_parent_asset_id,
+            asset_schema=asset_schema,
+        )
         db.session.commit()
-        return new_asset
+        return copied_root
     except Exception as e:
         db.session.rollback()
         raise e
diff --git a/flexmeasures/api/v3_0/tests/test_assets_api.py b/flexmeasures/api/v3_0/tests/test_assets_api.py
index 127a6e8c4..669b95e31 100644
--- a/flexmeasures/api/v3_0/tests/test_assets_api.py
+++ b/flexmeasures/api/v3_0/tests/test_assets_api.py
@@ -6,7 +6,7 @@
 from sqlalchemy import select, func
 
 from flexmeasures.data.models.audit_log import AssetAuditLog
-from flexmeasures.data.models.generic_assets import GenericAsset, GenericAssetType
+from flexmeasures.data.models.generic_assets import GenericAsset
 from flexmeasures.data.models.time_series import Sensor
 from flexmeasures.data.services.users import find_user_by_email
 from flexmeasures.api.tests.utils import get_auth_token, UserContext, AccountContext
@@ -692,7 +692,7 @@ def test_consultant_get_asset(
 
 def test_copy_asset(setup_api_test_data, setup_accounts, db):
     """
-    Test all four placement use cases for copy_asset:
+    Test all four placement use cases for copy_asset and verify direct sensors are copied.
 
     1. Neither account nor parent given  → same account, same parent (sibling copy).
     2. Only account given                → top-level asset in the given account.
@@ -717,6 +717,16 @@ def test_copy_asset(setup_api_test_data, setup_accounts, db):
     assert battery is not None, "Battery asset must exist in Prosumer account"
     assert turbine is not None, "Wind turbine asset must exist in Supplier account"
 
+    # Add a deterministic sensor on battery so we can verify deep copy behavior.
+    source_sensor = Sensor(
+        name="copy-source-sensor",
+        generic_asset=battery,
+        event_resolution=timedelta(minutes=15),
+        unit="kW",
+    )
+    db.session.add(source_sensor)
+    db.session.flush()
+
     # Create a parent asset in the Supplier account for use cases 3 and 4.
     parent = GenericAsset(
         name="Test parent for copy",
@@ -732,6 +742,16 @@ def test_copy_asset(setup_api_test_data, setup_accounts, db):
     assert copy1.account_id == battery.account_id
     assert copy1.parent_asset_id == battery.parent_asset_id  # None
 
+    copied_sensor = db.session.scalars(
+        select(Sensor).filter(
+            Sensor.generic_asset_id == copy1.id,
+            Sensor.name == source_sensor.name,
+        )
+    ).first()
+    assert copied_sensor is not None
+    assert copied_sensor.unit == source_sensor.unit
+    assert copied_sensor.event_resolution == source_sensor.event_resolution
+
     # 2. Only account given → top-level in target account
     # Use the turbine so the name doesn't clash with copy1 (parent_asset_id is None for both).
     copy2 = copy_asset(turbine, account=prosumer_account)
@@ -777,8 +797,8 @@ def test_copy_asset_fails_on_duplicate_name_under_same_parent(
     # Create a dedicated parent so this test is independent of others.
     parent = GenericAsset(
         name="Test parent for duplicate-name failure",
-        generic_asset_type=battery.generic_asset_type,
-        owner=prosumer_account,
+        generic_asset_type_id=battery.generic_asset_type_id,
+        account_id=prosumer_account.id,
     )
     db.session.add(parent)
     db.session.flush()
@@ -817,7 +837,8 @@ def test_copy_asset_to_another_account_preserves_config(
     1. The copy lands in the Supplier account with the expected name.
     2. The copy is a top-level asset (no parent).
     3. flex_context is preserved verbatim (sensor IDs are unchanged).
-    4. copy_asset is a *shallow* copy: the original child assets are not duplicated.
+    4. copy_asset performs a deep subtree copy: child assets are duplicated.
+    5. Sensors on copied child assets are duplicated.
     """
     prosumer_account = setup_accounts["Prosumer"]
     supplier_account = setup_accounts["Supplier"]
@@ -890,8 +911,27 @@ def test_copy_asset_to_another_account_preserves_config(
     # 3. flex_context is preserved verbatim.
     assert house_copy.flex_context == original_flex_context
 
-    # 4. Shallow copy: the original children have *not* been duplicated.
+    # 4. Direct sensors on the copied asset are duplicated.
+    copied_house_sensors = db.session.scalars(
+        select(Sensor).filter(Sensor.generic_asset_id == house_copy.id)
+    ).all()
+    assert len(copied_house_sensors) == 1
+    assert copied_house_sensors[0].name == site_capacity_sensor.name
+    assert copied_house_sensors[0].unit == site_capacity_sensor.unit
+    assert (
+        copied_house_sensors[0].event_resolution
+        == site_capacity_sensor.event_resolution
+    )
+
+    # 5. Deep copy for hierarchy: original child assets are duplicated.
     children_of_copy = db.session.scalars(
         select(GenericAsset).filter_by(parent_asset_id=house_copy.id)
     ).all()
-    assert len(children_of_copy) == 0, "copy_asset should not recursively copy children"
+    assert len(children_of_copy) == 2
+
+    # 6. Sensors on copied child assets are also duplicated.
+    copied_child_ids = [child.id for child in children_of_copy]
+    copied_child_sensors = db.session.scalars(
+        select(Sensor).filter(Sensor.generic_asset_id.in_(copied_child_ids))
+    ).all()
+    assert len(copied_child_sensors) == 4