[CDF-26937] Events-to-records migration part 2 (#2767)

# Description

Prepares the migration infrastructure to support targeting
Records/Streams in addition to FDM instances. Generalizes
`create_properties()` to accept `ContainerPropertyDefinition` alongside
`ViewResponseProperty`, and makes the `container_id` parameter optional
in `_prepare_asset_centric_arguments`, since this will not be used for
records.

I also changed my mind from previously and have made
`iter_last_updated_time_windows` to `last_updated_time_windows`
(iterator → list), since I realize it would have needed to be wrapped in
list() anyways.

## Bump

- [ ] Patch
- [x] Skip

---------

Co-authored-by: Magnus Schjølberg <magnus.schjolberg@cognitedata.com>

Author: Magssch

cognite_toolkit/_cdf_tk/apps/_migrate_app.py

@@ -45,6 +45,7 @@
     MigrateDataSetSelector,
     MigrationCSVFileSelector,
 )
+from cognite_toolkit._cdf_tk.exceptions import ToolkitValueError
 from cognite_toolkit._cdf_tk.feature_flags import Flags
 from cognite_toolkit._cdf_tk.storageio import CanvasIO, ChartIO, InstanceIO
 from cognite_toolkit._cdf_tk.storageio.selectors import (
@@ -410,7 +411,7 @@ def _prepare_asset_centric_arguments(
         skip_existing: bool,
         kind: AssetCentricKind,
         resource_type: str,
-        container_id: ContainerId,
+        container_id: ContainerId | None = None,
     ) -> tuple[AssetCentricMigrationSelector, bool, bool, bool]:
         if data_set_id is not None and mapping_file is not None:
             raise typer.BadParameter("Cannot specify both data_set_id and mapping_file")
@@ -444,6 +445,11 @@ def _prepare_asset_centric_arguments(
             )
         else:
             # Interactive selection of data set.
+            if container_id is None:
+                raise ToolkitValueError(
+                    "container_id is required when neither mapping_file nor data_set_id is provided "
+                    "(interactive migration)."
+                )
             selector = AssetInteractiveSelect(client, "migrate")
             selected_data_set_id = selector.select_data_set(allow_empty=False)
             asset_mapping = ResourceViewMappingInteractiveSelect(client, "migrate").select_resource_view_mapping(

cognite_toolkit/_cdf_tk/commands/_migrate/conversion.py

@@ -8,10 +8,17 @@
 from pydantic import JsonValue
 
 from cognite_toolkit._cdf_tk.client import ToolkitClient
-from cognite_toolkit._cdf_tk.client.identifiers import AssetCentricExternalId, EdgeTypeId, ExternalId, InternalId
+from cognite_toolkit._cdf_tk.client.identifiers import (
+    AssetCentricExternalId,
+    ContainerId,
+    EdgeTypeId,
+    ExternalId,
+    InternalId,
+)
 from cognite_toolkit._cdf_tk.client.resource_classes.annotation import AnnotationResponse, AssetLinkData, FileLinkData
 from cognite_toolkit._cdf_tk.client.resource_classes.asset import AssetResponse
 from cognite_toolkit._cdf_tk.client.resource_classes.data_modeling import (
+    ContainerPropertyDefinition,
     DirectNodeRelation,
     EdgeId,
     EdgeProperty,
@@ -318,50 +325,64 @@ def _lookup_resource_type(resource_type: AssetCentricResourceExtended) -> AssetC
 
 def create_properties(
     dumped: dict[str, Any],
-    view_properties: dict[str, ViewResponseProperty],
+    properties: dict[str, ViewResponseProperty] | dict[str, ContainerPropertyDefinition],
     property_mapping: dict[str, str],
     resource_type: AssetCentricTypeExtended,
     issue: ConversionIssue,
     direct_relation_cache: DirectRelationCache,
+    container_id: ContainerId | None = None,
 ) -> dict[str, JsonValue | NodeId | list[NodeId]]:
     """
-    Create properties for a data model instance from an asset-centric resource.
+    Create properties for a data model instance or record from an asset-centric resource.
+
+    When ``container_id`` is provided the values in ``properties`` are treated as
+    ``ContainerPropertyDefinition`` objects (records/streams path).  Otherwise they must be
+    ``ViewResponseProperty`` objects (FDM/view path).
 
     Args:
         dumped: Dict representation of the asset-centric resource.
-        view_properties: Defined properties referenced in the view source mapping.
-        property_mapping:  Mapping from asset-centric property IDs to data model property IDs.
+        properties: Defined properties referenced in the mapping.
+        property_mapping: Mapping from asset-centric property JSON-paths to target property IDs.
         resource_type: The type of the asset-centric resource (e.g., "asset", "timeseries").
         issue: ConversionIssue object to log any issues encountered during conversion.
-        direct_relation_cache: Cache for direct relation references to look up target of direct relations.
+        direct_relation_cache: Cache for direct relation references.
+        container_id: Target container ID, only used if container properties are being populated
 
     Returns:
-        Dict of property IDs to PropertyValueWrite objects.
+        Dict of property IDs to property values.
 
     """
     flatten_dump = flatten_dict_json_path(dumped, keep_structured=set(property_mapping.keys()))
-    properties: dict[str, JsonValue | NodeId | list[NodeId]] = {}
+    result: dict[str, JsonValue | NodeId | list[NodeId]] = {}
     ignored_asset_centric_properties: set[str] = set()
     for prop_json_path, prop_id in property_mapping.items():
         if prop_json_path not in flatten_dump:
             continue
-        if prop_id not in view_properties:
+        if prop_id not in properties:
             continue
-        if prop_id in properties:
+        if prop_id in result:
             ignored_asset_centric_properties.add(prop_json_path)
             continue
-        dm_prop = view_properties[prop_id]
-        if not isinstance(dm_prop, ViewCorePropertyResponse):
+        prop_def = properties[prop_id]
+        if isinstance(prop_def, ContainerPropertyDefinition):
+            if container_id is None:
+                raise ValueError("Cannot create mapping directly to container property without providing Container ID")
+            destination_container_property = (container_id, prop_id)
+        elif isinstance(prop_def, ViewCorePropertyResponse):
+            destination_container_property = (prop_def.container, prop_def.container_property_identifier)
+        else:
             issue.invalid_instance_property_types.append(
                 InvalidPropertyDataType(property_id=prop_id, expected_type=ViewCorePropertyResponse.__name__)
             )
             continue
+        data_type = prop_def.type
+        nullable = prop_def.nullable or False
         try:
             value = asset_centric_convert_to_primary_property(
                 flatten_dump[prop_json_path],
-                dm_prop.type,
-                dm_prop.nullable or False,
-                destination_container_property=(dm_prop.container, dm_prop.container_property_identifier),
+                data_type,
+                nullable,
+                destination_container_property=destination_container_property,
                 source_property=(resource_type, prop_json_path),
                 direct_relation_lookup=direct_relation_cache.get_cache(resource_type, prop_json_path),
             )
@@ -372,12 +393,12 @@ def create_properties(
             continue
         if isinstance(value, datetime):
             # Convert datetime to ISO format string, as the data model expects datetimes as strings in ISO format.
-            properties[prop_id] = value.isoformat(timespec="milliseconds")
+            result[prop_id] = value.isoformat(timespec="milliseconds")
         elif isinstance(value, date):
             # Convert date to ISO format string, as the data model expects dates as strings in ISO format.
-            properties[prop_id] = value.isoformat()
+            result[prop_id] = value.isoformat()
         else:
-            properties[prop_id] = value
+            result[prop_id] = value
 
     issue.ignored_asset_centric_properties = sorted(
         (set(flatten_dump.keys()) - set(property_mapping.keys())) | ignored_asset_centric_properties
@@ -390,9 +411,9 @@ def create_properties(
             for prop_id in property_mapping.values()
             if not (prop_id.startswith("edge.") or prop_id.startswith("node."))
         }
-        - set(view_properties.keys())
+        - set(properties.keys())
     )
-    return properties
+    return result
 
 
 def create_edge_properties(

cognite_toolkit/_cdf_tk/cruds/_resource_cruds/streams.py

@@ -1,5 +1,5 @@
 import time
-from collections.abc import Hashable, Iterable, Iterator, Sequence
+from collections.abc import Hashable, Iterable, Sequence
 from datetime import timedelta
 from typing import Any, Literal, final
 
@@ -87,34 +87,34 @@ def _iterate(
         all_streams = self.client.streams.list()
         return iter(all_streams)
 
-    def iter_last_updated_time_windows(
+    def last_updated_time_windows(
         self, stream_external_id: str, start_ms: int | None = None
-    ) -> Iterator[dict[str, int] | None]:
-        """Yield lastUpdatedTime filter dicts to use in record queries.
+    ) -> list[dict[str, int] | None]:
+        """Return lastUpdatedTime filter dicts to use in record queries.
 
-        Each yielded dict is {"gte": ..., "lt": ...} representing one query window.
-        None is yielded for Mutable streams with no start_ms — meaning no time filter is needed.
-        Yields nothing if the stream does not exist.
+        Each dict is {"gte": ..., "lt": ...} representing one query window.
+        None is returned for Mutable streams with no start_ms — meaning no time filter is needed.
+        Returns an empty list if the stream does not exist.
 
         Immutable streams enforce a maxFilteringInterval per request, so the range
         [start_ms (or stream.createdTime), now) is split into consecutive windows.
         Mutable streams have no such constraint and are covered in a single pass.
         """
         streams = self.retrieve(ExternalId.from_external_ids([stream_external_id]))
         if not streams:
-            return
+            return []
         stream = streams[0]
         now_ms = int(time.time() * 1000)
         if stream.type == "Mutable":
             if start_ms is None:
-                yield None
-            else:
-                yield {"gte": start_ms, "lt": now_ms}
-            return
+                return [None]
+            return [{"gte": start_ms, "lt": now_ms}]
         effective_start_ms = start_ms if start_ms is not None else stream.created_time
         max_interval_ms: int | None = None
         if stream.settings and stream.settings.limits.max_filtering_interval:
             td = _TIMEDELTA_ADAPTER.validate_python(stream.settings.limits.max_filtering_interval)
             max_interval_ms = int(td.total_seconds() * 1000)
-        for window_start, window_end in time_windows_ms(effective_start_ms, now_ms, max_interval_ms):
-            yield {"gte": window_start, "lt": window_end}
+        return [
+            {"gte": window_start, "lt": window_end}
+            for window_start, window_end in time_windows_ms(effective_start_ms, now_ms, max_interval_ms)
+        ]

cognite_toolkit/_cdf_tk/storageio/_records.py

@@ -58,9 +58,7 @@ def count(self, selector: RecordContainerSelector) -> int | None:
         start_ms = timestamp_to_ms(selector.initialize_cursor)
         total = 0
         stream_crud = StreamCRUD.create_loader(self.client)
-        for last_updated_time in stream_crud.iter_last_updated_time_windows(
-            selector.stream.external_id, start_ms=start_ms
-        ):
+        for last_updated_time in stream_crud.last_updated_time_windows(selector.stream.external_id, start_ms=start_ms):
             body: dict[str, object] = {
                 "filter": sync_filter,
                 "aggregates": {"total": {"count": {}}},

tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/test_conversion.py

@@ -15,6 +15,7 @@
     BooleanProperty,
     ConstraintOrIndexState,
     ContainerId,
+    ContainerPropertyDefinition,
     DateProperty,
     DirectNodeRelation,
     EdgeId,
@@ -119,7 +120,7 @@ class TestCreateProperties:
     EVENT_CENTRIC_ID = AssetCentricId(resource_type="event", id_=456)
 
     @pytest.mark.parametrize(
-        "dumped,view_properties,property_mapping,expected_properties,expected_issue",
+        "dumped,property_definitions,property_mapping,expected_properties,expected_issue,container_id",
         [
             pytest.param(
                 {"name": "MyAsset", "description": "An asset,", "metadata": {"categoryNo": "1"}},
@@ -146,8 +147,37 @@ class TestCreateProperties:
                 {"name": "nameId", "description": "descriptionId", "metadata.categoryNo": "categoryNoId"},
                 {"nameId": "MyAsset", "descriptionId": "An asset,", "categoryNoId": 1},
                 ConversionIssue(id=str(ASSET_CENTRIC_ID), instance_id=INSTANCE_ID, asset_centric_id=ASSET_CENTRIC_ID),
+                None,
                 id="Basic property mapping with integer conversion and no issues",
             ),
+            pytest.param(
+                {"name": "MyAsset", "description": "An asset,", "metadata": {"categoryNo": "1"}},
+                {
+                    "nameId": ContainerPropertyDefinition(
+                        type=TextProperty(),
+                        nullable=True,
+                        immutable=False,
+                        auto_increment=False,
+                    ),
+                    "descriptionId": ContainerPropertyDefinition(
+                        type=TextProperty(),
+                        nullable=True,
+                        immutable=False,
+                        auto_increment=False,
+                    ),
+                    "categoryNoId": ContainerPropertyDefinition(
+                        type=Int64Property(),
+                        nullable=True,
+                        immutable=False,
+                        auto_increment=False,
+                    ),
+                },
+                {"name": "nameId", "description": "descriptionId", "metadata.categoryNo": "categoryNoId"},
+                {"nameId": "MyAsset", "descriptionId": "An asset,", "categoryNoId": 1},
+                ConversionIssue(id=str(ASSET_CENTRIC_ID), instance_id=INSTANCE_ID, asset_centric_id=ASSET_CENTRIC_ID),
+                CONTAINER_ID,
+                id="Basic property mapping with container property definitions",
+            ),
             pytest.param(
                 {"name": "MyAsset", "created": "2023-01-01T12:00:00Z", "active": True},
                 {
@@ -177,6 +207,7 @@ class TestCreateProperties:
                     "activeId": True,
                 },
                 ConversionIssue(id=str(ASSET_CENTRIC_ID), asset_centric_id=ASSET_CENTRIC_ID, instance_id=INSTANCE_ID),
+                None,
                 id="Multiple data types conversion",
             ),
             pytest.param(
@@ -203,6 +234,7 @@ class TestCreateProperties:
                     instance_id=INSTANCE_ID,
                     missing_asset_centric_properties=["description"],
                 ),
+                None,
                 id="Missing property in flattened dump",
             ),
             pytest.param(
@@ -223,6 +255,7 @@ class TestCreateProperties:
                     instance_id=INSTANCE_ID,
                     missing_instance_properties=["descriptionId"],
                 ),
+                None,
                 id="Missing property in view properties",
             ),
             pytest.param(
@@ -247,6 +280,7 @@ class TestCreateProperties:
                         InvalidPropertyDataType(property_id="nameId", expected_type="ViewCorePropertyResponse")
                     ],
                 ),
+                None,
                 id="Invalid property type",
             ),
             pytest.param(
@@ -271,6 +305,7 @@ class TestCreateProperties:
                         )
                     ],
                 ),
+                None,
                 id="Conversion error",
             ),
             pytest.param(
@@ -291,6 +326,7 @@ class TestCreateProperties:
                     instance_id=INSTANCE_ID,
                     missing_asset_centric_properties=["name"],
                 ),
+                None,
                 id="Empty dictionary",
             ),
             pytest.param(
@@ -316,6 +352,7 @@ class TestCreateProperties:
                     asset_centric_id=ASSET_CENTRIC_ID,
                     instance_id=INSTANCE_ID,
                 ),
+                None,
                 id="List of simple types (labels to list of strings)",
             ),
             pytest.param(
@@ -352,6 +389,7 @@ class TestCreateProperties:
                     instance_id=INSTANCE_ID,
                     ignored_asset_centric_properties=["labels[1].externalId"],
                 ),
+                None,
                 id="Mapping the first label and entire metadata.",
             ),
             pytest.param(
@@ -382,6 +420,7 @@ class TestCreateProperties:
                     instance_id=INSTANCE_ID,
                     ignored_asset_centric_properties=["metadata.category"],
                 ),
+                None,
                 id="Duplicated mapping target",
             ),
             pytest.param(
@@ -410,24 +449,34 @@ class TestCreateProperties:
                     asset_centric_id=ASSET_CENTRIC_ID,
                     instance_id=INSTANCE_ID,
                 ),
+                None,
                 id="Japanese characters in property names and values",
             ),
         ],
     )
     def test_create_properties(
         self,
         dumped: dict[str, Any],
-        view_properties: dict[str, ViewResponseProperty],
+        property_definitions: dict[str, ViewResponseProperty] | dict[str, ContainerPropertyDefinition],
         property_mapping: dict[str, str],
         expected_properties: dict[str, JsonValue],
         expected_issue: ConversionIssue,
+        container_id: ContainerId | None,
         direct_relation_cache: DirectRelationCache,
     ) -> None:
         issue = ConversionIssue(
             asset_centric_id=self.ASSET_CENTRIC_ID, instance_id=self.INSTANCE_ID, id=str(self.ASSET_CENTRIC_ID)
         )
 
-        properties = create_properties(dumped, view_properties, property_mapping, "asset", issue, direct_relation_cache)
+        properties = create_properties(
+            dumped,
+            property_definitions,
+            property_mapping,
+            "asset",
+            issue,
+            direct_relation_cache,
+            container_id,
+        )
 
         assert properties == expected_properties