[PYG-259] 🌑 Skip read-only properties (#364)

* refactor; store container information

* fix: not include read-only properties in write class

* refactor: regen

* fix: other case

* refactor: regen

cognite/pygen/_constants.py

@@ -10,5 +10,5 @@
 }
 
 
-def is_readonly_property(prop: dm.MappedProperty) -> bool:
-    return prop.container in _READONLY_PROPERTIES and prop.name in _READONLY_PROPERTIES[prop.container]
+def is_readonly_property(container: dm.ContainerId, identifier: str) -> bool:
+    return container in _READONLY_PROPERTIES and identifier in _READONLY_PROPERTIES[container]

cognite/pygen/_core/models/fields/__init__.py

@@ -11,7 +11,7 @@
     OneToManyConnectionField,
     OneToOneConnectionField,
 )
-from .primitive import BasePrimitiveField, PrimitiveField, PrimitiveListField
+from .primitive import BasePrimitiveField, ContainerProperty, PrimitiveField, PrimitiveListField
 
 __all__ = [
     "Field",
@@ -26,4 +26,5 @@
     "BaseConnectionField",
     "OneToOneConnectionField",
     "OneToManyConnectionField",
+    "ContainerProperty",
 ]

cognite/pygen/_core/models/fields/cdf_reference.py

@@ -11,7 +11,7 @@
 from cognite.client.data_classes.data_modeling.data_types import ListablePropertyType
 
 from .base import Field
-from .primitive import BasePrimitiveField, PrimitiveListField
+from .primitive import BasePrimitiveField, ContainerProperty, PrimitiveListField
 
 
 @dataclass(frozen=True)
@@ -104,6 +104,7 @@ def as_value(self) -> str:
     def load(cls, base: Field, prop: dm.MappedProperty, variable: str) -> CDFExternalField | None:
         if not isinstance(prop.type, dm.CDFExternalIdReference):
             return None
+        container = ContainerProperty(prop.container, prop.container_property_identifier)
         if prop.type.is_list:
             return CDFExternalListField(
                 name=base.name,
@@ -114,6 +115,7 @@ def load(cls, base: Field, prop: dm.MappedProperty, variable: str) -> CDFExterna
                 description=prop.description,
                 pydantic_field=base.pydantic_field,
                 variable=variable,
+                container=container,
             )
         else:
             return CDFExternalField(
@@ -124,6 +126,7 @@ def load(cls, base: Field, prop: dm.MappedProperty, variable: str) -> CDFExterna
                 is_nullable=prop.nullable,
                 description=prop.description,
                 pydantic_field=base.pydantic_field,
+                container=container,
             )
 
 

cognite/pygen/_core/models/fields/connections.py

@@ -16,9 +16,11 @@
     SingleReverseDirectRelation,
 )
 
+from cognite.pygen._constants import is_readonly_property
 from cognite.pygen._warnings import MissingReverseDirectRelationTargetWarning
 
 from .base import Field
+from .primitive import ContainerProperty
 
 if TYPE_CHECKING:
     from cognite.pygen._core.models.data_classes import EdgeDataClass, NodeDataClass
@@ -145,6 +147,7 @@ class BaseConnectionField(Field, ABC):
     through: dm.PropertyId | None
     destination_class: NodeDataClass | None
     edge_class: EdgeDataClass | None
+    container: ContainerProperty | None
 
     @property
     def linked_class(self) -> NodeDataClass | EdgeDataClass:
@@ -185,7 +188,9 @@ def is_reverse_direct_relation(self) -> bool:
     @property
     def is_write_field(self) -> bool:
         """Returns True if the connection is writable."""
-        return not self.is_reverse_direct_relation
+        return not self.is_reverse_direct_relation and not (
+            self.container and is_readonly_property(self.container.source, self.container.identifier)
+        )
 
     @property
     def is_edge(self) -> bool:
@@ -261,12 +266,14 @@ def load(
                     MissingReverseDirectRelationTargetWarning(prop.through, view_id, base.prop_name), stacklevel=2
                 )
                 return None
+        container: ContainerProperty | None = None
         edge_type = prop.type if isinstance(prop, dm.EdgeConnection) else None
         direction: Literal["outwards", "inwards"]
         if isinstance(prop, dm.EdgeConnection):
             direction = prop.direction
         elif isinstance(prop, dm.MappedProperty):
             direction = "outwards"
+            container = ContainerProperty(prop.container, prop.container_property_identifier)
         elif isinstance(prop, ReverseDirectRelation):
             direction = "inwards"
         else:
@@ -299,6 +306,7 @@ def load(
                 type_hint_node_reference=type_hint_node_reference,
                 destination_class=destination_class,
                 edge_class=edge_class,
+                container=container,
             )
         elif cls._is_supported_one_to_one_connection(prop):
             return OneToOneConnectionField(
@@ -313,6 +321,7 @@ def load(
                 type_hint_node_reference=type_hint_node_reference,
                 destination_class=destination_class,
                 edge_class=edge_class,
+                container=container,
             )
         else:
             return None

cognite/pygen/_core/models/fields/primitive.py

@@ -10,9 +10,17 @@
 from cognite.client.data_classes import data_modeling as dm
 from cognite.client.data_classes.data_modeling.data_types import Enum, ListablePropertyType
 
+from cognite.pygen._constants import is_readonly_property
+
 from .base import Field
 
 
+@dataclass(frozen=True)
+class ContainerProperty:
+    source: dm.ContainerId
+    identifier: str
+
+
 @dataclass(frozen=True)
 class BasePrimitiveField(Field, ABC):
     """This is a base class for all primitive fields
@@ -23,6 +31,11 @@ class BasePrimitiveField(Field, ABC):
 
     type_: dm.PropertyType
     is_nullable: bool
+    container: ContainerProperty
+
+    @property
+    def is_write_field(self) -> bool:
+        return not is_readonly_property(self.container.source, self.container.identifier)
 
     @property
     def is_time_field(self) -> bool:
@@ -56,6 +69,7 @@ def as_typed_hint(self, operation: Literal["write", "read"] = "write") -> str:
 
     @classmethod
     def load(cls, base: Field, prop: dm.MappedProperty, variable: str) -> BasePrimitiveField | None:
+        container = ContainerProperty(prop.container, prop.container_property_identifier)
         if isinstance(prop.type, ListablePropertyType) and prop.type.is_list:
             return PrimitiveListField(
                 name=base.name,
@@ -66,6 +80,7 @@ def load(cls, base: Field, prop: dm.MappedProperty, variable: str) -> BasePrimit
                 type_=prop.type,
                 is_nullable=prop.nullable,
                 variable=variable,
+                container=container,
             )
         else:
             return PrimitiveField(
@@ -77,6 +92,7 @@ def load(cls, base: Field, prop: dm.MappedProperty, variable: str) -> BasePrimit
                 type_=prop.type,
                 is_nullable=prop.nullable,
                 default=prop.default_value,
+                container=container,
             )
 
 

cognite/pygen/_core/models/filter_methods.py

@@ -11,7 +11,14 @@
 from cognite.pygen import config as pygen_config
 from cognite.pygen.config.reserved_words import is_reserved_word
 
-from .fields import BaseConnectionField, Field, OneToManyConnectionField, OneToOneConnectionField, PrimitiveField
+from .fields import (
+    BaseConnectionField,
+    ContainerProperty,
+    Field,
+    OneToManyConnectionField,
+    OneToOneConnectionField,
+    PrimitiveField,
+)
 
 
 @dataclass
@@ -333,6 +340,10 @@ def from_fields(
     default=None,
     description=None,
     pydantic_field="Field",
+    container=ContainerProperty(
+        source=dm.ContainerId("irrelevant", "also_irrelevant"),
+        identifier="not_relevant",
+    ),
 )
 _SPACE_FIELD = PrimitiveField(
     name="space",
@@ -343,4 +354,8 @@ def from_fields(
     default=None,
     description=None,
     pydantic_field="Field",
+    container=ContainerProperty(
+        source=dm.ContainerId("irrelevant", "also_irrelevant"),
+        identifier="not_relevant",
+    ),
 )

cognite/pygen/utils/mock_generator.py

@@ -356,7 +356,7 @@ def _generate_mock_values(
         external = ViewMockData(view_id, self._instance_space)
         values: typing.Sequence[ListAbleDataType]
         for name, prop in properties.items():
-            if is_readonly_property(prop):
+            if is_readonly_property(prop.container, prop.container_property_identifier):
                 continue
 
             if name in config.properties:

docs/CHANGELOG.md

@@ -16,6 +16,8 @@ Changes are grouped as follows
 ## TBD
 ### Fixed
 - Doing `.query()` over an edge with properties without filtering no longer raises a `ValueError`.
+- When using the `CogniteCore` model, either directly or an extension of it, the generated SDK now
+  respects the read-only properties in `CogniteAsset` and `CogniteFile`.
 
 ## [0.99.48] - 24-11-14
 ### Fixed

examples/cognite_core/_api/cognite_asset.py

@@ -221,7 +221,7 @@ def apply(
         """Add or update (upsert) Cognite assets.
 
         Note: This method iterates through all nodes and timeseries linked to cognite_asset and creates them including the edges
-        between the nodes. For example, if any of `asset_class`, `object_3d`, `parent`, `path`, `root`, `source` or `type_` are set, then these
+        between the nodes. For example, if any of `asset_class`, `object_3d`, `parent`, `source` or `type_` are set, then these
         nodes as well as any nodes linked to them, and all the edges linking these nodes will be created.
 
         Args:

examples/cognite_core/data_classes/_cognite_asset.py

@@ -300,9 +300,6 @@ def as_write(self) -> CogniteAssetWrite:
             name=self.name,
             object_3d=self.object_3d.as_write() if isinstance(self.object_3d, GraphQLCore) else self.object_3d,
             parent=self.parent.as_write() if isinstance(self.parent, GraphQLCore) else self.parent,
-            path=[path.as_write() for path in self.path or []],
-            path_last_updated_time=self.path_last_updated_time,
-            root=self.root.as_write() if isinstance(self.root, GraphQLCore) else self.root,
             source=self.source.as_write() if isinstance(self.source, GraphQLCore) else self.source,
             source_context=self.source_context,
             source_created_time=self.source_created_time,
@@ -376,9 +373,6 @@ def as_write(self) -> CogniteAssetWrite:
             name=self.name,
             object_3d=self.object_3d.as_write() if isinstance(self.object_3d, DomainModel) else self.object_3d,
             parent=self.parent.as_write() if isinstance(self.parent, DomainModel) else self.parent,
-            path=[path.as_write() if isinstance(path, DomainModel) else path for path in self.path or []],
-            path_last_updated_time=self.path_last_updated_time,
-            root=self.root.as_write() if isinstance(self.root, DomainModel) else self.root,
             source=self.source.as_write() if isinstance(self.source, DomainModel) else self.source,
             source_context=self.source_context,
             source_created_time=self.source_created_time,
@@ -521,9 +515,6 @@ class CogniteAssetWrite(CogniteVisualizableWrite, CogniteDescribableNodeWrite, C
         name: Name of the instance
         object_3d: Direct relation to an Object3D instance representing the 3D resource
         parent: The parent of the asset.
-        path: An automatically updated ordered list of this asset's ancestors, starting with the root asset. Enables subtree filtering to find all assets under a parent.
-        path_last_updated_time: The last time the path was updated for this asset.
-        root: An automatically updated reference to the top-level asset of the hierarchy.
         source: Direct relation to a source system
         source_context: Context of the source id. For systems where the sourceId is globally unique, the sourceContext is expected to not be set.
         source_created_time: When the instance was created in source system (if available)
@@ -542,12 +533,9 @@ class CogniteAssetWrite(CogniteVisualizableWrite, CogniteDescribableNodeWrite, C
         default=None, repr=False, alias="assetClass"
     )
     parent: Union[CogniteAssetWrite, str, dm.NodeId, None] = Field(default=None, repr=False)
-    path: Optional[list[Union[CogniteAssetWrite, str, dm.NodeId]]] = Field(default=None, repr=False)
-    path_last_updated_time: Optional[datetime.datetime] = Field(None, alias="pathLastUpdatedTime")
-    root: Union[CogniteAssetWrite, str, dm.NodeId, None] = Field(default=None, repr=False)
     type_: Union[CogniteAssetTypeWrite, str, dm.NodeId, None] = Field(default=None, repr=False, alias="type")
 
-    @field_validator("asset_class", "parent", "path", "root", "type_", mode="before")
+    @field_validator("asset_class", "parent", "type_", mode="before")
     def as_node_id(cls, value: Any) -> Any:
         if isinstance(value, dm.DirectRelationReference):
             return dm.NodeId(value.space, value.external_id)
@@ -839,16 +827,6 @@ def object_3d(self) -> Cognite3DObjectWriteList:
     def parent(self) -> CogniteAssetWriteList:
         return CogniteAssetWriteList([item.parent for item in self.data if isinstance(item.parent, CogniteAssetWrite)])
 
-    @property
-    def path(self) -> CogniteAssetWriteList:
-        return CogniteAssetWriteList(
-            [item for items in self.data for item in items.path or [] if isinstance(item, CogniteAssetWrite)]
-        )
-
-    @property
-    def root(self) -> CogniteAssetWriteList:
-        return CogniteAssetWriteList([item.root for item in self.data if isinstance(item.root, CogniteAssetWrite)])
-
     @property
     def source(self) -> CogniteSourceSystemWriteList:
         from ._cognite_source_system import CogniteSourceSystemWrite, CogniteSourceSystemWriteList

examples/cognite_core/data_classes/_cognite_file.py

@@ -250,7 +250,6 @@ def as_write(self) -> CogniteFileWrite:
             category=self.category.as_write() if isinstance(self.category, GraphQLCore) else self.category,
             description=self.description,
             directory=self.directory,
-            is_uploaded=self.is_uploaded,
             mime_type=self.mime_type,
             name=self.name,
             source=self.source.as_write() if isinstance(self.source, GraphQLCore) else self.source,
@@ -261,7 +260,6 @@ def as_write(self) -> CogniteFileWrite:
             source_updated_time=self.source_updated_time,
             source_updated_user=self.source_updated_user,
             tags=self.tags,
-            uploaded_time=self.uploaded_time,
         )
 
 
@@ -316,7 +314,6 @@ def as_write(self) -> CogniteFileWrite:
             category=self.category.as_write() if isinstance(self.category, DomainModel) else self.category,
             description=self.description,
             directory=self.directory,
-            is_uploaded=self.is_uploaded,
             mime_type=self.mime_type,
             name=self.name,
             source=self.source.as_write() if isinstance(self.source, DomainModel) else self.source,
@@ -327,7 +324,6 @@ def as_write(self) -> CogniteFileWrite:
             source_updated_time=self.source_updated_time,
             source_updated_user=self.source_updated_user,
             tags=self.tags,
-            uploaded_time=self.uploaded_time,
         )
 
     def as_apply(self) -> CogniteFileWrite:
@@ -398,7 +394,6 @@ class CogniteFileWrite(CogniteDescribableNodeWrite, CogniteSourceableNodeWrite):
         category: Specifies the detected category the file belongs to. It's a direct relation to an instance of CogniteFileCategory.
         description: Description of the instance
         directory: Contains the path elements from the source (if the source system has a file system hierarchy or similar.)
-        is_uploaded: Specifies if the file content has been uploaded to Cognite Data Fusion or not.
         mime_type: The MIME type of the file.
         name: Name of the instance
         source: Direct relation to a source system
@@ -409,7 +404,6 @@ class CogniteFileWrite(CogniteDescribableNodeWrite, CogniteSourceableNodeWrite):
         source_updated_time: When the instance was last updated in the source system (if available)
         source_updated_user: User identifier from the source system on who last updated the source data. This identifier is not guaranteed to match the user identifiers in CDF
         tags: Text based labels for generic use, limited to 1000
-        uploaded_time: The time the file upload completed.
     """
 
     _view_id: ClassVar[dm.ViewId] = dm.ViewId("cdf_cdm", "CogniteFile", "v1")
@@ -418,9 +412,7 @@ class CogniteFileWrite(CogniteDescribableNodeWrite, CogniteSourceableNodeWrite):
     assets: Optional[list[Union[CogniteAssetWrite, str, dm.NodeId]]] = Field(default=None, repr=False)
     category: Union[CogniteFileCategoryWrite, str, dm.NodeId, None] = Field(default=None, repr=False)
     directory: Optional[str] = None
-    is_uploaded: Optional[bool] = Field(False, alias="isUploaded")
     mime_type: Optional[str] = Field(None, alias="mimeType")
-    uploaded_time: Optional[datetime.datetime] = Field(None, alias="uploadedTime")
 
     @field_validator("assets", "category", mode="before")
     def as_node_id(cls, value: Any) -> Any:

examples/wind_turbine/data_classes/_data_sheet.py

@@ -137,10 +137,8 @@ def as_write(self) -> DataSheetWrite:
             data_record=DataRecordWrite(existing_version=0),
             description=self.description,
             directory=self.directory,
-            is_uploaded=self.is_uploaded,
             mime_type=self.mime_type,
             name=self.name,
-            uploaded_time=self.uploaded_time,
         )
 
 
@@ -180,10 +178,8 @@ def as_write(self) -> DataSheetWrite:
             data_record=DataRecordWrite(existing_version=self.data_record.version),
             description=self.description,
             directory=self.directory,
-            is_uploaded=self.is_uploaded,
             mime_type=self.mime_type,
             name=self.name,
-            uploaded_time=self.uploaded_time,
         )
 
     def as_apply(self) -> DataSheetWrite:
@@ -207,10 +203,8 @@ class DataSheetWrite(DomainModelWrite):
         data_record: The data record of the data sheet node.
         description: Description of the instance
         directory: Contains the path elements from the source (if the source system has a file system hierarchy or similar.)
-        is_uploaded: Specifies if the file content has been uploaded to Cognite Data Fusion or not.
         mime_type: The MIME type of the file.
         name: Name of the instance
-        uploaded_time: The time the file upload completed.
     """
 
     _view_id: ClassVar[dm.ViewId] = dm.ViewId("sp_pygen_power", "DataSheet", "1")
@@ -219,10 +213,8 @@ class DataSheetWrite(DomainModelWrite):
     node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = None
     description: Optional[str] = None
     directory: Optional[str] = None
-    is_uploaded: Optional[bool] = Field(False, alias="isUploaded")
     mime_type: Optional[str] = Field(None, alias="mimeType")
     name: Optional[str] = None
-    uploaded_time: Optional[datetime.datetime] = Field(None, alias="uploadedTime")
 
     def _to_instances_write(
         self,