Omni Data Classes

examples.omni.data_classes

CDFExternalReferences

Bases: DomainModel

This represents the reading version of cdf external reference.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the cdf external reference.	required
data_record		The data record of the cdf external reference node.	required
file		The file field.	required
sequence		The sequence field.	required
timeseries		The timesery field.	required

Source code in examples/omni/data_classes/_cdf_external_references.py

class CDFExternalReferences(DomainModel):
    """This represents the reading version of cdf external reference.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the cdf external reference.
        data_record: The data record of the cdf external reference node.
        file: The file field.
        sequence: The sequence field.
        timeseries: The timesery field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "CDFExternalReferences", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, None] = None
    file: Union[FileMetadata, str, None] = None
    sequence: Union[SequenceRead, str, None] = None
    timeseries: Union[TimeSeries, str, None] = None

    def as_write(self) -> CDFExternalReferencesWrite:
        """Convert this read version of cdf external reference to the writing version."""
        return CDFExternalReferencesWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            file=self.file.as_write() if isinstance(self.file, CogniteFileMetadata) else self.file,
            sequence=self.sequence.as_write() if isinstance(self.sequence, CogniteSequence) else self.sequence,
            timeseries=(
                self.timeseries.as_write() if isinstance(self.timeseries, CogniteTimeSeries) else self.timeseries
            ),
        )

    def as_apply(self) -> CDFExternalReferencesWrite:
        """Convert this read version of cdf external reference to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert this read version of cdf external reference to the writing version.

Source code in examples/omni/data_classes/_cdf_external_references.py

def as_apply(self) -> CDFExternalReferencesWrite:
    """Convert this read version of cdf external reference to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of cdf external reference to the writing version.

Source code in examples/omni/data_classes/_cdf_external_references.py

def as_write(self) -> CDFExternalReferencesWrite:
    """Convert this read version of cdf external reference to the writing version."""
    return CDFExternalReferencesWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        file=self.file.as_write() if isinstance(self.file, CogniteFileMetadata) else self.file,
        sequence=self.sequence.as_write() if isinstance(self.sequence, CogniteSequence) else self.sequence,
        timeseries=(
            self.timeseries.as_write() if isinstance(self.timeseries, CogniteTimeSeries) else self.timeseries
        ),
    )

CDFExternalReferencesGraphQL

Bases: GraphQLCore

This represents the reading version of cdf external reference, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the cdf external reference.	required
data_record		The data record of the cdf external reference node.	required
file		The file field.	required
sequence		The sequence field.	required
timeseries		The timesery field.	required

Source code in examples/omni/data_classes/_cdf_external_references.py

class CDFExternalReferencesGraphQL(GraphQLCore):
    """This represents the reading version of cdf external reference, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the cdf external reference.
        data_record: The data record of the cdf external reference node.
        file: The file field.
        sequence: The sequence field.
        timeseries: The timesery field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "CDFExternalReferences", "1")
    file: Optional[FileMetadataGraphQL] = None
    sequence: Optional[SequenceGraphQL] = None
    timeseries: Optional[TimeSeriesGraphQL] = None

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> CDFExternalReferences:
        """Convert this GraphQL format of cdf external reference to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return CDFExternalReferences(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            file=self.file.as_read() if self.file else None,
            sequence=self.sequence.as_read() if self.sequence else None,
            timeseries=self.timeseries.as_read() if self.timeseries else None,
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> CDFExternalReferencesWrite:
        """Convert this GraphQL format of cdf external reference to the writing format."""
        return CDFExternalReferencesWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            file=self.file.as_write() if self.file else None,
            sequence=self.sequence.as_write() if self.sequence else None,
            timeseries=self.timeseries.as_write() if self.timeseries else None,
        )

as_read()

Convert this GraphQL format of cdf external reference to the reading format.

Source code in examples/omni/data_classes/_cdf_external_references.py

@no_type_check
def as_read(self) -> CDFExternalReferences:
    """Convert this GraphQL format of cdf external reference to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return CDFExternalReferences(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        file=self.file.as_read() if self.file else None,
        sequence=self.sequence.as_read() if self.sequence else None,
        timeseries=self.timeseries.as_read() if self.timeseries else None,
    )

as_write()

Convert this GraphQL format of cdf external reference to the writing format.

Source code in examples/omni/data_classes/_cdf_external_references.py

@no_type_check
def as_write(self) -> CDFExternalReferencesWrite:
    """Convert this GraphQL format of cdf external reference to the writing format."""
    return CDFExternalReferencesWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        file=self.file.as_write() if self.file else None,
        sequence=self.sequence.as_write() if self.sequence else None,
        timeseries=self.timeseries.as_write() if self.timeseries else None,
    )

CDFExternalReferencesList

Bases: DomainModelList[CDFExternalReferences]

List of cdf external references in the read version.

Source code in examples/omni/data_classes/_cdf_external_references.py

class CDFExternalReferencesList(DomainModelList[CDFExternalReferences]):
    """List of cdf external references in the read version."""

    _INSTANCE = CDFExternalReferences

    def as_write(self) -> CDFExternalReferencesWriteList:
        """Convert these read versions of cdf external reference to the writing versions."""
        return CDFExternalReferencesWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> CDFExternalReferencesWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/omni/data_classes/_cdf_external_references.py

def as_apply(self) -> CDFExternalReferencesWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of cdf external reference to the writing versions.

Source code in examples/omni/data_classes/_cdf_external_references.py

def as_write(self) -> CDFExternalReferencesWriteList:
    """Convert these read versions of cdf external reference to the writing versions."""
    return CDFExternalReferencesWriteList([node.as_write() for node in self.data])

CDFExternalReferencesListed

Bases: DomainModel

This represents the reading version of cdf external references listed.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the cdf external references listed.	required
data_record		The data record of the cdf external references listed node.	required
files		The file field.	required
sequences		The sequence field.	required
timeseries		The timesery field.	required

Source code in examples/omni/data_classes/_cdf_external_references_listed.py

class CDFExternalReferencesListed(DomainModel):
    """This represents the reading version of cdf external references listed.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the cdf external references listed.
        data_record: The data record of the cdf external references listed node.
        files: The file field.
        sequences: The sequence field.
        timeseries: The timesery field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "CDFExternalReferencesListed", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, None] = None
    files: Optional[list[Union[FileMetadata, str]]] = None
    sequences: Optional[list[Union[SequenceRead, str]]] = None
    timeseries: Optional[list[Union[TimeSeries, str]]] = None

    def as_write(self) -> CDFExternalReferencesListedWrite:
        """Convert this read version of cdf external references listed to the writing version."""
        return CDFExternalReferencesListedWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            files=[file.as_write() if isinstance(file, CogniteFileMetadata) else file for file in self.files or []]
            or None,
            sequences=[
                sequence.as_write() if isinstance(sequence, CogniteSequence) else sequence
                for sequence in self.sequences or []
            ]
            or None,
            timeseries=[
                timesery.as_write() if isinstance(timesery, CogniteTimeSeries) else timesery
                for timesery in self.timeseries or []
            ]
            or None,
        )

    def as_apply(self) -> CDFExternalReferencesListedWrite:
        """Convert this read version of cdf external references listed to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert this read version of cdf external references listed to the writing version.

Source code in examples/omni/data_classes/_cdf_external_references_listed.py

def as_apply(self) -> CDFExternalReferencesListedWrite:
    """Convert this read version of cdf external references listed to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of cdf external references listed to the writing version.

Source code in examples/omni/data_classes/_cdf_external_references_listed.py

def as_write(self) -> CDFExternalReferencesListedWrite:
    """Convert this read version of cdf external references listed to the writing version."""
    return CDFExternalReferencesListedWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        files=[file.as_write() if isinstance(file, CogniteFileMetadata) else file for file in self.files or []]
        or None,
        sequences=[
            sequence.as_write() if isinstance(sequence, CogniteSequence) else sequence
            for sequence in self.sequences or []
        ]
        or None,
        timeseries=[
            timesery.as_write() if isinstance(timesery, CogniteTimeSeries) else timesery
            for timesery in self.timeseries or []
        ]
        or None,
    )

CDFExternalReferencesListedGraphQL

Bases: GraphQLCore

This represents the reading version of cdf external references listed, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the cdf external references listed.	required
data_record		The data record of the cdf external references listed node.	required
files		The file field.	required
sequences		The sequence field.	required
timeseries		The timesery field.	required

Source code in examples/omni/data_classes/_cdf_external_references_listed.py

class CDFExternalReferencesListedGraphQL(GraphQLCore):
    """This represents the reading version of cdf external references listed, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the cdf external references listed.
        data_record: The data record of the cdf external references listed node.
        files: The file field.
        sequences: The sequence field.
        timeseries: The timesery field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "CDFExternalReferencesListed", "1")
    files: Optional[list[FileMetadataGraphQL]] = None
    sequences: Optional[list[SequenceGraphQL]] = None
    timeseries: Optional[list[TimeSeriesGraphQL]] = None

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    @field_validator("files", "sequences", "timeseries", mode="before")
    def clean_list(cls, value: Any) -> Any:
        if isinstance(value, list):
            return [v for v in value if v is not None] or None
        return value

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> CDFExternalReferencesListed:
        """Convert this GraphQL format of cdf external references listed to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return CDFExternalReferencesListed(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            files=[file.as_read() for file in self.files or []] or None,
            sequences=[sequence.as_read() for sequence in self.sequences or []] or None,
            timeseries=[timesery.as_read() for timesery in self.timeseries or []] or None,
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> CDFExternalReferencesListedWrite:
        """Convert this GraphQL format of cdf external references listed to the writing format."""
        return CDFExternalReferencesListedWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            files=[file.as_write() for file in self.files or []] or None,
            sequences=[sequence.as_write() for sequence in self.sequences or []] or None,
            timeseries=[timesery.as_write() for timesery in self.timeseries or []] or None,
        )

as_read()

Convert this GraphQL format of cdf external references listed to the reading format.

Source code in examples/omni/data_classes/_cdf_external_references_listed.py

@no_type_check
def as_read(self) -> CDFExternalReferencesListed:
    """Convert this GraphQL format of cdf external references listed to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return CDFExternalReferencesListed(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        files=[file.as_read() for file in self.files or []] or None,
        sequences=[sequence.as_read() for sequence in self.sequences or []] or None,
        timeseries=[timesery.as_read() for timesery in self.timeseries or []] or None,
    )

as_write()

Convert this GraphQL format of cdf external references listed to the writing format.

Source code in examples/omni/data_classes/_cdf_external_references_listed.py

@no_type_check
def as_write(self) -> CDFExternalReferencesListedWrite:
    """Convert this GraphQL format of cdf external references listed to the writing format."""
    return CDFExternalReferencesListedWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        files=[file.as_write() for file in self.files or []] or None,
        sequences=[sequence.as_write() for sequence in self.sequences or []] or None,
        timeseries=[timesery.as_write() for timesery in self.timeseries or []] or None,
    )

CDFExternalReferencesListedList

Bases: DomainModelList[CDFExternalReferencesListed]

List of cdf external references listeds in the read version.

Source code in examples/omni/data_classes/_cdf_external_references_listed.py

class CDFExternalReferencesListedList(DomainModelList[CDFExternalReferencesListed]):
    """List of cdf external references listeds in the read version."""

    _INSTANCE = CDFExternalReferencesListed

    def as_write(self) -> CDFExternalReferencesListedWriteList:
        """Convert these read versions of cdf external references listed to the writing versions."""
        return CDFExternalReferencesListedWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> CDFExternalReferencesListedWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/omni/data_classes/_cdf_external_references_listed.py

def as_apply(self) -> CDFExternalReferencesListedWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of cdf external references listed to the writing versions.

Source code in examples/omni/data_classes/_cdf_external_references_listed.py

def as_write(self) -> CDFExternalReferencesListedWriteList:
    """Convert these read versions of cdf external references listed to the writing versions."""
    return CDFExternalReferencesListedWriteList([node.as_write() for node in self.data])

CDFExternalReferencesListedWrite

Bases: DomainModelWrite

This represents the writing version of cdf external references listed.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the cdf external references listed.	required
data_record		The data record of the cdf external references listed node.	required
files		The file field.	required
sequences		The sequence field.	required
timeseries		The timesery field.	required

Source code in examples/omni/data_classes/_cdf_external_references_listed.py

class CDFExternalReferencesListedWrite(DomainModelWrite):
    """This represents the writing version of cdf external references listed.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the cdf external references listed.
        data_record: The data record of the cdf external references listed node.
        files: The file field.
        sequences: The sequence field.
        timeseries: The timesery field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "CDFExternalReferencesListed", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = None
    files: Optional[list[Union[FileMetadataWrite, str]]] = None
    sequences: Optional[list[Union[SequenceWrite, str]]] = None
    timeseries: Optional[list[Union[TimeSeriesWrite, str]]] = None

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.files is not None or write_none:
            properties["files"] = [
                file if isinstance(file, str) else file.external_id for file in self.files or []
            ] or None

        if self.sequences is not None or write_none:
            properties["sequences"] = [
                sequence if isinstance(sequence, str) else sequence.external_id for sequence in self.sequences or []
            ] or None

        if self.timeseries is not None or write_none:
            properties["timeseries"] = [
                timesery if isinstance(timesery, str) else timesery.external_id for timesery in self.timeseries or []
            ] or None

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        for file in self.files or []:
            if isinstance(file, CogniteFileMetadataWrite):
                resources.files.append(file)

        for sequence in self.sequences or []:
            if isinstance(sequence, CogniteSequenceWrite):
                resources.sequences.append(sequence)

        for timesery in self.timeseries or []:
            if isinstance(timesery, CogniteTimeSeriesWrite):
                resources.time_series.append(timesery)

        return resources

CDFExternalReferencesListedWriteList

Bases: DomainModelWriteList[CDFExternalReferencesListedWrite]

List of cdf external references listeds in the writing version.

Source code in examples/omni/data_classes/_cdf_external_references_listed.py

class CDFExternalReferencesListedWriteList(DomainModelWriteList[CDFExternalReferencesListedWrite]):
    """List of cdf external references listeds in the writing version."""

    _INSTANCE = CDFExternalReferencesListedWrite

CDFExternalReferencesWrite

Bases: DomainModelWrite

This represents the writing version of cdf external reference.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the cdf external reference.	required
data_record		The data record of the cdf external reference node.	required
file		The file field.	required
sequence		The sequence field.	required
timeseries		The timesery field.	required

Source code in examples/omni/data_classes/_cdf_external_references.py

class CDFExternalReferencesWrite(DomainModelWrite):
    """This represents the writing version of cdf external reference.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the cdf external reference.
        data_record: The data record of the cdf external reference node.
        file: The file field.
        sequence: The sequence field.
        timeseries: The timesery field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "CDFExternalReferences", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = None
    file: Union[FileMetadataWrite, str, None] = None
    sequence: Union[SequenceWrite, str, None] = None
    timeseries: Union[TimeSeriesWrite, str, None] = None

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.file is not None or write_none:
            properties["file"] = self.file if isinstance(self.file, str) or self.file is None else self.file.external_id

        if self.sequence is not None or write_none:
            properties["sequence"] = (
                self.sequence if isinstance(self.sequence, str) or self.sequence is None else self.sequence.external_id
            )

        if self.timeseries is not None or write_none:
            properties["timeseries"] = (
                self.timeseries
                if isinstance(self.timeseries, str) or self.timeseries is None
                else self.timeseries.external_id
            )

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        if isinstance(self.file, CogniteFileMetadataWrite):
            resources.files.append(self.file)

        if isinstance(self.sequence, CogniteSequenceWrite):
            resources.sequences.append(self.sequence)

        if isinstance(self.timeseries, CogniteTimeSeriesWrite):
            resources.time_series.append(self.timeseries)

        return resources

CDFExternalReferencesWriteList

Bases: DomainModelWriteList[CDFExternalReferencesWrite]

List of cdf external references in the writing version.

Source code in examples/omni/data_classes/_cdf_external_references.py

class CDFExternalReferencesWriteList(DomainModelWriteList[CDFExternalReferencesWrite]):
    """List of cdf external references in the writing version."""

    _INSTANCE = CDFExternalReferencesWrite

ConnectionEdgeA

Bases: DomainRelation

This represents the reading version of connection edge a.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the connection edge a.	required
data_record		The data record of the connection edge a edge.	required
end_node		The end node of this edge.	required
end_time		The end time field.	required
name		The name field.	required
start_time		The start time field.	required

Source code in examples/omni/data_classes/_connection_edge_a.py

class ConnectionEdgeA(DomainRelation):
    """This represents the reading version of connection edge a.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the connection edge a.
        data_record: The data record of the connection edge a edge.
        end_node: The end node of this edge.
        end_time: The end time field.
        name: The name field.
        start_time: The start time field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "ConnectionEdgeA", "1")
    space: str = DEFAULT_INSTANCE_SPACE
    end_node: Union[ConnectionItemE, ConnectionItemF, ConnectionItemG, str, dm.NodeId]
    end_time: Optional[datetime.datetime] = Field(None, alias="endTime")
    name: Optional[str] = None
    start_time: Optional[datetime.datetime] = Field(None, alias="startTime")

    def as_write(self) -> ConnectionEdgeAWrite:
        """Convert this read version of connection edge a to the writing version."""
        return ConnectionEdgeAWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            end_node=self.end_node.as_write() if isinstance(self.end_node, DomainModel) else self.end_node,
            end_time=self.end_time,
            name=self.name,
            start_time=self.start_time,
        )

    def as_apply(self) -> ConnectionEdgeAWrite:
        """Convert this read version of connection edge a to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert this read version of connection edge a to the writing version.

Source code in examples/omni/data_classes/_connection_edge_a.py

def as_apply(self) -> ConnectionEdgeAWrite:
    """Convert this read version of connection edge a to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of connection edge a to the writing version.

Source code in examples/omni/data_classes/_connection_edge_a.py

def as_write(self) -> ConnectionEdgeAWrite:
    """Convert this read version of connection edge a to the writing version."""
    return ConnectionEdgeAWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        end_node=self.end_node.as_write() if isinstance(self.end_node, DomainModel) else self.end_node,
        end_time=self.end_time,
        name=self.name,
        start_time=self.start_time,
    )

ConnectionEdgeAGraphQL

Bases: GraphQLCore

This represents the reading version of connection edge a, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the connection edge a.	required
data_record		The data record of the connection edge a node.	required
end_node		The end node of this edge.	required
end_time		The end time field.	required
name		The name field.	required
start_time		The start time field.	required

Source code in examples/omni/data_classes/_connection_edge_a.py

class ConnectionEdgeAGraphQL(GraphQLCore):
    """This represents the reading version of connection edge a, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the connection edge a.
        data_record: The data record of the connection edge a node.
        end_node: The end node of this edge.
        end_time: The end time field.
        name: The name field.
        start_time: The start time field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "ConnectionEdgeA", "1")
    end_node: Union[ConnectionItemEGraphQL, ConnectionItemFGraphQL, ConnectionItemGGraphQL, None] = None
    end_time: Optional[datetime.datetime] = Field(None, alias="endTime")
    name: Optional[str] = None
    start_time: Optional[datetime.datetime] = Field(None, alias="startTime")

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> ConnectionEdgeA:
        """Convert this GraphQL format of connection edge a to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return ConnectionEdgeA(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            end_node=self.end_node.as_read() if isinstance(self.end_node, GraphQLCore) else self.end_node,
            end_time=self.end_time,
            name=self.name,
            start_time=self.start_time,
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> ConnectionEdgeAWrite:
        """Convert this GraphQL format of connection edge a to the writing format."""
        return ConnectionEdgeAWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            end_node=self.end_node.as_write() if isinstance(self.end_node, DomainModel) else self.end_node,
            end_time=self.end_time,
            name=self.name,
            start_time=self.start_time,
        )

as_read()

Convert this GraphQL format of connection edge a to the reading format.

Source code in examples/omni/data_classes/_connection_edge_a.py

@no_type_check
def as_read(self) -> ConnectionEdgeA:
    """Convert this GraphQL format of connection edge a to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return ConnectionEdgeA(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        end_node=self.end_node.as_read() if isinstance(self.end_node, GraphQLCore) else self.end_node,
        end_time=self.end_time,
        name=self.name,
        start_time=self.start_time,
    )

as_write()

Convert this GraphQL format of connection edge a to the writing format.

Source code in examples/omni/data_classes/_connection_edge_a.py

@no_type_check
def as_write(self) -> ConnectionEdgeAWrite:
    """Convert this GraphQL format of connection edge a to the writing format."""
    return ConnectionEdgeAWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        end_node=self.end_node.as_write() if isinstance(self.end_node, DomainModel) else self.end_node,
        end_time=self.end_time,
        name=self.name,
        start_time=self.start_time,
    )

ConnectionEdgeAList

Bases: DomainRelationList[ConnectionEdgeA]

List of connection edge as in the reading version.

Source code in examples/omni/data_classes/_connection_edge_a.py

class ConnectionEdgeAList(DomainRelationList[ConnectionEdgeA]):
    """List of connection edge as in the reading version."""

    _INSTANCE = ConnectionEdgeA

    def as_write(self) -> ConnectionEdgeAWriteList:
        """Convert this read version of connection edge a list to the writing version."""
        return ConnectionEdgeAWriteList([edge.as_write() for edge in self])

    def as_apply(self) -> ConnectionEdgeAWriteList:
        """Convert these read versions of connection edge a list to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of connection edge a list to the writing versions.

Source code in examples/omni/data_classes/_connection_edge_a.py

def as_apply(self) -> ConnectionEdgeAWriteList:
    """Convert these read versions of connection edge a list to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of connection edge a list to the writing version.

Source code in examples/omni/data_classes/_connection_edge_a.py

def as_write(self) -> ConnectionEdgeAWriteList:
    """Convert this read version of connection edge a list to the writing version."""
    return ConnectionEdgeAWriteList([edge.as_write() for edge in self])

ConnectionEdgeAWrite

Bases: DomainRelationWrite

This represents the writing version of connection edge a.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the connection edge a.	required
data_record		The data record of the connection edge a edge.	required
end_node		The end node of this edge.	required
end_time		The end time field.	required
name		The name field.	required
start_time		The start time field.	required

Source code in examples/omni/data_classes/_connection_edge_a.py

class ConnectionEdgeAWrite(DomainRelationWrite):
    """This represents the writing version of connection edge a.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the connection edge a.
        data_record: The data record of the connection edge a edge.
        end_node: The end node of this edge.
        end_time: The end time field.
        name: The name field.
        start_time: The start time field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "ConnectionEdgeA", "1")
    space: str = DEFAULT_INSTANCE_SPACE
    end_node: Union[ConnectionItemEWrite, ConnectionItemFWrite, ConnectionItemGWrite, str, dm.NodeId]
    end_time: Optional[datetime.datetime] = Field(None, alias="endTime")
    name: Optional[str] = None
    start_time: Optional[datetime.datetime] = Field(None, alias="startTime")

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        start_node: DomainModelWrite,
        edge_type: dm.DirectRelationReference,
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.external_id and (self.space, self.external_id) in cache:
            return resources

        _validate_end_node(start_node, self.end_node)

        if isinstance(self.end_node, DomainModelWrite):
            end_node = self.end_node.as_direct_reference()
        elif isinstance(self.end_node, str):
            end_node = dm.DirectRelationReference(self.space, self.end_node)
        elif isinstance(self.end_node, dm.NodeId):
            end_node = dm.DirectRelationReference(self.end_node.space, self.end_node.external_id)
        else:
            raise ValueError(f"Invalid type for equipment_module: {type(self.end_node)}")

        external_id = self.external_id or DomainRelationWrite.external_id_factory(start_node, self.end_node, edge_type)

        properties: dict[str, Any] = {}

        if self.end_time is not None or write_none:
            properties["endTime"] = self.end_time.isoformat(timespec="milliseconds") if self.end_time else None

        if self.name is not None or write_none:
            properties["name"] = self.name

        if self.start_time is not None or write_none:
            properties["startTime"] = self.start_time.isoformat(timespec="milliseconds") if self.start_time else None

        if properties:
            this_edge = dm.EdgeApply(
                space=self.space,
                external_id=external_id,
                type=edge_type,
                start_node=start_node.as_direct_reference(),
                end_node=end_node,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.edges.append(this_edge)
            cache.add((self.space, external_id))

        if isinstance(self.end_node, DomainModelWrite):
            other_resources = self.end_node._to_instances_write(cache)
            resources.extend(other_resources)

        return resources

ConnectionEdgeAWriteList

Bases: DomainRelationWriteList[ConnectionEdgeAWrite]

List of connection edge as in the writing version.

Source code in examples/omni/data_classes/_connection_edge_a.py

class ConnectionEdgeAWriteList(DomainRelationWriteList[ConnectionEdgeAWrite]):
    """List of connection edge as in the writing version."""

    _INSTANCE = ConnectionEdgeAWrite

ConnectionItemA

Bases: DomainModel

This represents the reading version of connection item a.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the connection item a.	required
data_record		The data record of the connection item a node.	required
name		The name field.	required
other_direct		The other direct field.	required
outwards		The outward field.	required
self_direct		The self direct field.	required

Source code in examples/omni/data_classes/_connection_item_a.py

class ConnectionItemA(DomainModel):
    """This represents the reading version of connection item a.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the connection item a.
        data_record: The data record of the connection item a node.
        name: The name field.
        other_direct: The other direct field.
        outwards: The outward field.
        self_direct: The self direct field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "ConnectionItemA", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, None] = dm.DirectRelationReference("pygen-models", "ConnectionItemA")
    name: Optional[str] = None
    other_direct: Union[ConnectionItemCNode, str, dm.NodeId, None] = Field(
        default=None, repr=False, alias="otherDirect"
    )
    outwards: Optional[list[Union[ConnectionItemB, str, dm.NodeId]]] = Field(default=None, repr=False)
    self_direct: Union[ConnectionItemA, str, dm.NodeId, None] = Field(default=None, repr=False, alias="selfDirect")

    def as_write(self) -> ConnectionItemAWrite:
        """Convert this read version of connection item a to the writing version."""
        return ConnectionItemAWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            name=self.name,
            other_direct=(
                self.other_direct.as_write() if isinstance(self.other_direct, DomainModel) else self.other_direct
            ),
            outwards=[
                outward.as_write() if isinstance(outward, DomainModel) else outward for outward in self.outwards or []
            ],
            self_direct=self.self_direct.as_write() if isinstance(self.self_direct, DomainModel) else self.self_direct,
        )

    def as_apply(self) -> ConnectionItemAWrite:
        """Convert this read version of connection item a to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

    @classmethod
    def _update_connections(
        cls,
        instances: dict[dm.NodeId | str, ConnectionItemA],  # type: ignore[override]
        nodes_by_id: dict[dm.NodeId | str, DomainModel],
        edges_by_source_node: dict[dm.NodeId, list[dm.Edge | DomainRelation]],
    ) -> None:
        from ._connection_item_b import ConnectionItemB
        from ._connection_item_c_node import ConnectionItemCNode

        for instance in instances.values():
            if (
                isinstance(instance.other_direct, (dm.NodeId, str))
                and (other_direct := nodes_by_id.get(instance.other_direct))
                and isinstance(other_direct, ConnectionItemCNode)
            ):
                instance.other_direct = other_direct
            if (
                isinstance(instance.self_direct, (dm.NodeId, str))
                and (self_direct := nodes_by_id.get(instance.self_direct))
                and isinstance(self_direct, ConnectionItemA)
            ):
                instance.self_direct = self_direct
            if edges := edges_by_source_node.get(instance.as_id()):
                outwards: list[ConnectionItemB | str | dm.NodeId] = []
                for edge in edges:
                    value: DomainModel | DomainRelation | str | dm.NodeId
                    if isinstance(edge, DomainRelation):
                        value = edge
                    else:
                        other_end: dm.DirectRelationReference = (
                            edge.end_node
                            if edge.start_node.space == instance.space
                            and edge.start_node.external_id == instance.external_id
                            else edge.start_node
                        )
                        destination: dm.NodeId | str = (
                            as_node_id(other_end)
                            if other_end.space != DEFAULT_INSTANCE_SPACE
                            else other_end.external_id
                        )
                        if destination in nodes_by_id:
                            value = nodes_by_id[destination]
                        else:
                            value = destination
                    edge_type = edge.edge_type if isinstance(edge, DomainRelation) else edge.type

                    if edge_type == dm.DirectRelationReference("pygen-models", "bidirectional") and isinstance(
                        value, (ConnectionItemB, str, dm.NodeId)
                    ):
                        outwards.append(value)

                instance.outwards = outwards or None

as_apply()

Convert this read version of connection item a to the writing version.

Source code in examples/omni/data_classes/_connection_item_a.py

def as_apply(self) -> ConnectionItemAWrite:
    """Convert this read version of connection item a to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of connection item a to the writing version.

Source code in examples/omni/data_classes/_connection_item_a.py

def as_write(self) -> ConnectionItemAWrite:
    """Convert this read version of connection item a to the writing version."""
    return ConnectionItemAWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        name=self.name,
        other_direct=(
            self.other_direct.as_write() if isinstance(self.other_direct, DomainModel) else self.other_direct
        ),
        outwards=[
            outward.as_write() if isinstance(outward, DomainModel) else outward for outward in self.outwards or []
        ],
        self_direct=self.self_direct.as_write() if isinstance(self.self_direct, DomainModel) else self.self_direct,
    )

ConnectionItemAGraphQL

Bases: GraphQLCore

This represents the reading version of connection item a, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the connection item a.	required
data_record		The data record of the connection item a node.	required
name		The name field.	required
other_direct		The other direct field.	required
outwards		The outward field.	required
self_direct		The self direct field.	required

Source code in examples/omni/data_classes/_connection_item_a.py

class ConnectionItemAGraphQL(GraphQLCore):
    """This represents the reading version of connection item a, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the connection item a.
        data_record: The data record of the connection item a node.
        name: The name field.
        other_direct: The other direct field.
        outwards: The outward field.
        self_direct: The self direct field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "ConnectionItemA", "1")
    name: Optional[str] = None
    other_direct: Optional[ConnectionItemCNodeGraphQL] = Field(default=None, repr=False, alias="otherDirect")
    outwards: Optional[list[ConnectionItemBGraphQL]] = Field(default=None, repr=False)
    self_direct: Optional[ConnectionItemAGraphQL] = Field(default=None, repr=False, alias="selfDirect")

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    @field_validator("other_direct", "outwards", "self_direct", mode="before")
    def parse_graphql(cls, value: Any) -> Any:
        if not isinstance(value, dict):
            return value
        if "items" in value:
            return value["items"]
        return value

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> ConnectionItemA:
        """Convert this GraphQL format of connection item a to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return ConnectionItemA(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            name=self.name,
            other_direct=(
                self.other_direct.as_read() if isinstance(self.other_direct, GraphQLCore) else self.other_direct
            ),
            outwards=[outward.as_read() for outward in self.outwards or []],
            self_direct=self.self_direct.as_read() if isinstance(self.self_direct, GraphQLCore) else self.self_direct,
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> ConnectionItemAWrite:
        """Convert this GraphQL format of connection item a to the writing format."""
        return ConnectionItemAWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            name=self.name,
            other_direct=(
                self.other_direct.as_write() if isinstance(self.other_direct, GraphQLCore) else self.other_direct
            ),
            outwards=[outward.as_write() for outward in self.outwards or []],
            self_direct=self.self_direct.as_write() if isinstance(self.self_direct, GraphQLCore) else self.self_direct,
        )

as_read()

Convert this GraphQL format of connection item a to the reading format.

Source code in examples/omni/data_classes/_connection_item_a.py

@no_type_check
def as_read(self) -> ConnectionItemA:
    """Convert this GraphQL format of connection item a to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return ConnectionItemA(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        name=self.name,
        other_direct=(
            self.other_direct.as_read() if isinstance(self.other_direct, GraphQLCore) else self.other_direct
        ),
        outwards=[outward.as_read() for outward in self.outwards or []],
        self_direct=self.self_direct.as_read() if isinstance(self.self_direct, GraphQLCore) else self.self_direct,
    )

as_write()

Convert this GraphQL format of connection item a to the writing format.

Source code in examples/omni/data_classes/_connection_item_a.py

@no_type_check
def as_write(self) -> ConnectionItemAWrite:
    """Convert this GraphQL format of connection item a to the writing format."""
    return ConnectionItemAWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        name=self.name,
        other_direct=(
            self.other_direct.as_write() if isinstance(self.other_direct, GraphQLCore) else self.other_direct
        ),
        outwards=[outward.as_write() for outward in self.outwards or []],
        self_direct=self.self_direct.as_write() if isinstance(self.self_direct, GraphQLCore) else self.self_direct,
    )

ConnectionItemAList

Bases: DomainModelList[ConnectionItemA]

List of connection item as in the read version.

Source code in examples/omni/data_classes/_connection_item_a.py

class ConnectionItemAList(DomainModelList[ConnectionItemA]):
    """List of connection item as in the read version."""

    _INSTANCE = ConnectionItemA

    def as_write(self) -> ConnectionItemAWriteList:
        """Convert these read versions of connection item a to the writing versions."""
        return ConnectionItemAWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> ConnectionItemAWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/omni/data_classes/_connection_item_a.py

def as_apply(self) -> ConnectionItemAWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of connection item a to the writing versions.

Source code in examples/omni/data_classes/_connection_item_a.py

def as_write(self) -> ConnectionItemAWriteList:
    """Convert these read versions of connection item a to the writing versions."""
    return ConnectionItemAWriteList([node.as_write() for node in self.data])

ConnectionItemAWrite

Bases: DomainModelWrite

This represents the writing version of connection item a.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the connection item a.	required
data_record		The data record of the connection item a node.	required
name		The name field.	required
other_direct		The other direct field.	required
outwards		The outward field.	required
self_direct		The self direct field.	required

Source code in examples/omni/data_classes/_connection_item_a.py

class ConnectionItemAWrite(DomainModelWrite):
    """This represents the writing version of connection item a.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the connection item a.
        data_record: The data record of the connection item a node.
        name: The name field.
        other_direct: The other direct field.
        outwards: The outward field.
        self_direct: The self direct field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "ConnectionItemA", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = dm.DirectRelationReference(
        "pygen-models", "ConnectionItemA"
    )
    name: Optional[str] = None
    other_direct: Union[ConnectionItemCNodeWrite, str, dm.NodeId, None] = Field(
        default=None, repr=False, alias="otherDirect"
    )
    outwards: Optional[list[Union[ConnectionItemBWrite, str, dm.NodeId]]] = Field(default=None, repr=False)
    self_direct: Union[ConnectionItemAWrite, str, dm.NodeId, None] = Field(default=None, repr=False, alias="selfDirect")

    @field_validator("other_direct", "outwards", "self_direct", mode="before")
    def as_node_id(cls, value: Any) -> Any:
        if isinstance(value, dm.DirectRelationReference):
            return dm.NodeId(value.space, value.external_id)
        elif isinstance(value, tuple) and len(value) == 2 and all(isinstance(item, str) for item in value):
            return dm.NodeId(value[0], value[1])
        elif isinstance(value, list):
            return [cls.as_node_id(item) for item in value]
        return value

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.name is not None or write_none:
            properties["name"] = self.name

        if self.other_direct is not None:
            properties["otherDirect"] = {
                "space": self.space if isinstance(self.other_direct, str) else self.other_direct.space,
                "externalId": (
                    self.other_direct if isinstance(self.other_direct, str) else self.other_direct.external_id
                ),
            }

        if self.self_direct is not None:
            properties["selfDirect"] = {
                "space": self.space if isinstance(self.self_direct, str) else self.self_direct.space,
                "externalId": self.self_direct if isinstance(self.self_direct, str) else self.self_direct.external_id,
            }

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        edge_type = dm.DirectRelationReference("pygen-models", "bidirectional")
        for outward in self.outwards or []:
            other_resources = DomainRelationWrite.from_edge_to_resources(
                cache,
                start_node=self,
                end_node=outward,
                edge_type=edge_type,
                write_none=write_none,
                allow_version_increase=allow_version_increase,
            )
            resources.extend(other_resources)

        if isinstance(self.other_direct, DomainModelWrite):
            other_resources = self.other_direct._to_instances_write(cache)
            resources.extend(other_resources)

        if isinstance(self.self_direct, DomainModelWrite):
            other_resources = self.self_direct._to_instances_write(cache)
            resources.extend(other_resources)

        return resources

ConnectionItemAWriteList

Bases: DomainModelWriteList[ConnectionItemAWrite]

List of connection item as in the writing version.

Source code in examples/omni/data_classes/_connection_item_a.py

class ConnectionItemAWriteList(DomainModelWriteList[ConnectionItemAWrite]):
    """List of connection item as in the writing version."""

    _INSTANCE = ConnectionItemAWrite

ConnectionItemB

Bases: DomainModel

This represents the reading version of connection item b.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the connection item b.	required
data_record		The data record of the connection item b node.	required
inwards		The inward field.	required
name		The name field.	required
self_edge		The self edge field.	required

Source code in examples/omni/data_classes/_connection_item_b.py

class ConnectionItemB(DomainModel):
    """This represents the reading version of connection item b.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the connection item b.
        data_record: The data record of the connection item b node.
        inwards: The inward field.
        name: The name field.
        self_edge: The self edge field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "ConnectionItemB", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, None] = dm.DirectRelationReference("pygen-models", "ConnectionItemB")
    inwards: Optional[list[Union[ConnectionItemA, str, dm.NodeId]]] = Field(default=None, repr=False)
    name: Optional[str] = None
    self_edge: Optional[list[Union[ConnectionItemB, str, dm.NodeId]]] = Field(
        default=None, repr=False, alias="selfEdge"
    )

    def as_write(self) -> ConnectionItemBWrite:
        """Convert this read version of connection item b to the writing version."""
        return ConnectionItemBWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            inwards=[inward.as_write() if isinstance(inward, DomainModel) else inward for inward in self.inwards or []],
            name=self.name,
            self_edge=[
                self_edge.as_write() if isinstance(self_edge, DomainModel) else self_edge
                for self_edge in self.self_edge or []
            ],
        )

    def as_apply(self) -> ConnectionItemBWrite:
        """Convert this read version of connection item b to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

    @classmethod
    def _update_connections(
        cls,
        instances: dict[dm.NodeId | str, ConnectionItemB],  # type: ignore[override]
        nodes_by_id: dict[dm.NodeId | str, DomainModel],
        edges_by_source_node: dict[dm.NodeId, list[dm.Edge | DomainRelation]],
    ) -> None:
        from ._connection_item_a import ConnectionItemA

        for instance in instances.values():
            if edges := edges_by_source_node.get(instance.as_id()):
                inwards: list[ConnectionItemA | str | dm.NodeId] = []
                self_edge: list[ConnectionItemB | str | dm.NodeId] = []
                for edge in edges:
                    value: DomainModel | DomainRelation | str | dm.NodeId
                    if isinstance(edge, DomainRelation):
                        value = edge
                    else:
                        other_end: dm.DirectRelationReference = (
                            edge.end_node
                            if edge.start_node.space == instance.space
                            and edge.start_node.external_id == instance.external_id
                            else edge.start_node
                        )
                        destination: dm.NodeId | str = (
                            as_node_id(other_end)
                            if other_end.space != DEFAULT_INSTANCE_SPACE
                            else other_end.external_id
                        )
                        if destination in nodes_by_id:
                            value = nodes_by_id[destination]
                        else:
                            value = destination
                    edge_type = edge.edge_type if isinstance(edge, DomainRelation) else edge.type

                    if edge_type == dm.DirectRelationReference("pygen-models", "bidirectional") and isinstance(
                        value, (ConnectionItemA, str, dm.NodeId)
                    ):
                        inwards.append(value)
                    if edge_type == dm.DirectRelationReference("pygen-models", "reflexive") and isinstance(
                        value, (ConnectionItemB, str, dm.NodeId)
                    ):
                        self_edge.append(value)

                instance.inwards = inwards or None
                instance.self_edge = self_edge or None

as_apply()

Convert this read version of connection item b to the writing version.

Source code in examples/omni/data_classes/_connection_item_b.py

def as_apply(self) -> ConnectionItemBWrite:
    """Convert this read version of connection item b to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of connection item b to the writing version.

Source code in examples/omni/data_classes/_connection_item_b.py

def as_write(self) -> ConnectionItemBWrite:
    """Convert this read version of connection item b to the writing version."""
    return ConnectionItemBWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        inwards=[inward.as_write() if isinstance(inward, DomainModel) else inward for inward in self.inwards or []],
        name=self.name,
        self_edge=[
            self_edge.as_write() if isinstance(self_edge, DomainModel) else self_edge
            for self_edge in self.self_edge or []
        ],
    )

ConnectionItemBGraphQL

Bases: GraphQLCore

This represents the reading version of connection item b, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the connection item b.	required
data_record		The data record of the connection item b node.	required
inwards		The inward field.	required
name		The name field.	required
self_edge		The self edge field.	required

Source code in examples/omni/data_classes/_connection_item_b.py

class ConnectionItemBGraphQL(GraphQLCore):
    """This represents the reading version of connection item b, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the connection item b.
        data_record: The data record of the connection item b node.
        inwards: The inward field.
        name: The name field.
        self_edge: The self edge field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "ConnectionItemB", "1")
    inwards: Optional[list[ConnectionItemAGraphQL]] = Field(default=None, repr=False)
    name: Optional[str] = None
    self_edge: Optional[list[ConnectionItemBGraphQL]] = Field(default=None, repr=False, alias="selfEdge")

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    @field_validator("inwards", "self_edge", mode="before")
    def parse_graphql(cls, value: Any) -> Any:
        if not isinstance(value, dict):
            return value
        if "items" in value:
            return value["items"]
        return value

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> ConnectionItemB:
        """Convert this GraphQL format of connection item b to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return ConnectionItemB(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            inwards=[inward.as_read() for inward in self.inwards or []],
            name=self.name,
            self_edge=[self_edge.as_read() for self_edge in self.self_edge or []],
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> ConnectionItemBWrite:
        """Convert this GraphQL format of connection item b to the writing format."""
        return ConnectionItemBWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            inwards=[inward.as_write() for inward in self.inwards or []],
            name=self.name,
            self_edge=[self_edge.as_write() for self_edge in self.self_edge or []],
        )

as_read()

Convert this GraphQL format of connection item b to the reading format.

Source code in examples/omni/data_classes/_connection_item_b.py

@no_type_check
def as_read(self) -> ConnectionItemB:
    """Convert this GraphQL format of connection item b to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return ConnectionItemB(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        inwards=[inward.as_read() for inward in self.inwards or []],
        name=self.name,
        self_edge=[self_edge.as_read() for self_edge in self.self_edge or []],
    )

as_write()

Convert this GraphQL format of connection item b to the writing format.

Source code in examples/omni/data_classes/_connection_item_b.py

@no_type_check
def as_write(self) -> ConnectionItemBWrite:
    """Convert this GraphQL format of connection item b to the writing format."""
    return ConnectionItemBWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        inwards=[inward.as_write() for inward in self.inwards or []],
        name=self.name,
        self_edge=[self_edge.as_write() for self_edge in self.self_edge or []],
    )

ConnectionItemBList

Bases: DomainModelList[ConnectionItemB]

List of connection item bs in the read version.

Source code in examples/omni/data_classes/_connection_item_b.py

class ConnectionItemBList(DomainModelList[ConnectionItemB]):
    """List of connection item bs in the read version."""

    _INSTANCE = ConnectionItemB

    def as_write(self) -> ConnectionItemBWriteList:
        """Convert these read versions of connection item b to the writing versions."""
        return ConnectionItemBWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> ConnectionItemBWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/omni/data_classes/_connection_item_b.py

def as_apply(self) -> ConnectionItemBWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of connection item b to the writing versions.

Source code in examples/omni/data_classes/_connection_item_b.py

def as_write(self) -> ConnectionItemBWriteList:
    """Convert these read versions of connection item b to the writing versions."""
    return ConnectionItemBWriteList([node.as_write() for node in self.data])

ConnectionItemBWrite

Bases: DomainModelWrite

This represents the writing version of connection item b.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the connection item b.	required
data_record		The data record of the connection item b node.	required
inwards		The inward field.	required
name		The name field.	required
self_edge		The self edge field.	required

Source code in examples/omni/data_classes/_connection_item_b.py

class ConnectionItemBWrite(DomainModelWrite):
    """This represents the writing version of connection item b.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the connection item b.
        data_record: The data record of the connection item b node.
        inwards: The inward field.
        name: The name field.
        self_edge: The self edge field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "ConnectionItemB", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = dm.DirectRelationReference(
        "pygen-models", "ConnectionItemB"
    )
    inwards: Optional[list[Union[ConnectionItemAWrite, str, dm.NodeId]]] = Field(default=None, repr=False)
    name: Optional[str] = None
    self_edge: Optional[list[Union[ConnectionItemBWrite, str, dm.NodeId]]] = Field(
        default=None, repr=False, alias="selfEdge"
    )

    @field_validator("inwards", "self_edge", mode="before")
    def as_node_id(cls, value: Any) -> Any:
        if isinstance(value, dm.DirectRelationReference):
            return dm.NodeId(value.space, value.external_id)
        elif isinstance(value, tuple) and len(value) == 2 and all(isinstance(item, str) for item in value):
            return dm.NodeId(value[0], value[1])
        elif isinstance(value, list):
            return [cls.as_node_id(item) for item in value]
        return value

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.name is not None or write_none:
            properties["name"] = self.name

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        edge_type = dm.DirectRelationReference("pygen-models", "bidirectional")
        for inward in self.inwards or []:
            other_resources = DomainRelationWrite.from_edge_to_resources(
                cache,
                start_node=inward,
                end_node=self,
                edge_type=edge_type,
                write_none=write_none,
                allow_version_increase=allow_version_increase,
            )
            resources.extend(other_resources)

        edge_type = dm.DirectRelationReference("pygen-models", "reflexive")
        for self_edge in self.self_edge or []:
            other_resources = DomainRelationWrite.from_edge_to_resources(
                cache,
                start_node=self,
                end_node=self_edge,
                edge_type=edge_type,
                write_none=write_none,
                allow_version_increase=allow_version_increase,
            )
            resources.extend(other_resources)

        return resources

ConnectionItemBWriteList

Bases: DomainModelWriteList[ConnectionItemBWrite]

List of connection item bs in the writing version.

Source code in examples/omni/data_classes/_connection_item_b.py

class ConnectionItemBWriteList(DomainModelWriteList[ConnectionItemBWrite]):
    """List of connection item bs in the writing version."""

    _INSTANCE = ConnectionItemBWrite

ConnectionItemCEdge

Bases: DomainRelation

This represents the reading version of connection item c edge.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the connection item c edge.	required
data_record		The data record of the connection item c edge edge.	required
end_node		The end node of this edge.	required
connection_item_a		The connection item a field.	required
connection_item_b		The connection item b field.	required

Source code in examples/omni/data_classes/_connection_item_c_edge.py

class ConnectionItemCEdge(DomainRelation):
    """This represents the reading version of connection item c edge.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the connection item c edge.
        data_record: The data record of the connection item c edge edge.
        end_node: The end node of this edge.
        connection_item_a: The connection item a field.
        connection_item_b: The connection item b field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "ConnectionItemC", "1")
    space: str = DEFAULT_INSTANCE_SPACE
    end_node: Union[str, dm.NodeId]
    connection_item_a: Optional[list[Union[ConnectionItemA, str, dm.NodeId]]] = Field(
        default=None, repr=False, alias="connectionItemA"
    )
    connection_item_b: Optional[list[Union[ConnectionItemB, str, dm.NodeId]]] = Field(
        default=None, repr=False, alias="connectionItemB"
    )

ConnectionItemCEdgeGraphQL

Bases: GraphQLCore

This represents the reading version of connection item c edge, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the connection item c edge.	required
data_record		The data record of the connection item c edge node.	required
end_node		The end node of this edge.	required
connection_item_a		The connection item a field.	required
connection_item_b		The connection item b field.	required

Source code in examples/omni/data_classes/_connection_item_c_edge.py

class ConnectionItemCEdgeGraphQL(GraphQLCore):
    """This represents the reading version of connection item c edge, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the connection item c edge.
        data_record: The data record of the connection item c edge node.
        end_node: The end node of this edge.
        connection_item_a: The connection item a field.
        connection_item_b: The connection item b field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "ConnectionItemC", "1")
    end_node: Union[dm.NodeId, None] = None
    connection_item_a: Optional[list[ConnectionItemAGraphQL]] = Field(default=None, repr=False, alias="connectionItemA")
    connection_item_b: Optional[list[ConnectionItemBGraphQL]] = Field(default=None, repr=False, alias="connectionItemB")

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> ConnectionItemCEdge:
        """Convert this GraphQL format of connection item c edge to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return ConnectionItemCEdge(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            end_node=self.end_node.as_read() if isinstance(self.end_node, GraphQLCore) else self.end_node,
            connection_item_a=[connection_item_a.as_read() for connection_item_a in self.connection_item_a or []],
            connection_item_b=[connection_item_b.as_read() for connection_item_b in self.connection_item_b or []],
        )

as_read()

Convert this GraphQL format of connection item c edge to the reading format.

Source code in examples/omni/data_classes/_connection_item_c_edge.py

@no_type_check
def as_read(self) -> ConnectionItemCEdge:
    """Convert this GraphQL format of connection item c edge to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return ConnectionItemCEdge(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        end_node=self.end_node.as_read() if isinstance(self.end_node, GraphQLCore) else self.end_node,
        connection_item_a=[connection_item_a.as_read() for connection_item_a in self.connection_item_a or []],
        connection_item_b=[connection_item_b.as_read() for connection_item_b in self.connection_item_b or []],
    )

ConnectionItemCEdgeList

Bases: DomainRelationList[ConnectionItemCEdge]

List of connection item c edges in the reading version.

Source code in examples/omni/data_classes/_connection_item_c_edge.py

class ConnectionItemCEdgeList(DomainRelationList[ConnectionItemCEdge]):
    """List of connection item c edges in the reading version."""

    _INSTANCE = ConnectionItemCEdge

ConnectionItemCNode

Bases: DomainModel

This represents the reading version of connection item c node.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the connection item c node.	required
data_record		The data record of the connection item c node node.	required
connection_item_a		The connection item a field.	required
connection_item_b		The connection item b field.	required

Source code in examples/omni/data_classes/_connection_item_c_node.py

class ConnectionItemCNode(DomainModel):
    """This represents the reading version of connection item c node.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the connection item c node.
        data_record: The data record of the connection item c node node.
        connection_item_a: The connection item a field.
        connection_item_b: The connection item b field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "ConnectionItemC", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, None] = dm.DirectRelationReference("pygen-models", "ConnectionItemC")
    connection_item_a: Optional[list[Union[ConnectionItemA, str, dm.NodeId]]] = Field(
        default=None, repr=False, alias="connectionItemA"
    )
    connection_item_b: Optional[list[Union[ConnectionItemB, str, dm.NodeId]]] = Field(
        default=None, repr=False, alias="connectionItemB"
    )

    def as_write(self) -> ConnectionItemCNodeWrite:
        """Convert this read version of connection item c node to the writing version."""
        return ConnectionItemCNodeWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            connection_item_a=[
                connection_item_a.as_write() if isinstance(connection_item_a, DomainModel) else connection_item_a
                for connection_item_a in self.connection_item_a or []
            ],
            connection_item_b=[
                connection_item_b.as_write() if isinstance(connection_item_b, DomainModel) else connection_item_b
                for connection_item_b in self.connection_item_b or []
            ],
        )

    def as_apply(self) -> ConnectionItemCNodeWrite:
        """Convert this read version of connection item c node to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

    @classmethod
    def _update_connections(
        cls,
        instances: dict[dm.NodeId | str, ConnectionItemCNode],  # type: ignore[override]
        nodes_by_id: dict[dm.NodeId | str, DomainModel],
        edges_by_source_node: dict[dm.NodeId, list[dm.Edge | DomainRelation]],
    ) -> None:
        from ._connection_item_a import ConnectionItemA
        from ._connection_item_b import ConnectionItemB

        for instance in instances.values():
            if edges := edges_by_source_node.get(instance.as_id()):
                connection_item_a: list[ConnectionItemA | str | dm.NodeId] = []
                connection_item_b: list[ConnectionItemB | str | dm.NodeId] = []
                for edge in edges:
                    value: DomainModel | DomainRelation | str | dm.NodeId
                    if isinstance(edge, DomainRelation):
                        value = edge
                    else:
                        other_end: dm.DirectRelationReference = (
                            edge.end_node
                            if edge.start_node.space == instance.space
                            and edge.start_node.external_id == instance.external_id
                            else edge.start_node
                        )
                        destination: dm.NodeId | str = (
                            as_node_id(other_end)
                            if other_end.space != DEFAULT_INSTANCE_SPACE
                            else other_end.external_id
                        )
                        if destination in nodes_by_id:
                            value = nodes_by_id[destination]
                        else:
                            value = destination
                    edge_type = edge.edge_type if isinstance(edge, DomainRelation) else edge.type

                    if edge_type == dm.DirectRelationReference("pygen-models", "unidirectional") and isinstance(
                        value, (ConnectionItemA, str, dm.NodeId)
                    ):
                        connection_item_a.append(value)
                    if edge_type == dm.DirectRelationReference("pygen-models", "unidirectional") and isinstance(
                        value, (ConnectionItemB, str, dm.NodeId)
                    ):
                        connection_item_b.append(value)

                instance.connection_item_a = connection_item_a or None
                instance.connection_item_b = connection_item_b or None

as_apply()

Convert this read version of connection item c node to the writing version.

Source code in examples/omni/data_classes/_connection_item_c_node.py

def as_apply(self) -> ConnectionItemCNodeWrite:
    """Convert this read version of connection item c node to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of connection item c node to the writing version.

Source code in examples/omni/data_classes/_connection_item_c_node.py

def as_write(self) -> ConnectionItemCNodeWrite:
    """Convert this read version of connection item c node to the writing version."""
    return ConnectionItemCNodeWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        connection_item_a=[
            connection_item_a.as_write() if isinstance(connection_item_a, DomainModel) else connection_item_a
            for connection_item_a in self.connection_item_a or []
        ],
        connection_item_b=[
            connection_item_b.as_write() if isinstance(connection_item_b, DomainModel) else connection_item_b
            for connection_item_b in self.connection_item_b or []
        ],
    )

ConnectionItemCNodeGraphQL

Bases: GraphQLCore

This represents the reading version of connection item c node, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the connection item c node.	required
data_record		The data record of the connection item c node node.	required
connection_item_a		The connection item a field.	required
connection_item_b		The connection item b field.	required

Source code in examples/omni/data_classes/_connection_item_c_node.py

class ConnectionItemCNodeGraphQL(GraphQLCore):
    """This represents the reading version of connection item c node, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the connection item c node.
        data_record: The data record of the connection item c node node.
        connection_item_a: The connection item a field.
        connection_item_b: The connection item b field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "ConnectionItemC", "1")
    connection_item_a: Optional[list[ConnectionItemAGraphQL]] = Field(default=None, repr=False, alias="connectionItemA")
    connection_item_b: Optional[list[ConnectionItemBGraphQL]] = Field(default=None, repr=False, alias="connectionItemB")

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    @field_validator("connection_item_a", "connection_item_b", mode="before")
    def parse_graphql(cls, value: Any) -> Any:
        if not isinstance(value, dict):
            return value
        if "items" in value:
            return value["items"]
        return value

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> ConnectionItemCNode:
        """Convert this GraphQL format of connection item c node to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return ConnectionItemCNode(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            connection_item_a=[connection_item_a.as_read() for connection_item_a in self.connection_item_a or []],
            connection_item_b=[connection_item_b.as_read() for connection_item_b in self.connection_item_b or []],
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> ConnectionItemCNodeWrite:
        """Convert this GraphQL format of connection item c node to the writing format."""
        return ConnectionItemCNodeWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            connection_item_a=[connection_item_a.as_write() for connection_item_a in self.connection_item_a or []],
            connection_item_b=[connection_item_b.as_write() for connection_item_b in self.connection_item_b or []],
        )

as_read()

Convert this GraphQL format of connection item c node to the reading format.

Source code in examples/omni/data_classes/_connection_item_c_node.py

@no_type_check
def as_read(self) -> ConnectionItemCNode:
    """Convert this GraphQL format of connection item c node to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return ConnectionItemCNode(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        connection_item_a=[connection_item_a.as_read() for connection_item_a in self.connection_item_a or []],
        connection_item_b=[connection_item_b.as_read() for connection_item_b in self.connection_item_b or []],
    )

as_write()

Convert this GraphQL format of connection item c node to the writing format.

Source code in examples/omni/data_classes/_connection_item_c_node.py

@no_type_check
def as_write(self) -> ConnectionItemCNodeWrite:
    """Convert this GraphQL format of connection item c node to the writing format."""
    return ConnectionItemCNodeWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        connection_item_a=[connection_item_a.as_write() for connection_item_a in self.connection_item_a or []],
        connection_item_b=[connection_item_b.as_write() for connection_item_b in self.connection_item_b or []],
    )

ConnectionItemCNodeList

Bases: DomainModelList[ConnectionItemCNode]

List of connection item c nodes in the read version.

Source code in examples/omni/data_classes/_connection_item_c_node.py

class ConnectionItemCNodeList(DomainModelList[ConnectionItemCNode]):
    """List of connection item c nodes in the read version."""

    _INSTANCE = ConnectionItemCNode

    def as_write(self) -> ConnectionItemCNodeWriteList:
        """Convert these read versions of connection item c node to the writing versions."""
        return ConnectionItemCNodeWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> ConnectionItemCNodeWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/omni/data_classes/_connection_item_c_node.py

def as_apply(self) -> ConnectionItemCNodeWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of connection item c node to the writing versions.

Source code in examples/omni/data_classes/_connection_item_c_node.py

def as_write(self) -> ConnectionItemCNodeWriteList:
    """Convert these read versions of connection item c node to the writing versions."""
    return ConnectionItemCNodeWriteList([node.as_write() for node in self.data])

ConnectionItemCNodeWrite

Bases: DomainModelWrite

This represents the writing version of connection item c node.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the connection item c node.	required
data_record		The data record of the connection item c node node.	required
connection_item_a		The connection item a field.	required
connection_item_b		The connection item b field.	required

Source code in examples/omni/data_classes/_connection_item_c_node.py

class ConnectionItemCNodeWrite(DomainModelWrite):
    """This represents the writing version of connection item c node.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the connection item c node.
        data_record: The data record of the connection item c node node.
        connection_item_a: The connection item a field.
        connection_item_b: The connection item b field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "ConnectionItemC", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = dm.DirectRelationReference(
        "pygen-models", "ConnectionItemC"
    )
    connection_item_a: Optional[list[Union[ConnectionItemAWrite, str, dm.NodeId]]] = Field(
        default=None, repr=False, alias="connectionItemA"
    )
    connection_item_b: Optional[list[Union[ConnectionItemBWrite, str, dm.NodeId]]] = Field(
        default=None, repr=False, alias="connectionItemB"
    )

    @field_validator("connection_item_a", "connection_item_b", mode="before")
    def as_node_id(cls, value: Any) -> Any:
        if isinstance(value, dm.DirectRelationReference):
            return dm.NodeId(value.space, value.external_id)
        elif isinstance(value, tuple) and len(value) == 2 and all(isinstance(item, str) for item in value):
            return dm.NodeId(value[0], value[1])
        elif isinstance(value, list):
            return [cls.as_node_id(item) for item in value]
        return value

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources
        cache.add(self.as_tuple_id())

        this_node = dm.NodeApply(
            space=self.space,
            external_id=self.external_id,
            existing_version=None if allow_version_increase else self.data_record.existing_version,
            type=as_direct_relation_reference(self.node_type),
            sources=None,
        )
        resources.nodes.append(this_node)

        edge_type = dm.DirectRelationReference("pygen-models", "unidirectional")
        for connection_item_a in self.connection_item_a or []:
            other_resources = DomainRelationWrite.from_edge_to_resources(
                cache,
                start_node=self,
                end_node=connection_item_a,
                edge_type=edge_type,
                write_none=write_none,
                allow_version_increase=allow_version_increase,
            )
            resources.extend(other_resources)

        edge_type = dm.DirectRelationReference("pygen-models", "unidirectional")
        for connection_item_b in self.connection_item_b or []:
            other_resources = DomainRelationWrite.from_edge_to_resources(
                cache,
                start_node=self,
                end_node=connection_item_b,
                edge_type=edge_type,
                write_none=write_none,
                allow_version_increase=allow_version_increase,
            )
            resources.extend(other_resources)

        return resources

ConnectionItemCNodeWriteList

Bases: DomainModelWriteList[ConnectionItemCNodeWrite]

List of connection item c nodes in the writing version.

Source code in examples/omni/data_classes/_connection_item_c_node.py

class ConnectionItemCNodeWriteList(DomainModelWriteList[ConnectionItemCNodeWrite]):
    """List of connection item c nodes in the writing version."""

    _INSTANCE = ConnectionItemCNodeWrite

ConnectionItemD

Bases: DomainModel

This represents the reading version of connection item d.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the connection item d.	required
data_record		The data record of the connection item d node.	required
direct_multi		The direct multi field.	required
direct_single		The direct single field.	required
name		The name field.	required
outwards_single		The outwards single field.	required

Source code in examples/omni/data_classes/_connection_item_d.py

class ConnectionItemD(DomainModel):
    """This represents the reading version of connection item d.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the connection item d.
        data_record: The data record of the connection item d node.
        direct_multi: The direct multi field.
        direct_single: The direct single field.
        name: The name field.
        outwards_single: The outwards single field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "ConnectionItemD", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, None] = dm.DirectRelationReference("pygen-models", "ConnectionItemD")
    direct_multi: Union[ConnectionItemE, str, dm.NodeId, None] = Field(default=None, repr=False, alias="directMulti")
    direct_single: Union[ConnectionItemE, str, dm.NodeId, None] = Field(default=None, repr=False, alias="directSingle")
    name: Optional[str] = None
    outwards_single: Union[ConnectionItemE, str, dm.NodeId, None] = Field(
        default=None, repr=False, alias="outwardsSingle"
    )

    def as_write(self) -> ConnectionItemDWrite:
        """Convert this read version of connection item d to the writing version."""
        return ConnectionItemDWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            direct_multi=(
                self.direct_multi.as_write() if isinstance(self.direct_multi, DomainModel) else self.direct_multi
            ),
            direct_single=(
                self.direct_single.as_write() if isinstance(self.direct_single, DomainModel) else self.direct_single
            ),
            name=self.name,
            outwards_single=(
                self.outwards_single.as_write()
                if isinstance(self.outwards_single, DomainModel)
                else self.outwards_single
            ),
        )

    def as_apply(self) -> ConnectionItemDWrite:
        """Convert this read version of connection item d to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

    @classmethod
    def _update_connections(
        cls,
        instances: dict[dm.NodeId | str, ConnectionItemD],  # type: ignore[override]
        nodes_by_id: dict[dm.NodeId | str, DomainModel],
        edges_by_source_node: dict[dm.NodeId, list[dm.Edge | DomainRelation]],
    ) -> None:
        from ._connection_item_e import ConnectionItemE

        for instance in instances.values():
            if (
                isinstance(instance.direct_multi, (dm.NodeId, str))
                and (direct_multi := nodes_by_id.get(instance.direct_multi))
                and isinstance(direct_multi, ConnectionItemE)
            ):
                instance.direct_multi = direct_multi
            if (
                isinstance(instance.direct_single, (dm.NodeId, str))
                and (direct_single := nodes_by_id.get(instance.direct_single))
                and isinstance(direct_single, ConnectionItemE)
            ):
                instance.direct_single = direct_single
            if edges := edges_by_source_node.get(instance.as_id()):
                for edge in edges:
                    value: DomainModel | DomainRelation | str | dm.NodeId
                    if isinstance(edge, DomainRelation):
                        value = edge
                    else:
                        other_end: dm.DirectRelationReference = (
                            edge.end_node
                            if edge.start_node.space == instance.space
                            and edge.start_node.external_id == instance.external_id
                            else edge.start_node
                        )
                        destination: dm.NodeId | str = (
                            as_node_id(other_end)
                            if other_end.space != DEFAULT_INSTANCE_SPACE
                            else other_end.external_id
                        )
                        if destination in nodes_by_id:
                            value = nodes_by_id[destination]
                        else:
                            value = destination
                    edge_type = edge.edge_type if isinstance(edge, DomainRelation) else edge.type

                    if edge_type == dm.DirectRelationReference("pygen-models", "bidirectionalSingle") and isinstance(
                        value, (ConnectionItemE, str, dm.NodeId)
                    ):
                        if instance.outwards_single is None:
                            instance.outwards_single = value
                        elif are_nodes_equal(value, instance.outwards_single):
                            instance.outwards_single = select_best_node(value, instance.outwards_single)
                        else:
                            warnings.warn(
                                f"Expected one edge for 'outwards_single' in {instance.as_id()}."
                                f"Ignoring new edge {value!s} in favor of {instance.outwards_single!s}."
                            )

as_apply()

Convert this read version of connection item d to the writing version.

Source code in examples/omni/data_classes/_connection_item_d.py

def as_apply(self) -> ConnectionItemDWrite:
    """Convert this read version of connection item d to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of connection item d to the writing version.

Source code in examples/omni/data_classes/_connection_item_d.py

def as_write(self) -> ConnectionItemDWrite:
    """Convert this read version of connection item d to the writing version."""
    return ConnectionItemDWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        direct_multi=(
            self.direct_multi.as_write() if isinstance(self.direct_multi, DomainModel) else self.direct_multi
        ),
        direct_single=(
            self.direct_single.as_write() if isinstance(self.direct_single, DomainModel) else self.direct_single
        ),
        name=self.name,
        outwards_single=(
            self.outwards_single.as_write()
            if isinstance(self.outwards_single, DomainModel)
            else self.outwards_single
        ),
    )

ConnectionItemDGraphQL

Bases: GraphQLCore

This represents the reading version of connection item d, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the connection item d.	required
data_record		The data record of the connection item d node.	required
direct_multi		The direct multi field.	required
direct_single		The direct single field.	required
name		The name field.	required
outwards_single		The outwards single field.	required

Source code in examples/omni/data_classes/_connection_item_d.py

class ConnectionItemDGraphQL(GraphQLCore):
    """This represents the reading version of connection item d, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the connection item d.
        data_record: The data record of the connection item d node.
        direct_multi: The direct multi field.
        direct_single: The direct single field.
        name: The name field.
        outwards_single: The outwards single field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "ConnectionItemD", "1")
    direct_multi: Optional[ConnectionItemEGraphQL] = Field(default=None, repr=False, alias="directMulti")
    direct_single: Optional[ConnectionItemEGraphQL] = Field(default=None, repr=False, alias="directSingle")
    name: Optional[str] = None
    outwards_single: Optional[ConnectionItemEGraphQL] = Field(default=None, repr=False, alias="outwardsSingle")

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    @field_validator("direct_multi", "direct_single", "outwards_single", mode="before")
    def parse_graphql(cls, value: Any) -> Any:
        if not isinstance(value, dict):
            return value
        if "items" in value:
            return value["items"]
        return value

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> ConnectionItemD:
        """Convert this GraphQL format of connection item d to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return ConnectionItemD(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            direct_multi=(
                self.direct_multi.as_read() if isinstance(self.direct_multi, GraphQLCore) else self.direct_multi
            ),
            direct_single=(
                self.direct_single.as_read() if isinstance(self.direct_single, GraphQLCore) else self.direct_single
            ),
            name=self.name,
            outwards_single=(
                self.outwards_single.as_read()
                if isinstance(self.outwards_single, GraphQLCore)
                else self.outwards_single
            ),
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> ConnectionItemDWrite:
        """Convert this GraphQL format of connection item d to the writing format."""
        return ConnectionItemDWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            direct_multi=(
                self.direct_multi.as_write() if isinstance(self.direct_multi, GraphQLCore) else self.direct_multi
            ),
            direct_single=(
                self.direct_single.as_write() if isinstance(self.direct_single, GraphQLCore) else self.direct_single
            ),
            name=self.name,
            outwards_single=(
                self.outwards_single.as_write()
                if isinstance(self.outwards_single, GraphQLCore)
                else self.outwards_single
            ),
        )

as_read()

Convert this GraphQL format of connection item d to the reading format.

Source code in examples/omni/data_classes/_connection_item_d.py

@no_type_check
def as_read(self) -> ConnectionItemD:
    """Convert this GraphQL format of connection item d to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return ConnectionItemD(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        direct_multi=(
            self.direct_multi.as_read() if isinstance(self.direct_multi, GraphQLCore) else self.direct_multi
        ),
        direct_single=(
            self.direct_single.as_read() if isinstance(self.direct_single, GraphQLCore) else self.direct_single
        ),
        name=self.name,
        outwards_single=(
            self.outwards_single.as_read()
            if isinstance(self.outwards_single, GraphQLCore)
            else self.outwards_single
        ),
    )

as_write()

Convert this GraphQL format of connection item d to the writing format.

Source code in examples/omni/data_classes/_connection_item_d.py

@no_type_check
def as_write(self) -> ConnectionItemDWrite:
    """Convert this GraphQL format of connection item d to the writing format."""
    return ConnectionItemDWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        direct_multi=(
            self.direct_multi.as_write() if isinstance(self.direct_multi, GraphQLCore) else self.direct_multi
        ),
        direct_single=(
            self.direct_single.as_write() if isinstance(self.direct_single, GraphQLCore) else self.direct_single
        ),
        name=self.name,
        outwards_single=(
            self.outwards_single.as_write()
            if isinstance(self.outwards_single, GraphQLCore)
            else self.outwards_single
        ),
    )

ConnectionItemDList

Bases: DomainModelList[ConnectionItemD]

List of connection item ds in the read version.

Source code in examples/omni/data_classes/_connection_item_d.py

class ConnectionItemDList(DomainModelList[ConnectionItemD]):
    """List of connection item ds in the read version."""

    _INSTANCE = ConnectionItemD

    def as_write(self) -> ConnectionItemDWriteList:
        """Convert these read versions of connection item d to the writing versions."""
        return ConnectionItemDWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> ConnectionItemDWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/omni/data_classes/_connection_item_d.py

def as_apply(self) -> ConnectionItemDWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of connection item d to the writing versions.

Source code in examples/omni/data_classes/_connection_item_d.py

def as_write(self) -> ConnectionItemDWriteList:
    """Convert these read versions of connection item d to the writing versions."""
    return ConnectionItemDWriteList([node.as_write() for node in self.data])

ConnectionItemDWrite

Bases: DomainModelWrite

This represents the writing version of connection item d.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the connection item d.	required
data_record		The data record of the connection item d node.	required
direct_multi		The direct multi field.	required
direct_single		The direct single field.	required
name		The name field.	required
outwards_single		The outwards single field.	required

Source code in examples/omni/data_classes/_connection_item_d.py

class ConnectionItemDWrite(DomainModelWrite):
    """This represents the writing version of connection item d.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the connection item d.
        data_record: The data record of the connection item d node.
        direct_multi: The direct multi field.
        direct_single: The direct single field.
        name: The name field.
        outwards_single: The outwards single field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "ConnectionItemD", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = dm.DirectRelationReference(
        "pygen-models", "ConnectionItemD"
    )
    direct_multi: Union[ConnectionItemEWrite, str, dm.NodeId, None] = Field(
        default=None, repr=False, alias="directMulti"
    )
    direct_single: Union[ConnectionItemEWrite, str, dm.NodeId, None] = Field(
        default=None, repr=False, alias="directSingle"
    )
    name: Optional[str] = None
    outwards_single: Union[ConnectionItemEWrite, str, dm.NodeId, None] = Field(
        default=None, repr=False, alias="outwardsSingle"
    )

    @field_validator("direct_multi", "direct_single", "outwards_single", mode="before")
    def as_node_id(cls, value: Any) -> Any:
        if isinstance(value, dm.DirectRelationReference):
            return dm.NodeId(value.space, value.external_id)
        elif isinstance(value, tuple) and len(value) == 2 and all(isinstance(item, str) for item in value):
            return dm.NodeId(value[0], value[1])
        elif isinstance(value, list):
            return [cls.as_node_id(item) for item in value]
        return value

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.direct_multi is not None:
            properties["directMulti"] = {
                "space": self.space if isinstance(self.direct_multi, str) else self.direct_multi.space,
                "externalId": (
                    self.direct_multi if isinstance(self.direct_multi, str) else self.direct_multi.external_id
                ),
            }

        if self.direct_single is not None:
            properties["directSingle"] = {
                "space": self.space if isinstance(self.direct_single, str) else self.direct_single.space,
                "externalId": (
                    self.direct_single if isinstance(self.direct_single, str) else self.direct_single.external_id
                ),
            }

        if self.name is not None or write_none:
            properties["name"] = self.name

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        if isinstance(self.direct_multi, DomainModelWrite):
            other_resources = self.direct_multi._to_instances_write(cache)
            resources.extend(other_resources)

        if isinstance(self.direct_single, DomainModelWrite):
            other_resources = self.direct_single._to_instances_write(cache)
            resources.extend(other_resources)

        if self.outwards_single is not None:
            other_resources = DomainRelationWrite.from_edge_to_resources(
                cache,
                start_node=self,
                end_node=self.outwards_single,
                edge_type=dm.DirectRelationReference("pygen-models", "bidirectionalSingle"),
                write_none=write_none,
                allow_version_increase=allow_version_increase,
            )
            resources.extend(other_resources)

        return resources

ConnectionItemDWriteList

Bases: DomainModelWriteList[ConnectionItemDWrite]

List of connection item ds in the writing version.

Source code in examples/omni/data_classes/_connection_item_d.py

class ConnectionItemDWriteList(DomainModelWriteList[ConnectionItemDWrite]):
    """List of connection item ds in the writing version."""

    _INSTANCE = ConnectionItemDWrite

ConnectionItemE

Bases: DomainModel

This represents the reading version of connection item e.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the connection item e.	required
data_record		The data record of the connection item e node.	required
direct_no_source		The direct no source field.	required
direct_reverse_multi		The direct reverse multi field.	required
direct_reverse_single		The direct reverse single field.	required
inwards_single		The inwards single field.	required
inwards_single_property		The inwards single property field.	required
name		The name field.	required

Source code in examples/omni/data_classes/_connection_item_e.py

class ConnectionItemE(DomainModel):
    """This represents the reading version of connection item e.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the connection item e.
        data_record: The data record of the connection item e node.
        direct_no_source: The direct no source field.
        direct_reverse_multi: The direct reverse multi field.
        direct_reverse_single: The direct reverse single field.
        inwards_single: The inwards single field.
        inwards_single_property: The inwards single property field.
        name: The name field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "ConnectionItemE", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, None] = dm.DirectRelationReference("pygen-models", "ConnectionItemE")
    direct_no_source: Union[str, dm.NodeId, None] = Field(default=None, alias="directNoSource")
    direct_reverse_multi: Optional[list[ConnectionItemD]] = Field(default=None, repr=False, alias="directReverseMulti")
    direct_reverse_single: Optional[ConnectionItemD] = Field(default=None, repr=False, alias="directReverseSingle")
    inwards_single: Union[ConnectionItemD, str, dm.NodeId, None] = Field(
        default=None, repr=False, alias="inwardsSingle"
    )
    inwards_single_property: Optional[ConnectionEdgeA] = Field(default=None, repr=False, alias="inwardsSingleProperty")
    name: Optional[str] = None

    def as_write(self) -> ConnectionItemEWrite:
        """Convert this read version of connection item e to the writing version."""
        return ConnectionItemEWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            direct_no_source=self.direct_no_source,
            inwards_single=(
                self.inwards_single.as_write() if isinstance(self.inwards_single, DomainModel) else self.inwards_single
            ),
            inwards_single_property=(
                self.inwards_single_property.as_write()
                if isinstance(self.inwards_single_property, DomainRelation)
                else self.inwards_single_property
            ),
            name=self.name,
        )

    def as_apply(self) -> ConnectionItemEWrite:
        """Convert this read version of connection item e to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

    @classmethod
    def _update_connections(
        cls,
        instances: dict[dm.NodeId | str, ConnectionItemE],  # type: ignore[override]
        nodes_by_id: dict[dm.NodeId | str, DomainModel],
        edges_by_source_node: dict[dm.NodeId, list[dm.Edge | DomainRelation]],
    ) -> None:
        from ._connection_edge_a import ConnectionEdgeA
        from ._connection_item_d import ConnectionItemD

        for instance in instances.values():
            if edges := edges_by_source_node.get(instance.as_id()):
                for edge in edges:
                    value: DomainModel | DomainRelation | str | dm.NodeId
                    if isinstance(edge, DomainRelation):
                        value = edge
                    else:
                        other_end: dm.DirectRelationReference = (
                            edge.end_node
                            if edge.start_node.space == instance.space
                            and edge.start_node.external_id == instance.external_id
                            else edge.start_node
                        )
                        destination: dm.NodeId | str = (
                            as_node_id(other_end)
                            if other_end.space != DEFAULT_INSTANCE_SPACE
                            else other_end.external_id
                        )
                        if destination in nodes_by_id:
                            value = nodes_by_id[destination]
                        else:
                            value = destination
                    edge_type = edge.edge_type if isinstance(edge, DomainRelation) else edge.type

                    if edge_type == dm.DirectRelationReference("pygen-models", "bidirectionalSingle") and isinstance(
                        value, (ConnectionItemD, str, dm.NodeId)
                    ):
                        if instance.inwards_single is None:
                            instance.inwards_single = value
                        elif are_nodes_equal(value, instance.inwards_single):
                            instance.inwards_single = select_best_node(value, instance.inwards_single)
                        else:
                            warnings.warn(
                                f"Expected one edge for 'inwards_single' in {instance.as_id()}."
                                f"Ignoring new edge {value!s} in favor of {instance.inwards_single!s}."
                            )
                    if edge_type == dm.DirectRelationReference("pygen-models", "multiProperty") and isinstance(
                        value, ConnectionEdgeA
                    ):
                        if instance.inwards_single_property is None:
                            instance.inwards_single_property = value
                        elif instance.inwards_single_property == value:
                            # This is the same edge, so we don't need to do anything...
                            ...
                        else:
                            warnings.warn(
                                f"Expected one edge for 'inwards_single_property' in {instance.as_id()}."
                                f"Ignoring new edge {value!s} in favor of {instance.inwards_single_property!s}."
                            )

                        if end_node := nodes_by_id.get(as_pygen_node_id(value.end_node)):
                            value.end_node = end_node  # type: ignore[assignment]

        for node in nodes_by_id.values():
            if (
                isinstance(node, ConnectionItemD)
                and node.direct_single is not None
                and (direct_single := instances.get(as_pygen_node_id(node.direct_single)))
            ):
                node.direct_single = direct_single
                if direct_single.direct_reverse_single is None:
                    direct_single.direct_reverse_single = node
                elif are_nodes_equal(node, direct_single.direct_reverse_single):
                    # This is the same node, so we don't need to do anything...
                    ...
                else:
                    warnings.warn(
                        f"Expected one direct relation for 'direct_reverse_single' in {direct_single.as_id()}."
                        f"Ignoring new relation {node!s} in favor of {direct_single.direct_reverse_single!s}."
                    )
            if (
                isinstance(node, ConnectionItemD)
                and node.direct_multi is not None
                and (direct_multi := instances.get(as_pygen_node_id(node.direct_multi)))
            ):
                node.direct_multi = direct_multi
                if direct_multi.direct_reverse_multi is None:
                    direct_multi.direct_reverse_multi = []
                direct_multi.direct_reverse_multi.append(node)

as_apply()

Convert this read version of connection item e to the writing version.

Source code in examples/omni/data_classes/_connection_item_e.py

def as_apply(self) -> ConnectionItemEWrite:
    """Convert this read version of connection item e to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of connection item e to the writing version.

Source code in examples/omni/data_classes/_connection_item_e.py

def as_write(self) -> ConnectionItemEWrite:
    """Convert this read version of connection item e to the writing version."""
    return ConnectionItemEWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        direct_no_source=self.direct_no_source,
        inwards_single=(
            self.inwards_single.as_write() if isinstance(self.inwards_single, DomainModel) else self.inwards_single
        ),
        inwards_single_property=(
            self.inwards_single_property.as_write()
            if isinstance(self.inwards_single_property, DomainRelation)
            else self.inwards_single_property
        ),
        name=self.name,
    )

ConnectionItemEGraphQL

Bases: GraphQLCore

This represents the reading version of connection item e, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the connection item e.	required
data_record		The data record of the connection item e node.	required
direct_no_source		The direct no source field.	required
direct_reverse_multi		The direct reverse multi field.	required
direct_reverse_single		The direct reverse single field.	required
inwards_single		The inwards single field.	required
inwards_single_property		The inwards single property field.	required
name		The name field.	required

Source code in examples/omni/data_classes/_connection_item_e.py

class ConnectionItemEGraphQL(GraphQLCore):
    """This represents the reading version of connection item e, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the connection item e.
        data_record: The data record of the connection item e node.
        direct_no_source: The direct no source field.
        direct_reverse_multi: The direct reverse multi field.
        direct_reverse_single: The direct reverse single field.
        inwards_single: The inwards single field.
        inwards_single_property: The inwards single property field.
        name: The name field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "ConnectionItemE", "1")
    direct_no_source: Optional[str] = Field(default=None, alias="directNoSource")
    direct_reverse_multi: Optional[list[ConnectionItemDGraphQL]] = Field(
        default=None, repr=False, alias="directReverseMulti"
    )
    direct_reverse_single: Optional[ConnectionItemDGraphQL] = Field(
        default=None, repr=False, alias="directReverseSingle"
    )
    inwards_single: Optional[ConnectionItemDGraphQL] = Field(default=None, repr=False, alias="inwardsSingle")
    inwards_single_property: Optional[ConnectionEdgeAGraphQL] = Field(
        default=None, repr=False, alias="inwardsSingleProperty"
    )
    name: Optional[str] = None

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    @field_validator(
        "direct_no_source",
        "direct_reverse_multi",
        "direct_reverse_single",
        "inwards_single",
        "inwards_single_property",
        mode="before",
    )
    def parse_graphql(cls, value: Any) -> Any:
        if not isinstance(value, dict):
            return value
        if "items" in value:
            return value["items"]
        return value

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> ConnectionItemE:
        """Convert this GraphQL format of connection item e to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return ConnectionItemE(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            direct_no_source=self.direct_no_source,
            direct_reverse_multi=[
                direct_reverse_multi.as_read() for direct_reverse_multi in self.direct_reverse_multi or []
            ],
            direct_reverse_single=(
                self.direct_reverse_single.as_read()
                if isinstance(self.direct_reverse_single, GraphQLCore)
                else self.direct_reverse_single
            ),
            inwards_single=(
                self.inwards_single.as_read() if isinstance(self.inwards_single, GraphQLCore) else self.inwards_single
            ),
            inwards_single_property=(
                self.inwards_single_property.as_read()
                if isinstance(self.inwards_single_property, GraphQLCore)
                else self.inwards_single_property
            ),
            name=self.name,
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> ConnectionItemEWrite:
        """Convert this GraphQL format of connection item e to the writing format."""
        return ConnectionItemEWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            direct_no_source=self.direct_no_source,
            inwards_single=(
                self.inwards_single.as_write() if isinstance(self.inwards_single, GraphQLCore) else self.inwards_single
            ),
            inwards_single_property=(
                self.inwards_single_property.as_write()
                if isinstance(self.inwards_single_property, GraphQLCore)
                else self.inwards_single_property
            ),
            name=self.name,
        )

as_read()

Convert this GraphQL format of connection item e to the reading format.

Source code in examples/omni/data_classes/_connection_item_e.py

@no_type_check
def as_read(self) -> ConnectionItemE:
    """Convert this GraphQL format of connection item e to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return ConnectionItemE(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        direct_no_source=self.direct_no_source,
        direct_reverse_multi=[
            direct_reverse_multi.as_read() for direct_reverse_multi in self.direct_reverse_multi or []
        ],
        direct_reverse_single=(
            self.direct_reverse_single.as_read()
            if isinstance(self.direct_reverse_single, GraphQLCore)
            else self.direct_reverse_single
        ),
        inwards_single=(
            self.inwards_single.as_read() if isinstance(self.inwards_single, GraphQLCore) else self.inwards_single
        ),
        inwards_single_property=(
            self.inwards_single_property.as_read()
            if isinstance(self.inwards_single_property, GraphQLCore)
            else self.inwards_single_property
        ),
        name=self.name,
    )

as_write()

Convert this GraphQL format of connection item e to the writing format.

Source code in examples/omni/data_classes/_connection_item_e.py

@no_type_check
def as_write(self) -> ConnectionItemEWrite:
    """Convert this GraphQL format of connection item e to the writing format."""
    return ConnectionItemEWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        direct_no_source=self.direct_no_source,
        inwards_single=(
            self.inwards_single.as_write() if isinstance(self.inwards_single, GraphQLCore) else self.inwards_single
        ),
        inwards_single_property=(
            self.inwards_single_property.as_write()
            if isinstance(self.inwards_single_property, GraphQLCore)
            else self.inwards_single_property
        ),
        name=self.name,
    )

ConnectionItemEList

Bases: DomainModelList[ConnectionItemE]

List of connection item es in the read version.

Source code in examples/omni/data_classes/_connection_item_e.py

class ConnectionItemEList(DomainModelList[ConnectionItemE]):
    """List of connection item es in the read version."""

    _INSTANCE = ConnectionItemE

    def as_write(self) -> ConnectionItemEWriteList:
        """Convert these read versions of connection item e to the writing versions."""
        return ConnectionItemEWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> ConnectionItemEWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/omni/data_classes/_connection_item_e.py

def as_apply(self) -> ConnectionItemEWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of connection item e to the writing versions.

Source code in examples/omni/data_classes/_connection_item_e.py

def as_write(self) -> ConnectionItemEWriteList:
    """Convert these read versions of connection item e to the writing versions."""
    return ConnectionItemEWriteList([node.as_write() for node in self.data])

ConnectionItemEWrite

Bases: DomainModelWrite

This represents the writing version of connection item e.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the connection item e.	required
data_record		The data record of the connection item e node.	required
direct_no_source		The direct no source field.	required
inwards_single		The inwards single field.	required
inwards_single_property		The inwards single property field.	required
name		The name field.	required

Source code in examples/omni/data_classes/_connection_item_e.py

class ConnectionItemEWrite(DomainModelWrite):
    """This represents the writing version of connection item e.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the connection item e.
        data_record: The data record of the connection item e node.
        direct_no_source: The direct no source field.
        inwards_single: The inwards single field.
        inwards_single_property: The inwards single property field.
        name: The name field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "ConnectionItemE", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = dm.DirectRelationReference(
        "pygen-models", "ConnectionItemE"
    )
    direct_no_source: Union[str, dm.NodeId, None] = Field(default=None, alias="directNoSource")
    inwards_single: Union[ConnectionItemDWrite, str, dm.NodeId, None] = Field(
        default=None, repr=False, alias="inwardsSingle"
    )
    inwards_single_property: Optional[ConnectionEdgeAWrite] = Field(
        default=None, repr=False, alias="inwardsSingleProperty"
    )
    name: Optional[str] = None

    @field_validator("inwards_single", "inwards_single_property", mode="before")
    def as_node_id(cls, value: Any) -> Any:
        if isinstance(value, dm.DirectRelationReference):
            return dm.NodeId(value.space, value.external_id)
        elif isinstance(value, tuple) and len(value) == 2 and all(isinstance(item, str) for item in value):
            return dm.NodeId(value[0], value[1])
        elif isinstance(value, list):
            return [cls.as_node_id(item) for item in value]
        return value

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.direct_no_source is not None:
            properties["directNoSource"] = {
                "space": self.space if isinstance(self.direct_no_source, str) else self.direct_no_source.space,
                "externalId": (
                    self.direct_no_source
                    if isinstance(self.direct_no_source, str)
                    else self.direct_no_source.external_id
                ),
            }

        if self.name is not None or write_none:
            properties["name"] = self.name

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        if self.inwards_single_property is not None:
            other_resources = self.inwards_single_property._to_instances_write(
                cache,
                self,
                dm.DirectRelationReference("pygen-models", "multiProperty"),
            )
            resources.extend(other_resources)

        if self.inwards_single is not None:
            other_resources = DomainRelationWrite.from_edge_to_resources(
                cache,
                start_node=self.inwards_single,
                end_node=self,
                edge_type=dm.DirectRelationReference("pygen-models", "bidirectionalSingle"),
                write_none=write_none,
                allow_version_increase=allow_version_increase,
            )
            resources.extend(other_resources)

        return resources

ConnectionItemEWriteList

Bases: DomainModelWriteList[ConnectionItemEWrite]

List of connection item es in the writing version.

Source code in examples/omni/data_classes/_connection_item_e.py

class ConnectionItemEWriteList(DomainModelWriteList[ConnectionItemEWrite]):
    """List of connection item es in the writing version."""

    _INSTANCE = ConnectionItemEWrite

ConnectionItemF

Bases: DomainModel

This represents the reading version of connection item f.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the connection item f.	required
data_record		The data record of the connection item f node.	required
direct_list		The direct list field.	required
name		The name field.	required
outwards_multi		The outwards multi field.	required
outwards_single		The outwards single field.	required

Source code in examples/omni/data_classes/_connection_item_f.py

class ConnectionItemF(DomainModel):
    """This represents the reading version of connection item f.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the connection item f.
        data_record: The data record of the connection item f node.
        direct_list: The direct list field.
        name: The name field.
        outwards_multi: The outwards multi field.
        outwards_single: The outwards single field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "ConnectionItemF", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, None] = dm.DirectRelationReference("pygen-models", "ConnectionItemF")
    direct_list: Optional[list[Union[ConnectionItemD, str, dm.NodeId]]] = Field(
        default=None, repr=False, alias="directList"
    )
    name: Optional[str] = None
    outwards_multi: Optional[list[ConnectionEdgeA]] = Field(default=None, repr=False, alias="outwardsMulti")
    outwards_single: Optional[ConnectionEdgeA] = Field(default=None, repr=False, alias="outwardsSingle")

    def as_write(self) -> ConnectionItemFWrite:
        """Convert this read version of connection item f to the writing version."""
        return ConnectionItemFWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            direct_list=[
                direct_list.as_write() if isinstance(direct_list, DomainModel) else direct_list
                for direct_list in self.direct_list or []
            ],
            name=self.name,
            outwards_multi=[outwards_multi.as_write() for outwards_multi in self.outwards_multi or []],
            outwards_single=(
                self.outwards_single.as_write()
                if isinstance(self.outwards_single, DomainRelation)
                else self.outwards_single
            ),
        )

    def as_apply(self) -> ConnectionItemFWrite:
        """Convert this read version of connection item f to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

    @classmethod
    def _update_connections(
        cls,
        instances: dict[dm.NodeId | str, ConnectionItemF],  # type: ignore[override]
        nodes_by_id: dict[dm.NodeId | str, DomainModel],
        edges_by_source_node: dict[dm.NodeId, list[dm.Edge | DomainRelation]],
    ) -> None:
        from ._connection_edge_a import ConnectionEdgeA
        from ._connection_item_d import ConnectionItemD

        for instance in instances.values():
            if instance.direct_list:
                new_direct_list: list[ConnectionItemD | str | dm.NodeId] = []
                for relation in instance.direct_list:
                    if isinstance(relation, ConnectionItemD):
                        new_direct_list.append(relation)
                    elif (other := nodes_by_id.get(relation)) and isinstance(other, ConnectionItemD):
                        new_direct_list.append(other)
                    else:
                        new_direct_list.append(relation)
                instance.direct_list = new_direct_list
            if edges := edges_by_source_node.get(instance.as_id()):
                outwards_multi: list[ConnectionEdgeA] = []
                for edge in edges:
                    value: DomainModel | DomainRelation | str | dm.NodeId
                    if isinstance(edge, DomainRelation):
                        value = edge
                    else:
                        other_end: dm.DirectRelationReference = (
                            edge.end_node
                            if edge.start_node.space == instance.space
                            and edge.start_node.external_id == instance.external_id
                            else edge.start_node
                        )
                        destination: dm.NodeId | str = (
                            as_node_id(other_end)
                            if other_end.space != DEFAULT_INSTANCE_SPACE
                            else other_end.external_id
                        )
                        if destination in nodes_by_id:
                            value = nodes_by_id[destination]
                        else:
                            value = destination
                    edge_type = edge.edge_type if isinstance(edge, DomainRelation) else edge.type

                    if edge_type == dm.DirectRelationReference("pygen-models", "multiProperty") and isinstance(
                        value, ConnectionEdgeA
                    ):
                        outwards_multi.append(value)
                        if end_node := nodes_by_id.get(as_pygen_node_id(value.end_node)):
                            value.end_node = end_node  # type: ignore[assignment]
                    if edge_type == dm.DirectRelationReference("pygen-models", "singleProperty") and isinstance(
                        value, ConnectionEdgeA
                    ):
                        if instance.outwards_single is None:
                            instance.outwards_single = value
                        elif instance.outwards_single == value:
                            # This is the same edge, so we don't need to do anything...
                            ...
                        else:
                            warnings.warn(
                                f"Expected one edge for 'outwards_single' in {instance.as_id()}."
                                f"Ignoring new edge {value!s} in favor of {instance.outwards_single!s}."
                            )

                        if end_node := nodes_by_id.get(as_pygen_node_id(value.end_node)):
                            value.end_node = end_node  # type: ignore[assignment]

                instance.outwards_multi = outwards_multi

as_apply()

Convert this read version of connection item f to the writing version.

Source code in examples/omni/data_classes/_connection_item_f.py

def as_apply(self) -> ConnectionItemFWrite:
    """Convert this read version of connection item f to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of connection item f to the writing version.

Source code in examples/omni/data_classes/_connection_item_f.py

def as_write(self) -> ConnectionItemFWrite:
    """Convert this read version of connection item f to the writing version."""
    return ConnectionItemFWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        direct_list=[
            direct_list.as_write() if isinstance(direct_list, DomainModel) else direct_list
            for direct_list in self.direct_list or []
        ],
        name=self.name,
        outwards_multi=[outwards_multi.as_write() for outwards_multi in self.outwards_multi or []],
        outwards_single=(
            self.outwards_single.as_write()
            if isinstance(self.outwards_single, DomainRelation)
            else self.outwards_single
        ),
    )

ConnectionItemFGraphQL

Bases: GraphQLCore

This represents the reading version of connection item f, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the connection item f.	required
data_record		The data record of the connection item f node.	required
direct_list		The direct list field.	required
name		The name field.	required
outwards_multi		The outwards multi field.	required
outwards_single		The outwards single field.	required

Source code in examples/omni/data_classes/_connection_item_f.py

class ConnectionItemFGraphQL(GraphQLCore):
    """This represents the reading version of connection item f, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the connection item f.
        data_record: The data record of the connection item f node.
        direct_list: The direct list field.
        name: The name field.
        outwards_multi: The outwards multi field.
        outwards_single: The outwards single field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "ConnectionItemF", "1")
    direct_list: Optional[list[ConnectionItemDGraphQL]] = Field(default=None, repr=False, alias="directList")
    name: Optional[str] = None
    outwards_multi: Optional[list[ConnectionEdgeAGraphQL]] = Field(default=None, repr=False, alias="outwardsMulti")
    outwards_single: Optional[ConnectionEdgeAGraphQL] = Field(default=None, repr=False, alias="outwardsSingle")

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    @field_validator("direct_list", "outwards_multi", "outwards_single", mode="before")
    def parse_graphql(cls, value: Any) -> Any:
        if not isinstance(value, dict):
            return value
        if "items" in value:
            return value["items"]
        return value

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> ConnectionItemF:
        """Convert this GraphQL format of connection item f to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return ConnectionItemF(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            direct_list=[direct_list.as_read() for direct_list in self.direct_list or []],
            name=self.name,
            outwards_multi=[outwards_multi.as_read() for outwards_multi in self.outwards_multi or []],
            outwards_single=(
                self.outwards_single.as_read()
                if isinstance(self.outwards_single, GraphQLCore)
                else self.outwards_single
            ),
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> ConnectionItemFWrite:
        """Convert this GraphQL format of connection item f to the writing format."""
        return ConnectionItemFWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            direct_list=[direct_list.as_write() for direct_list in self.direct_list or []],
            name=self.name,
            outwards_multi=[outwards_multi.as_write() for outwards_multi in self.outwards_multi or []],
            outwards_single=(
                self.outwards_single.as_write()
                if isinstance(self.outwards_single, GraphQLCore)
                else self.outwards_single
            ),
        )

as_read()

Convert this GraphQL format of connection item f to the reading format.

Source code in examples/omni/data_classes/_connection_item_f.py

@no_type_check
def as_read(self) -> ConnectionItemF:
    """Convert this GraphQL format of connection item f to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return ConnectionItemF(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        direct_list=[direct_list.as_read() for direct_list in self.direct_list or []],
        name=self.name,
        outwards_multi=[outwards_multi.as_read() for outwards_multi in self.outwards_multi or []],
        outwards_single=(
            self.outwards_single.as_read()
            if isinstance(self.outwards_single, GraphQLCore)
            else self.outwards_single
        ),
    )

as_write()

Convert this GraphQL format of connection item f to the writing format.

Source code in examples/omni/data_classes/_connection_item_f.py

@no_type_check
def as_write(self) -> ConnectionItemFWrite:
    """Convert this GraphQL format of connection item f to the writing format."""
    return ConnectionItemFWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        direct_list=[direct_list.as_write() for direct_list in self.direct_list or []],
        name=self.name,
        outwards_multi=[outwards_multi.as_write() for outwards_multi in self.outwards_multi or []],
        outwards_single=(
            self.outwards_single.as_write()
            if isinstance(self.outwards_single, GraphQLCore)
            else self.outwards_single
        ),
    )

ConnectionItemFList

Bases: DomainModelList[ConnectionItemF]

List of connection item fs in the read version.

Source code in examples/omni/data_classes/_connection_item_f.py

class ConnectionItemFList(DomainModelList[ConnectionItemF]):
    """List of connection item fs in the read version."""

    _INSTANCE = ConnectionItemF

    def as_write(self) -> ConnectionItemFWriteList:
        """Convert these read versions of connection item f to the writing versions."""
        return ConnectionItemFWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> ConnectionItemFWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/omni/data_classes/_connection_item_f.py

def as_apply(self) -> ConnectionItemFWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of connection item f to the writing versions.

Source code in examples/omni/data_classes/_connection_item_f.py

def as_write(self) -> ConnectionItemFWriteList:
    """Convert these read versions of connection item f to the writing versions."""
    return ConnectionItemFWriteList([node.as_write() for node in self.data])

ConnectionItemFWrite

Bases: DomainModelWrite

This represents the writing version of connection item f.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the connection item f.	required
data_record		The data record of the connection item f node.	required
direct_list		The direct list field.	required
name		The name field.	required
outwards_multi		The outwards multi field.	required
outwards_single		The outwards single field.	required

Source code in examples/omni/data_classes/_connection_item_f.py

class ConnectionItemFWrite(DomainModelWrite):
    """This represents the writing version of connection item f.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the connection item f.
        data_record: The data record of the connection item f node.
        direct_list: The direct list field.
        name: The name field.
        outwards_multi: The outwards multi field.
        outwards_single: The outwards single field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "ConnectionItemF", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = dm.DirectRelationReference(
        "pygen-models", "ConnectionItemF"
    )
    direct_list: Optional[list[Union[ConnectionItemDWrite, str, dm.NodeId]]] = Field(
        default=None, repr=False, alias="directList"
    )
    name: Optional[str] = None
    outwards_multi: Optional[list[ConnectionEdgeAWrite]] = Field(default=None, repr=False, alias="outwardsMulti")
    outwards_single: Optional[ConnectionEdgeAWrite] = Field(default=None, repr=False, alias="outwardsSingle")

    @field_validator("direct_list", "outwards_multi", "outwards_single", mode="before")
    def as_node_id(cls, value: Any) -> Any:
        if isinstance(value, dm.DirectRelationReference):
            return dm.NodeId(value.space, value.external_id)
        elif isinstance(value, tuple) and len(value) == 2 and all(isinstance(item, str) for item in value):
            return dm.NodeId(value[0], value[1])
        elif isinstance(value, list):
            return [cls.as_node_id(item) for item in value]
        return value

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.direct_list is not None:
            properties["directList"] = [
                {
                    "space": self.space if isinstance(direct_list, str) else direct_list.space,
                    "externalId": direct_list if isinstance(direct_list, str) else direct_list.external_id,
                }
                for direct_list in self.direct_list or []
            ]

        if self.name is not None or write_none:
            properties["name"] = self.name

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        for outwards_multi in self.outwards_multi or []:
            if isinstance(outwards_multi, DomainRelationWrite):
                other_resources = outwards_multi._to_instances_write(
                    cache,
                    self,
                    dm.DirectRelationReference("pygen-models", "multiProperty"),
                )
                resources.extend(other_resources)

        if self.outwards_single is not None:
            other_resources = self.outwards_single._to_instances_write(
                cache,
                self,
                dm.DirectRelationReference("pygen-models", "singleProperty"),
            )
            resources.extend(other_resources)

        for direct_list in self.direct_list or []:
            if isinstance(direct_list, DomainModelWrite):
                other_resources = direct_list._to_instances_write(cache)
                resources.extend(other_resources)

        return resources

ConnectionItemFWriteList

Bases: DomainModelWriteList[ConnectionItemFWrite]

List of connection item fs in the writing version.

Source code in examples/omni/data_classes/_connection_item_f.py

class ConnectionItemFWriteList(DomainModelWriteList[ConnectionItemFWrite]):
    """List of connection item fs in the writing version."""

    _INSTANCE = ConnectionItemFWrite

ConnectionItemG

Bases: DomainModel

This represents the reading version of connection item g.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the connection item g.	required
data_record		The data record of the connection item g node.	required
inwards_multi_property		The inwards multi property field.	required
name		The name field.	required

Source code in examples/omni/data_classes/_connection_item_g.py

class ConnectionItemG(DomainModel):
    """This represents the reading version of connection item g.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the connection item g.
        data_record: The data record of the connection item g node.
        inwards_multi_property: The inwards multi property field.
        name: The name field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "ConnectionItemG", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, None] = dm.DirectRelationReference("pygen-models", "ConnectionItemG")
    inwards_multi_property: Optional[list[ConnectionEdgeA]] = Field(
        default=None, repr=False, alias="inwardsMultiProperty"
    )
    name: Optional[str] = None

    def as_write(self) -> ConnectionItemGWrite:
        """Convert this read version of connection item g to the writing version."""
        return ConnectionItemGWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            inwards_multi_property=[
                inwards_multi_property.as_write() for inwards_multi_property in self.inwards_multi_property or []
            ],
            name=self.name,
        )

    def as_apply(self) -> ConnectionItemGWrite:
        """Convert this read version of connection item g to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

    @classmethod
    def _update_connections(
        cls,
        instances: dict[dm.NodeId | str, ConnectionItemG],  # type: ignore[override]
        nodes_by_id: dict[dm.NodeId | str, DomainModel],
        edges_by_source_node: dict[dm.NodeId, list[dm.Edge | DomainRelation]],
    ) -> None:
        from ._connection_edge_a import ConnectionEdgeA

        for instance in instances.values():
            if edges := edges_by_source_node.get(instance.as_id()):
                inwards_multi_property: list[ConnectionEdgeA] = []
                for edge in edges:
                    value: DomainModel | DomainRelation | str | dm.NodeId
                    if isinstance(edge, DomainRelation):
                        value = edge
                    else:
                        other_end: dm.DirectRelationReference = (
                            edge.end_node
                            if edge.start_node.space == instance.space
                            and edge.start_node.external_id == instance.external_id
                            else edge.start_node
                        )
                        destination: dm.NodeId | str = (
                            as_node_id(other_end)
                            if other_end.space != DEFAULT_INSTANCE_SPACE
                            else other_end.external_id
                        )
                        if destination in nodes_by_id:
                            value = nodes_by_id[destination]
                        else:
                            value = destination
                    edge_type = edge.edge_type if isinstance(edge, DomainRelation) else edge.type

                    if edge_type == dm.DirectRelationReference("pygen-models", "multiProperty") and isinstance(
                        value, ConnectionEdgeA
                    ):
                        inwards_multi_property.append(value)
                        if end_node := nodes_by_id.get(as_pygen_node_id(value.end_node)):
                            value.end_node = end_node  # type: ignore[assignment]

                instance.inwards_multi_property = inwards_multi_property

as_apply()

Convert this read version of connection item g to the writing version.

Source code in examples/omni/data_classes/_connection_item_g.py

def as_apply(self) -> ConnectionItemGWrite:
    """Convert this read version of connection item g to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of connection item g to the writing version.

Source code in examples/omni/data_classes/_connection_item_g.py

def as_write(self) -> ConnectionItemGWrite:
    """Convert this read version of connection item g to the writing version."""
    return ConnectionItemGWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        inwards_multi_property=[
            inwards_multi_property.as_write() for inwards_multi_property in self.inwards_multi_property or []
        ],
        name=self.name,
    )

ConnectionItemGGraphQL

Bases: GraphQLCore

This represents the reading version of connection item g, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the connection item g.	required
data_record		The data record of the connection item g node.	required
inwards_multi_property		The inwards multi property field.	required
name		The name field.	required

Source code in examples/omni/data_classes/_connection_item_g.py

class ConnectionItemGGraphQL(GraphQLCore):
    """This represents the reading version of connection item g, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the connection item g.
        data_record: The data record of the connection item g node.
        inwards_multi_property: The inwards multi property field.
        name: The name field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "ConnectionItemG", "1")
    inwards_multi_property: Optional[list[ConnectionEdgeAGraphQL]] = Field(
        default=None, repr=False, alias="inwardsMultiProperty"
    )
    name: Optional[str] = None

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    @field_validator("inwards_multi_property", mode="before")
    def parse_graphql(cls, value: Any) -> Any:
        if not isinstance(value, dict):
            return value
        if "items" in value:
            return value["items"]
        return value

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> ConnectionItemG:
        """Convert this GraphQL format of connection item g to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return ConnectionItemG(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            inwards_multi_property=[
                inwards_multi_property.as_read() for inwards_multi_property in self.inwards_multi_property or []
            ],
            name=self.name,
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> ConnectionItemGWrite:
        """Convert this GraphQL format of connection item g to the writing format."""
        return ConnectionItemGWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            inwards_multi_property=[
                inwards_multi_property.as_write() for inwards_multi_property in self.inwards_multi_property or []
            ],
            name=self.name,
        )

as_read()

Convert this GraphQL format of connection item g to the reading format.

Source code in examples/omni/data_classes/_connection_item_g.py

@no_type_check
def as_read(self) -> ConnectionItemG:
    """Convert this GraphQL format of connection item g to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return ConnectionItemG(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        inwards_multi_property=[
            inwards_multi_property.as_read() for inwards_multi_property in self.inwards_multi_property or []
        ],
        name=self.name,
    )

as_write()

Convert this GraphQL format of connection item g to the writing format.

Source code in examples/omni/data_classes/_connection_item_g.py

@no_type_check
def as_write(self) -> ConnectionItemGWrite:
    """Convert this GraphQL format of connection item g to the writing format."""
    return ConnectionItemGWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        inwards_multi_property=[
            inwards_multi_property.as_write() for inwards_multi_property in self.inwards_multi_property or []
        ],
        name=self.name,
    )

ConnectionItemGList

Bases: DomainModelList[ConnectionItemG]

List of connection item gs in the read version.

Source code in examples/omni/data_classes/_connection_item_g.py

class ConnectionItemGList(DomainModelList[ConnectionItemG]):
    """List of connection item gs in the read version."""

    _INSTANCE = ConnectionItemG

    def as_write(self) -> ConnectionItemGWriteList:
        """Convert these read versions of connection item g to the writing versions."""
        return ConnectionItemGWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> ConnectionItemGWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/omni/data_classes/_connection_item_g.py

def as_apply(self) -> ConnectionItemGWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of connection item g to the writing versions.

Source code in examples/omni/data_classes/_connection_item_g.py

def as_write(self) -> ConnectionItemGWriteList:
    """Convert these read versions of connection item g to the writing versions."""
    return ConnectionItemGWriteList([node.as_write() for node in self.data])

ConnectionItemGWrite

Bases: DomainModelWrite

This represents the writing version of connection item g.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the connection item g.	required
data_record		The data record of the connection item g node.	required
inwards_multi_property		The inwards multi property field.	required
name		The name field.	required

Source code in examples/omni/data_classes/_connection_item_g.py

class ConnectionItemGWrite(DomainModelWrite):
    """This represents the writing version of connection item g.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the connection item g.
        data_record: The data record of the connection item g node.
        inwards_multi_property: The inwards multi property field.
        name: The name field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "ConnectionItemG", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = dm.DirectRelationReference(
        "pygen-models", "ConnectionItemG"
    )
    inwards_multi_property: Optional[list[ConnectionEdgeAWrite]] = Field(
        default=None, repr=False, alias="inwardsMultiProperty"
    )
    name: Optional[str] = None

    @field_validator("inwards_multi_property", mode="before")
    def as_node_id(cls, value: Any) -> Any:
        if isinstance(value, dm.DirectRelationReference):
            return dm.NodeId(value.space, value.external_id)
        elif isinstance(value, tuple) and len(value) == 2 and all(isinstance(item, str) for item in value):
            return dm.NodeId(value[0], value[1])
        elif isinstance(value, list):
            return [cls.as_node_id(item) for item in value]
        return value

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.name is not None or write_none:
            properties["name"] = self.name

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        for inwards_multi_property in self.inwards_multi_property or []:
            if isinstance(inwards_multi_property, DomainRelationWrite):
                other_resources = inwards_multi_property._to_instances_write(
                    cache,
                    self,
                    dm.DirectRelationReference("pygen-models", "multiProperty"),
                )
                resources.extend(other_resources)

        return resources

ConnectionItemGWriteList

Bases: DomainModelWriteList[ConnectionItemGWrite]

List of connection item gs in the writing version.

Source code in examples/omni/data_classes/_connection_item_g.py

class ConnectionItemGWriteList(DomainModelWriteList[ConnectionItemGWrite]):
    """List of connection item gs in the writing version."""

    _INSTANCE = ConnectionItemGWrite

DataRecord

Bases: BaseModel

The data record represents the metadata of a node.

Parameters:

Name	Type	Description	Default
created_time		The created time of the node.	required
last_updated_time		The last updated time of the node.	required
deleted_time		If present, the deleted time of the node.	required
version		The version of the node.	required

Source code in examples/omni/data_classes/_core/base.py

class DataRecord(BaseModel):
    """The data record represents the metadata of a node.

    Args:
        created_time: The created time of the node.
        last_updated_time: The last updated time of the node.
        deleted_time: If present, the deleted time of the node.
        version: The version of the node.
    """

    version: int
    last_updated_time: datetime.datetime
    created_time: datetime.datetime
    deleted_time: Optional[datetime.datetime] = None

DataRecordWrite

Bases: BaseModel

The data record represents the metadata of a node.

Parameters:

Name	Type	Description	Default
existing_version		Fail the ingestion request if the node version is greater than or equal to this value. If no existingVersion is specified, the ingestion will always overwrite any existing data for the edge (for the specified container or instance). If existingVersion is set to 0, the upsert will behave as an insert, so it will fail the bulk if the item already exists. If skipOnVersionConflict is set on the ingestion request, then the item will be skipped instead of failing the ingestion request.	required

Source code in examples/omni/data_classes/_core/base.py

class DataRecordWrite(BaseModel):
    """The data record represents the metadata of a node.

    Args:
        existing_version: Fail the ingestion request if the node version is greater than or equal to this value.
            If no existingVersion is specified, the ingestion will always overwrite any existing data for the edge (for the specified container or instance).
            If existingVersion is set to 0, the upsert will behave as an insert, so it will fail the bulk if the item already exists.
            If skipOnVersionConflict is set on the ingestion request, then the item will be skipped instead of failing the ingestion request.
    """

    existing_version: Optional[int] = None

DependentOnNonWritable

Bases: DomainModel

This represents the reading version of dependent on non writable.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the dependent on non writable.	required
data_record		The data record of the dependent on non writable node.	required
a_value		The a value field.	required
to_non_writable		The to non writable field.	required

Source code in examples/omni/data_classes/_dependent_on_non_writable.py

class DependentOnNonWritable(DomainModel):
    """This represents the reading version of dependent on non writable.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the dependent on non writable.
        data_record: The data record of the dependent on non writable node.
        a_value: The a value field.
        to_non_writable: The to non writable field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "DependentOnNonWritable", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, None] = dm.DirectRelationReference(
        "pygen-models", "DependentOnNonWritable"
    )
    a_value: Optional[str] = Field(None, alias="aValue")
    to_non_writable: Optional[list[Union[Implementation1NonWriteable, str, dm.NodeId]]] = Field(
        default=None, repr=False, alias="toNonWritable"
    )

    def as_write(self) -> DependentOnNonWritableWrite:
        """Convert this read version of dependent on non writable to the writing version."""
        return DependentOnNonWritableWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            a_value=self.a_value,
            to_non_writable=[
                to_non_writable.as_id() if isinstance(to_non_writable, DomainModel) else to_non_writable
                for to_non_writable in self.to_non_writable or []
            ],
        )

    def as_apply(self) -> DependentOnNonWritableWrite:
        """Convert this read version of dependent on non writable to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

    @classmethod
    def _update_connections(
        cls,
        instances: dict[dm.NodeId | str, DependentOnNonWritable],  # type: ignore[override]
        nodes_by_id: dict[dm.NodeId | str, DomainModel],
        edges_by_source_node: dict[dm.NodeId, list[dm.Edge | DomainRelation]],
    ) -> None:
        from ._implementation_1_non_writeable import Implementation1NonWriteable

        for instance in instances.values():
            if edges := edges_by_source_node.get(instance.as_id()):
                to_non_writable: list[Implementation1NonWriteable | str | dm.NodeId] = []
                for edge in edges:
                    value: DomainModel | DomainRelation | str | dm.NodeId
                    if isinstance(edge, DomainRelation):
                        value = edge
                    else:
                        other_end: dm.DirectRelationReference = (
                            edge.end_node
                            if edge.start_node.space == instance.space
                            and edge.start_node.external_id == instance.external_id
                            else edge.start_node
                        )
                        destination: dm.NodeId | str = (
                            as_node_id(other_end)
                            if other_end.space != DEFAULT_INSTANCE_SPACE
                            else other_end.external_id
                        )
                        if destination in nodes_by_id:
                            value = nodes_by_id[destination]
                        else:
                            value = destination
                    edge_type = edge.edge_type if isinstance(edge, DomainRelation) else edge.type

                    if edge_type == dm.DirectRelationReference("pygen-models", "toNonWritable") and isinstance(
                        value, (Implementation1NonWriteable, str, dm.NodeId)
                    ):
                        to_non_writable.append(value)

                instance.to_non_writable = to_non_writable or None

as_apply()

Convert this read version of dependent on non writable to the writing version.

Source code in examples/omni/data_classes/_dependent_on_non_writable.py

def as_apply(self) -> DependentOnNonWritableWrite:
    """Convert this read version of dependent on non writable to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of dependent on non writable to the writing version.

Source code in examples/omni/data_classes/_dependent_on_non_writable.py

def as_write(self) -> DependentOnNonWritableWrite:
    """Convert this read version of dependent on non writable to the writing version."""
    return DependentOnNonWritableWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        a_value=self.a_value,
        to_non_writable=[
            to_non_writable.as_id() if isinstance(to_non_writable, DomainModel) else to_non_writable
            for to_non_writable in self.to_non_writable or []
        ],
    )

DependentOnNonWritableGraphQL

Bases: GraphQLCore

This represents the reading version of dependent on non writable, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the dependent on non writable.	required
data_record		The data record of the dependent on non writable node.	required
a_value		The a value field.	required
to_non_writable		The to non writable field.	required

Source code in examples/omni/data_classes/_dependent_on_non_writable.py

class DependentOnNonWritableGraphQL(GraphQLCore):
    """This represents the reading version of dependent on non writable, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the dependent on non writable.
        data_record: The data record of the dependent on non writable node.
        a_value: The a value field.
        to_non_writable: The to non writable field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "DependentOnNonWritable", "1")
    a_value: Optional[str] = Field(None, alias="aValue")
    to_non_writable: Optional[list[Implementation1NonWriteableGraphQL]] = Field(
        default=None, repr=False, alias="toNonWritable"
    )

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    @field_validator("to_non_writable", mode="before")
    def parse_graphql(cls, value: Any) -> Any:
        if not isinstance(value, dict):
            return value
        if "items" in value:
            return value["items"]
        return value

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> DependentOnNonWritable:
        """Convert this GraphQL format of dependent on non writable to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return DependentOnNonWritable(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            a_value=self.a_value,
            to_non_writable=[to_non_writable.as_read() for to_non_writable in self.to_non_writable or []],
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> DependentOnNonWritableWrite:
        """Convert this GraphQL format of dependent on non writable to the writing format."""
        return DependentOnNonWritableWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            a_value=self.a_value,
            to_non_writable=[to_non_writable.as_write() for to_non_writable in self.to_non_writable or []],
        )

as_read()

Convert this GraphQL format of dependent on non writable to the reading format.

Source code in examples/omni/data_classes/_dependent_on_non_writable.py

@no_type_check
def as_read(self) -> DependentOnNonWritable:
    """Convert this GraphQL format of dependent on non writable to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return DependentOnNonWritable(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        a_value=self.a_value,
        to_non_writable=[to_non_writable.as_read() for to_non_writable in self.to_non_writable or []],
    )

as_write()

Convert this GraphQL format of dependent on non writable to the writing format.

Source code in examples/omni/data_classes/_dependent_on_non_writable.py

@no_type_check
def as_write(self) -> DependentOnNonWritableWrite:
    """Convert this GraphQL format of dependent on non writable to the writing format."""
    return DependentOnNonWritableWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        a_value=self.a_value,
        to_non_writable=[to_non_writable.as_write() for to_non_writable in self.to_non_writable or []],
    )

DependentOnNonWritableList

Bases: DomainModelList[DependentOnNonWritable]

List of dependent on non writables in the read version.

Source code in examples/omni/data_classes/_dependent_on_non_writable.py

class DependentOnNonWritableList(DomainModelList[DependentOnNonWritable]):
    """List of dependent on non writables in the read version."""

    _INSTANCE = DependentOnNonWritable

    def as_write(self) -> DependentOnNonWritableWriteList:
        """Convert these read versions of dependent on non writable to the writing versions."""
        return DependentOnNonWritableWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> DependentOnNonWritableWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/omni/data_classes/_dependent_on_non_writable.py

def as_apply(self) -> DependentOnNonWritableWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of dependent on non writable to the writing versions.

Source code in examples/omni/data_classes/_dependent_on_non_writable.py

def as_write(self) -> DependentOnNonWritableWriteList:
    """Convert these read versions of dependent on non writable to the writing versions."""
    return DependentOnNonWritableWriteList([node.as_write() for node in self.data])

DependentOnNonWritableWrite

Bases: DomainModelWrite

This represents the writing version of dependent on non writable.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the dependent on non writable.	required
data_record		The data record of the dependent on non writable node.	required
a_value		The a value field.	required
to_non_writable		The to non writable field.	required

Source code in examples/omni/data_classes/_dependent_on_non_writable.py

class DependentOnNonWritableWrite(DomainModelWrite):
    """This represents the writing version of dependent on non writable.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the dependent on non writable.
        data_record: The data record of the dependent on non writable node.
        a_value: The a value field.
        to_non_writable: The to non writable field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "DependentOnNonWritable", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = dm.DirectRelationReference(
        "pygen-models", "DependentOnNonWritable"
    )
    a_value: Optional[str] = Field(None, alias="aValue")
    to_non_writable: Optional[list[Union[str, dm.NodeId]]] = Field(default=None, alias="toNonWritable")

    @field_validator("to_non_writable", mode="before")
    def as_node_id(cls, value: Any) -> Any:
        if isinstance(value, dm.DirectRelationReference):
            return dm.NodeId(value.space, value.external_id)
        elif isinstance(value, tuple) and len(value) == 2 and all(isinstance(item, str) for item in value):
            return dm.NodeId(value[0], value[1])
        elif isinstance(value, list):
            return [cls.as_node_id(item) for item in value]
        return value

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.a_value is not None or write_none:
            properties["aValue"] = self.a_value

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        edge_type = dm.DirectRelationReference("pygen-models", "toNonWritable")
        for to_non_writable in self.to_non_writable or []:
            other_resources = DomainRelationWrite.from_edge_to_resources(
                cache,
                start_node=self,
                end_node=to_non_writable,
                edge_type=edge_type,
                write_none=write_none,
                allow_version_increase=allow_version_increase,
            )
            resources.extend(other_resources)

        return resources

DependentOnNonWritableWriteList

Bases: DomainModelWriteList[DependentOnNonWritableWrite]

List of dependent on non writables in the writing version.

Source code in examples/omni/data_classes/_dependent_on_non_writable.py

class DependentOnNonWritableWriteList(DomainModelWriteList[DependentOnNonWritableWrite]):
    """List of dependent on non writables in the writing version."""

    _INSTANCE = DependentOnNonWritableWrite

Empty

Bases: DomainModel

This represents the reading version of empty.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the empty.	required
data_record		The data record of the empty node.	required
boolean		The boolean field.	required
date		The date field.	required
float_32		The float 32 field.	required
float_64		The float 64 field.	required
int_32		The int 32 field.	required
int_64		The int 64 field.	required
json_		The json field.	required
text		The text field.	required
timestamp		The timestamp field.	required

Source code in examples/omni/data_classes/_empty.py

class Empty(DomainModel):
    """This represents the reading version of empty.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the empty.
        data_record: The data record of the empty node.
        boolean: The boolean field.
        date: The date field.
        float_32: The float 32 field.
        float_64: The float 64 field.
        int_32: The int 32 field.
        int_64: The int 64 field.
        json_: The json field.
        text: The text field.
        timestamp: The timestamp field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "Empty", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, None] = dm.DirectRelationReference("pygen-models", "Empty")
    boolean: Optional[bool] = None
    date: Optional[datetime.date] = None
    float_32: Optional[float] = Field(None, alias="float32")
    float_64: Optional[float] = Field(None, alias="float64")
    int_32: Optional[int] = Field(None, alias="int32")
    int_64: Optional[int] = Field(None, alias="int64")
    json_: Optional[dict] = Field(None, alias="json")
    text: Optional[str] = None
    timestamp: Optional[datetime.datetime] = None

    def as_write(self) -> EmptyWrite:
        """Convert this read version of empty to the writing version."""
        return EmptyWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            boolean=self.boolean,
            date=self.date,
            float_32=self.float_32,
            float_64=self.float_64,
            int_32=self.int_32,
            int_64=self.int_64,
            json_=self.json_,
            text=self.text,
            timestamp=self.timestamp,
        )

    def as_apply(self) -> EmptyWrite:
        """Convert this read version of empty to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert this read version of empty to the writing version.

Source code in examples/omni/data_classes/_empty.py

def as_apply(self) -> EmptyWrite:
    """Convert this read version of empty to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of empty to the writing version.

Source code in examples/omni/data_classes/_empty.py

def as_write(self) -> EmptyWrite:
    """Convert this read version of empty to the writing version."""
    return EmptyWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        boolean=self.boolean,
        date=self.date,
        float_32=self.float_32,
        float_64=self.float_64,
        int_32=self.int_32,
        int_64=self.int_64,
        json_=self.json_,
        text=self.text,
        timestamp=self.timestamp,
    )

EmptyGraphQL

Bases: GraphQLCore

This represents the reading version of empty, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the empty.	required
data_record		The data record of the empty node.	required
boolean		The boolean field.	required
date		The date field.	required
float_32		The float 32 field.	required
float_64		The float 64 field.	required
int_32		The int 32 field.	required
int_64		The int 64 field.	required
json_		The json field.	required
text		The text field.	required
timestamp		The timestamp field.	required

Source code in examples/omni/data_classes/_empty.py

class EmptyGraphQL(GraphQLCore):
    """This represents the reading version of empty, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the empty.
        data_record: The data record of the empty node.
        boolean: The boolean field.
        date: The date field.
        float_32: The float 32 field.
        float_64: The float 64 field.
        int_32: The int 32 field.
        int_64: The int 64 field.
        json_: The json field.
        text: The text field.
        timestamp: The timestamp field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "Empty", "1")
    boolean: Optional[bool] = None
    date: Optional[datetime.date] = None
    float_32: Optional[float] = Field(None, alias="float32")
    float_64: Optional[float] = Field(None, alias="float64")
    int_32: Optional[int] = Field(None, alias="int32")
    int_64: Optional[int] = Field(None, alias="int64")
    json_: Optional[dict] = Field(None, alias="json")
    text: Optional[str] = None
    timestamp: Optional[datetime.datetime] = None

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> Empty:
        """Convert this GraphQL format of empty to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return Empty(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            boolean=self.boolean,
            date=self.date,
            float_32=self.float_32,
            float_64=self.float_64,
            int_32=self.int_32,
            int_64=self.int_64,
            json_=self.json_,
            text=self.text,
            timestamp=self.timestamp,
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> EmptyWrite:
        """Convert this GraphQL format of empty to the writing format."""
        return EmptyWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            boolean=self.boolean,
            date=self.date,
            float_32=self.float_32,
            float_64=self.float_64,
            int_32=self.int_32,
            int_64=self.int_64,
            json_=self.json_,
            text=self.text,
            timestamp=self.timestamp,
        )

as_read()

Convert this GraphQL format of empty to the reading format.

Source code in examples/omni/data_classes/_empty.py

@no_type_check
def as_read(self) -> Empty:
    """Convert this GraphQL format of empty to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return Empty(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        boolean=self.boolean,
        date=self.date,
        float_32=self.float_32,
        float_64=self.float_64,
        int_32=self.int_32,
        int_64=self.int_64,
        json_=self.json_,
        text=self.text,
        timestamp=self.timestamp,
    )

as_write()

Convert this GraphQL format of empty to the writing format.

Source code in examples/omni/data_classes/_empty.py

@no_type_check
def as_write(self) -> EmptyWrite:
    """Convert this GraphQL format of empty to the writing format."""
    return EmptyWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        boolean=self.boolean,
        date=self.date,
        float_32=self.float_32,
        float_64=self.float_64,
        int_32=self.int_32,
        int_64=self.int_64,
        json_=self.json_,
        text=self.text,
        timestamp=self.timestamp,
    )

EmptyList

Bases: DomainModelList[Empty]

List of empties in the read version.

Source code in examples/omni/data_classes/_empty.py

class EmptyList(DomainModelList[Empty]):
    """List of empties in the read version."""

    _INSTANCE = Empty

    def as_write(self) -> EmptyWriteList:
        """Convert these read versions of empty to the writing versions."""
        return EmptyWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> EmptyWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/omni/data_classes/_empty.py

def as_apply(self) -> EmptyWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of empty to the writing versions.

Source code in examples/omni/data_classes/_empty.py

def as_write(self) -> EmptyWriteList:
    """Convert these read versions of empty to the writing versions."""
    return EmptyWriteList([node.as_write() for node in self.data])

EmptyWrite

Bases: DomainModelWrite

This represents the writing version of empty.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the empty.	required
data_record		The data record of the empty node.	required
boolean		The boolean field.	required
date		The date field.	required
float_32		The float 32 field.	required
float_64		The float 64 field.	required
int_32		The int 32 field.	required
int_64		The int 64 field.	required
json_		The json field.	required
text		The text field.	required
timestamp		The timestamp field.	required

Source code in examples/omni/data_classes/_empty.py

class EmptyWrite(DomainModelWrite):
    """This represents the writing version of empty.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the empty.
        data_record: The data record of the empty node.
        boolean: The boolean field.
        date: The date field.
        float_32: The float 32 field.
        float_64: The float 64 field.
        int_32: The int 32 field.
        int_64: The int 64 field.
        json_: The json field.
        text: The text field.
        timestamp: The timestamp field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "Empty", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = dm.DirectRelationReference(
        "pygen-models", "Empty"
    )
    boolean: Optional[bool] = None
    date: Optional[datetime.date] = None
    float_32: Optional[float] = Field(None, alias="float32")
    float_64: Optional[float] = Field(None, alias="float64")
    int_32: Optional[int] = Field(None, alias="int32")
    int_64: Optional[int] = Field(None, alias="int64")
    json_: Optional[dict] = Field(None, alias="json")
    text: Optional[str] = None
    timestamp: Optional[datetime.datetime] = None

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.boolean is not None or write_none:
            properties["boolean"] = self.boolean

        if self.date is not None or write_none:
            properties["date"] = self.date.isoformat() if self.date else None

        if self.float_32 is not None or write_none:
            properties["float32"] = self.float_32

        if self.float_64 is not None or write_none:
            properties["float64"] = self.float_64

        if self.int_32 is not None or write_none:
            properties["int32"] = self.int_32

        if self.int_64 is not None or write_none:
            properties["int64"] = self.int_64

        if self.json_ is not None or write_none:
            properties["json"] = self.json_

        if self.text is not None or write_none:
            properties["text"] = self.text

        if self.timestamp is not None or write_none:
            properties["timestamp"] = self.timestamp.isoformat(timespec="milliseconds") if self.timestamp else None

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        return resources

EmptyWriteList

Bases: DomainModelWriteList[EmptyWrite]

List of empties in the writing version.

Source code in examples/omni/data_classes/_empty.py

class EmptyWriteList(DomainModelWriteList[EmptyWrite]):
    """List of empties in the writing version."""

    _INSTANCE = EmptyWrite

GraphQLList

Bases: UserList

Source code in examples/omni/data_classes/_core/base.py

class GraphQLList(UserList):
    def __init__(self, nodes: Collection[GraphQLCore] | None = None):
        super().__init__(nodes or [])
        self.page_info: PageInfo | None = None

    # The dunder implementations are to get proper type hints
    def __iter__(self) -> Iterator[GraphQLCore]:
        return super().__iter__()

    @overload
    def __getitem__(self, item: SupportsIndex) -> GraphQLCore: ...

    @overload
    def __getitem__(self, item: slice) -> GraphQLList: ...

    def __getitem__(self, item: SupportsIndex | slice) -> GraphQLCore | GraphQLList:
        value = self.data[item]
        if isinstance(item, slice):
            return type(self)(value)
        return cast(GraphQLCore, value)

    def dump(self) -> list[dict[str, Any]]:
        return [node.model_dump() for node in self.data]

    def to_pandas(self) -> pd.DataFrame:
        """
        Convert the list of nodes to a pandas.DataFrame.

        Returns:
            A pandas.DataFrame with the nodes as rows.
        """
        df = pd.DataFrame(self.dump())
        if df.empty:
            df = pd.DataFrame(columns=GraphQLCore.model_fields)
        # Reorder columns to have the most relevant first
        id_columns = ["space", "external_id"]
        end_columns = ["data_record"]
        fixed_columns = set(id_columns + end_columns)
        columns = (
            id_columns + [col for col in df if col not in fixed_columns] + [col for col in end_columns if col in df]
        )
        return df[columns]

    def _repr_html_(self) -> str:
        return self.to_pandas()._repr_html_()  # type: ignore[operator]

to_pandas()

Convert the list of nodes to a pandas.DataFrame.

Returns:

Type	Description
DataFrame	A pandas.DataFrame with the nodes as rows.

Source code in examples/omni/data_classes/_core/base.py

def to_pandas(self) -> pd.DataFrame:
    """
    Convert the list of nodes to a pandas.DataFrame.

    Returns:
        A pandas.DataFrame with the nodes as rows.
    """
    df = pd.DataFrame(self.dump())
    if df.empty:
        df = pd.DataFrame(columns=GraphQLCore.model_fields)
    # Reorder columns to have the most relevant first
    id_columns = ["space", "external_id"]
    end_columns = ["data_record"]
    fixed_columns = set(id_columns + end_columns)
    columns = (
        id_columns + [col for col in df if col not in fixed_columns] + [col for col in end_columns if col in df]
    )
    return df[columns]

Implementation1

Bases: SubInterface

This represents the reading version of implementation 1.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the implementation 1.	required
data_record		The data record of the implementation 1 node.	required
main_value		The main value field.	required
sub_value		The sub value field.	required
value_1		The value 1 field.	required
value_2		The value 2 field.	required

Source code in examples/omni/data_classes/_implementation_1.py

class Implementation1(SubInterface):
    """This represents the reading version of implementation 1.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the implementation 1.
        data_record: The data record of the implementation 1 node.
        main_value: The main value field.
        sub_value: The sub value field.
        value_1: The value 1 field.
        value_2: The value 2 field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "Implementation1", "1")

    node_type: Union[dm.DirectRelationReference, None] = dm.DirectRelationReference("pygen-models", "Implementation1")
    value_1: Optional[str] = Field(None, alias="value1")
    value_2: str = Field(alias="value2")

    def as_write(self) -> Implementation1Write:
        """Convert this read version of implementation 1 to the writing version."""
        return Implementation1Write(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            main_value=self.main_value,
            sub_value=self.sub_value,
            value_1=self.value_1,
            value_2=self.value_2,
        )

    def as_apply(self) -> Implementation1Write:
        """Convert this read version of implementation 1 to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert this read version of implementation 1 to the writing version.

Source code in examples/omni/data_classes/_implementation_1.py

def as_apply(self) -> Implementation1Write:
    """Convert this read version of implementation 1 to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of implementation 1 to the writing version.

Source code in examples/omni/data_classes/_implementation_1.py

def as_write(self) -> Implementation1Write:
    """Convert this read version of implementation 1 to the writing version."""
    return Implementation1Write(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        main_value=self.main_value,
        sub_value=self.sub_value,
        value_1=self.value_1,
        value_2=self.value_2,
    )

Implementation1GraphQL

Bases: GraphQLCore

This represents the reading version of implementation 1, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the implementation 1.	required
data_record		The data record of the implementation 1 node.	required
main_value		The main value field.	required
sub_value		The sub value field.	required
value_1		The value 1 field.	required
value_2		The value 2 field.	required

Source code in examples/omni/data_classes/_implementation_1.py

class Implementation1GraphQL(GraphQLCore):
    """This represents the reading version of implementation 1, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the implementation 1.
        data_record: The data record of the implementation 1 node.
        main_value: The main value field.
        sub_value: The sub value field.
        value_1: The value 1 field.
        value_2: The value 2 field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "Implementation1", "1")
    main_value: Optional[str] = Field(None, alias="mainValue")
    sub_value: Optional[str] = Field(None, alias="subValue")
    value_1: Optional[str] = Field(None, alias="value1")
    value_2: Optional[str] = Field(None, alias="value2")

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> Implementation1:
        """Convert this GraphQL format of implementation 1 to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return Implementation1(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            main_value=self.main_value,
            sub_value=self.sub_value,
            value_1=self.value_1,
            value_2=self.value_2,
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> Implementation1Write:
        """Convert this GraphQL format of implementation 1 to the writing format."""
        return Implementation1Write(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            main_value=self.main_value,
            sub_value=self.sub_value,
            value_1=self.value_1,
            value_2=self.value_2,
        )

as_read()

Convert this GraphQL format of implementation 1 to the reading format.

Source code in examples/omni/data_classes/_implementation_1.py

@no_type_check
def as_read(self) -> Implementation1:
    """Convert this GraphQL format of implementation 1 to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return Implementation1(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        main_value=self.main_value,
        sub_value=self.sub_value,
        value_1=self.value_1,
        value_2=self.value_2,
    )

as_write()

Convert this GraphQL format of implementation 1 to the writing format.

Source code in examples/omni/data_classes/_implementation_1.py

@no_type_check
def as_write(self) -> Implementation1Write:
    """Convert this GraphQL format of implementation 1 to the writing format."""
    return Implementation1Write(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        main_value=self.main_value,
        sub_value=self.sub_value,
        value_1=self.value_1,
        value_2=self.value_2,
    )

Implementation1List

Bases: DomainModelList[Implementation1]

List of implementation 1 in the read version.

Source code in examples/omni/data_classes/_implementation_1.py

class Implementation1List(DomainModelList[Implementation1]):
    """List of implementation 1 in the read version."""

    _INSTANCE = Implementation1

    def as_write(self) -> Implementation1WriteList:
        """Convert these read versions of implementation 1 to the writing versions."""
        return Implementation1WriteList([node.as_write() for node in self.data])

    def as_apply(self) -> Implementation1WriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/omni/data_classes/_implementation_1.py

def as_apply(self) -> Implementation1WriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of implementation 1 to the writing versions.

Source code in examples/omni/data_classes/_implementation_1.py

def as_write(self) -> Implementation1WriteList:
    """Convert these read versions of implementation 1 to the writing versions."""
    return Implementation1WriteList([node.as_write() for node in self.data])

Implementation1NonWriteable

Bases: SubInterface

This represents the reading version of implementation 1 non writeable.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the implementation 1 non writeable.	required
data_record		The data record of the implementation 1 non writeable node.	required
main_value		The main value field.	required
sub_value		The sub value field.	required
value_1		The value 1 field.	required

Source code in examples/omni/data_classes/_implementation_1_non_writeable.py

class Implementation1NonWriteable(SubInterface):
    """This represents the reading version of implementation 1 non writeable.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the implementation 1 non writeable.
        data_record: The data record of the implementation 1 non writeable node.
        main_value: The main value field.
        sub_value: The sub value field.
        value_1: The value 1 field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "Implementation1NonWriteable", "1")

    node_type: Union[dm.DirectRelationReference, None] = dm.DirectRelationReference("pygen-models", "Implementation1")
    value_1: Optional[str] = Field(None, alias="value1")

Implementation1NonWriteableGraphQL

Bases: GraphQLCore

This represents the reading version of implementation 1 non writeable, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the implementation 1 non writeable.	required
data_record		The data record of the implementation 1 non writeable node.	required
main_value		The main value field.	required
sub_value		The sub value field.	required
value_1		The value 1 field.	required

Source code in examples/omni/data_classes/_implementation_1_non_writeable.py

class Implementation1NonWriteableGraphQL(GraphQLCore):
    """This represents the reading version of implementation 1 non writeable, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the implementation 1 non writeable.
        data_record: The data record of the implementation 1 non writeable node.
        main_value: The main value field.
        sub_value: The sub value field.
        value_1: The value 1 field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "Implementation1NonWriteable", "1")
    main_value: Optional[str] = Field(None, alias="mainValue")
    sub_value: Optional[str] = Field(None, alias="subValue")
    value_1: Optional[str] = Field(None, alias="value1")

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> Implementation1NonWriteable:
        """Convert this GraphQL format of implementation 1 non writeable to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return Implementation1NonWriteable(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            main_value=self.main_value,
            sub_value=self.sub_value,
            value_1=self.value_1,
        )

as_read()

Convert this GraphQL format of implementation 1 non writeable to the reading format.

Source code in examples/omni/data_classes/_implementation_1_non_writeable.py

@no_type_check
def as_read(self) -> Implementation1NonWriteable:
    """Convert this GraphQL format of implementation 1 non writeable to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return Implementation1NonWriteable(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        main_value=self.main_value,
        sub_value=self.sub_value,
        value_1=self.value_1,
    )

Implementation1NonWriteableList

Bases: DomainModelList[Implementation1NonWriteable]

List of implementation 1 non writeables in the read version.

Source code in examples/omni/data_classes/_implementation_1_non_writeable.py

class Implementation1NonWriteableList(DomainModelList[Implementation1NonWriteable]):
    """List of implementation 1 non writeables in the read version."""

    _INSTANCE = Implementation1NonWriteable

Implementation1Write

Bases: SubInterfaceWrite

This represents the writing version of implementation 1.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the implementation 1.	required
data_record		The data record of the implementation 1 node.	required
main_value		The main value field.	required
sub_value		The sub value field.	required
value_1		The value 1 field.	required
value_2		The value 2 field.	required

Source code in examples/omni/data_classes/_implementation_1.py

class Implementation1Write(SubInterfaceWrite):
    """This represents the writing version of implementation 1.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the implementation 1.
        data_record: The data record of the implementation 1 node.
        main_value: The main value field.
        sub_value: The sub value field.
        value_1: The value 1 field.
        value_2: The value 2 field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "Implementation1", "1")

    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = dm.DirectRelationReference(
        "pygen-models", "Implementation1"
    )
    value_1: Optional[str] = Field(None, alias="value1")
    value_2: str = Field(alias="value2")

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.main_value is not None or write_none:
            properties["mainValue"] = self.main_value

        if self.sub_value is not None or write_none:
            properties["subValue"] = self.sub_value

        if self.value_1 is not None or write_none:
            properties["value1"] = self.value_1

        if self.value_2 is not None:
            properties["value2"] = self.value_2

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        return resources

Implementation1WriteList

Bases: DomainModelWriteList[Implementation1Write]

List of implementation 1 in the writing version.

Source code in examples/omni/data_classes/_implementation_1.py

class Implementation1WriteList(DomainModelWriteList[Implementation1Write]):
    """List of implementation 1 in the writing version."""

    _INSTANCE = Implementation1Write

Implementation2

Bases: SubInterface

This represents the reading version of implementation 2.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the implementation 2.	required
data_record		The data record of the implementation 2 node.	required
main_value		The main value field.	required
sub_value		The sub value field.	required

Source code in examples/omni/data_classes/_implementation_2.py

class Implementation2(SubInterface):
    """This represents the reading version of implementation 2.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the implementation 2.
        data_record: The data record of the implementation 2 node.
        main_value: The main value field.
        sub_value: The sub value field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "Implementation2", "1")

    node_type: Union[dm.DirectRelationReference, None] = dm.DirectRelationReference("pygen-models", "Implementation2")

    def as_write(self) -> Implementation2Write:
        """Convert this read version of implementation 2 to the writing version."""
        return Implementation2Write(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            main_value=self.main_value,
            sub_value=self.sub_value,
        )

    def as_apply(self) -> Implementation2Write:
        """Convert this read version of implementation 2 to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert this read version of implementation 2 to the writing version.

Source code in examples/omni/data_classes/_implementation_2.py

def as_apply(self) -> Implementation2Write:
    """Convert this read version of implementation 2 to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of implementation 2 to the writing version.

Source code in examples/omni/data_classes/_implementation_2.py

def as_write(self) -> Implementation2Write:
    """Convert this read version of implementation 2 to the writing version."""
    return Implementation2Write(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        main_value=self.main_value,
        sub_value=self.sub_value,
    )

Implementation2GraphQL

Bases: GraphQLCore

This represents the reading version of implementation 2, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the implementation 2.	required
data_record		The data record of the implementation 2 node.	required
main_value		The main value field.	required
sub_value		The sub value field.	required

Source code in examples/omni/data_classes/_implementation_2.py

class Implementation2GraphQL(GraphQLCore):
    """This represents the reading version of implementation 2, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the implementation 2.
        data_record: The data record of the implementation 2 node.
        main_value: The main value field.
        sub_value: The sub value field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "Implementation2", "1")
    main_value: Optional[str] = Field(None, alias="mainValue")
    sub_value: Optional[str] = Field(None, alias="subValue")

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> Implementation2:
        """Convert this GraphQL format of implementation 2 to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return Implementation2(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            main_value=self.main_value,
            sub_value=self.sub_value,
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> Implementation2Write:
        """Convert this GraphQL format of implementation 2 to the writing format."""
        return Implementation2Write(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            main_value=self.main_value,
            sub_value=self.sub_value,
        )

as_read()

Convert this GraphQL format of implementation 2 to the reading format.

Source code in examples/omni/data_classes/_implementation_2.py

@no_type_check
def as_read(self) -> Implementation2:
    """Convert this GraphQL format of implementation 2 to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return Implementation2(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        main_value=self.main_value,
        sub_value=self.sub_value,
    )

as_write()

Convert this GraphQL format of implementation 2 to the writing format.

Source code in examples/omni/data_classes/_implementation_2.py

@no_type_check
def as_write(self) -> Implementation2Write:
    """Convert this GraphQL format of implementation 2 to the writing format."""
    return Implementation2Write(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        main_value=self.main_value,
        sub_value=self.sub_value,
    )

Implementation2List

Bases: DomainModelList[Implementation2]

List of implementation 2 in the read version.

Source code in examples/omni/data_classes/_implementation_2.py

class Implementation2List(DomainModelList[Implementation2]):
    """List of implementation 2 in the read version."""

    _INSTANCE = Implementation2

    def as_write(self) -> Implementation2WriteList:
        """Convert these read versions of implementation 2 to the writing versions."""
        return Implementation2WriteList([node.as_write() for node in self.data])

    def as_apply(self) -> Implementation2WriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/omni/data_classes/_implementation_2.py

def as_apply(self) -> Implementation2WriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of implementation 2 to the writing versions.

Source code in examples/omni/data_classes/_implementation_2.py

def as_write(self) -> Implementation2WriteList:
    """Convert these read versions of implementation 2 to the writing versions."""
    return Implementation2WriteList([node.as_write() for node in self.data])

Implementation2Write

Bases: SubInterfaceWrite

This represents the writing version of implementation 2.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the implementation 2.	required
data_record		The data record of the implementation 2 node.	required
main_value		The main value field.	required
sub_value		The sub value field.	required

Source code in examples/omni/data_classes/_implementation_2.py

class Implementation2Write(SubInterfaceWrite):
    """This represents the writing version of implementation 2.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the implementation 2.
        data_record: The data record of the implementation 2 node.
        main_value: The main value field.
        sub_value: The sub value field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "Implementation2", "1")

    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = dm.DirectRelationReference(
        "pygen-models", "Implementation2"
    )

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.main_value is not None or write_none:
            properties["mainValue"] = self.main_value

        if self.sub_value is not None or write_none:
            properties["subValue"] = self.sub_value

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        return resources

Implementation2WriteList

Bases: DomainModelWriteList[Implementation2Write]

List of implementation 2 in the writing version.

Source code in examples/omni/data_classes/_implementation_2.py

class Implementation2WriteList(DomainModelWriteList[Implementation2Write]):
    """List of implementation 2 in the writing version."""

    _INSTANCE = Implementation2Write

MainInterface

Bases: DomainModel

This represents the reading version of main interface.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the main interface.	required
data_record		The data record of the main interface node.	required
main_value		The main value field.	required

Source code in examples/omni/data_classes/_main_interface.py

class MainInterface(DomainModel):
    """This represents the reading version of main interface.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the main interface.
        data_record: The data record of the main interface node.
        main_value: The main value field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "MainInterface", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, None] = None
    main_value: Optional[str] = Field(None, alias="mainValue")

    def as_write(self) -> MainInterfaceWrite:
        """Convert this read version of main interface to the writing version."""
        return MainInterfaceWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            main_value=self.main_value,
        )

    def as_apply(self) -> MainInterfaceWrite:
        """Convert this read version of main interface to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert this read version of main interface to the writing version.

Source code in examples/omni/data_classes/_main_interface.py

def as_apply(self) -> MainInterfaceWrite:
    """Convert this read version of main interface to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of main interface to the writing version.

Source code in examples/omni/data_classes/_main_interface.py

def as_write(self) -> MainInterfaceWrite:
    """Convert this read version of main interface to the writing version."""
    return MainInterfaceWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        main_value=self.main_value,
    )

MainInterfaceGraphQL

Bases: GraphQLCore

This represents the reading version of main interface, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the main interface.	required
data_record		The data record of the main interface node.	required
main_value		The main value field.	required

Source code in examples/omni/data_classes/_main_interface.py

class MainInterfaceGraphQL(GraphQLCore):
    """This represents the reading version of main interface, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the main interface.
        data_record: The data record of the main interface node.
        main_value: The main value field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "MainInterface", "1")
    main_value: Optional[str] = Field(None, alias="mainValue")

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> MainInterface:
        """Convert this GraphQL format of main interface to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return MainInterface(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            main_value=self.main_value,
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> MainInterfaceWrite:
        """Convert this GraphQL format of main interface to the writing format."""
        return MainInterfaceWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            main_value=self.main_value,
        )

as_read()

Convert this GraphQL format of main interface to the reading format.

Source code in examples/omni/data_classes/_main_interface.py

@no_type_check
def as_read(self) -> MainInterface:
    """Convert this GraphQL format of main interface to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return MainInterface(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        main_value=self.main_value,
    )

as_write()

Convert this GraphQL format of main interface to the writing format.

Source code in examples/omni/data_classes/_main_interface.py

@no_type_check
def as_write(self) -> MainInterfaceWrite:
    """Convert this GraphQL format of main interface to the writing format."""
    return MainInterfaceWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        main_value=self.main_value,
    )

MainInterfaceList

Bases: DomainModelList[MainInterface]

List of main interfaces in the read version.

Source code in examples/omni/data_classes/_main_interface.py

class MainInterfaceList(DomainModelList[MainInterface]):
    """List of main interfaces in the read version."""

    _INSTANCE = MainInterface

    def as_write(self) -> MainInterfaceWriteList:
        """Convert these read versions of main interface to the writing versions."""
        return MainInterfaceWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> MainInterfaceWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/omni/data_classes/_main_interface.py

def as_apply(self) -> MainInterfaceWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of main interface to the writing versions.

Source code in examples/omni/data_classes/_main_interface.py

def as_write(self) -> MainInterfaceWriteList:
    """Convert these read versions of main interface to the writing versions."""
    return MainInterfaceWriteList([node.as_write() for node in self.data])

MainInterfaceWrite

Bases: DomainModelWrite

This represents the writing version of main interface.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the main interface.	required
data_record		The data record of the main interface node.	required
main_value		The main value field.	required

Source code in examples/omni/data_classes/_main_interface.py

class MainInterfaceWrite(DomainModelWrite):
    """This represents the writing version of main interface.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the main interface.
        data_record: The data record of the main interface node.
        main_value: The main value field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "MainInterface", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = None
    main_value: Optional[str] = Field(None, alias="mainValue")

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.main_value is not None or write_none:
            properties["mainValue"] = self.main_value

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        return resources

MainInterfaceWriteList

Bases: DomainModelWriteList[MainInterfaceWrite]

List of main interfaces in the writing version.

Source code in examples/omni/data_classes/_main_interface.py

class MainInterfaceWriteList(DomainModelWriteList[MainInterfaceWrite]):
    """List of main interfaces in the writing version."""

    _INSTANCE = MainInterfaceWrite

PrimitiveNullable

Bases: DomainModel

This represents the reading version of primitive nullable.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the primitive nullable.	required
data_record		The data record of the primitive nullable node.	required
boolean		The boolean field.	required
date		The date field.	required
float_32		The float 32 field.	required
float_64		The float 64 field.	required
int_32		The int 32 field.	required
int_64		The int 64 field.	required
json_		The json field.	required
text		The text field.	required
timestamp		The timestamp field.	required

Source code in examples/omni/data_classes/_primitive_nullable.py

class PrimitiveNullable(DomainModel):
    """This represents the reading version of primitive nullable.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the primitive nullable.
        data_record: The data record of the primitive nullable node.
        boolean: The boolean field.
        date: The date field.
        float_32: The float 32 field.
        float_64: The float 64 field.
        int_32: The int 32 field.
        int_64: The int 64 field.
        json_: The json field.
        text: The text field.
        timestamp: The timestamp field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "PrimitiveNullable", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, None] = None
    boolean: Optional[bool] = None
    date: Optional[datetime.date] = None
    float_32: Optional[float] = Field(None, alias="float32")
    float_64: Optional[float] = Field(None, alias="float64")
    int_32: Optional[int] = Field(None, alias="int32")
    int_64: Optional[int] = Field(None, alias="int64")
    json_: Optional[dict] = Field(None, alias="json")
    text: Optional[str] = None
    timestamp: Optional[datetime.datetime] = None

    def as_write(self) -> PrimitiveNullableWrite:
        """Convert this read version of primitive nullable to the writing version."""
        return PrimitiveNullableWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            boolean=self.boolean,
            date=self.date,
            float_32=self.float_32,
            float_64=self.float_64,
            int_32=self.int_32,
            int_64=self.int_64,
            json_=self.json_,
            text=self.text,
            timestamp=self.timestamp,
        )

    def as_apply(self) -> PrimitiveNullableWrite:
        """Convert this read version of primitive nullable to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert this read version of primitive nullable to the writing version.

Source code in examples/omni/data_classes/_primitive_nullable.py

def as_apply(self) -> PrimitiveNullableWrite:
    """Convert this read version of primitive nullable to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of primitive nullable to the writing version.

Source code in examples/omni/data_classes/_primitive_nullable.py

def as_write(self) -> PrimitiveNullableWrite:
    """Convert this read version of primitive nullable to the writing version."""
    return PrimitiveNullableWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        boolean=self.boolean,
        date=self.date,
        float_32=self.float_32,
        float_64=self.float_64,
        int_32=self.int_32,
        int_64=self.int_64,
        json_=self.json_,
        text=self.text,
        timestamp=self.timestamp,
    )

PrimitiveNullableGraphQL

Bases: GraphQLCore

This represents the reading version of primitive nullable, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the primitive nullable.	required
data_record		The data record of the primitive nullable node.	required
boolean		The boolean field.	required
date		The date field.	required
float_32		The float 32 field.	required
float_64		The float 64 field.	required
int_32		The int 32 field.	required
int_64		The int 64 field.	required
json_		The json field.	required
text		The text field.	required
timestamp		The timestamp field.	required

Source code in examples/omni/data_classes/_primitive_nullable.py

class PrimitiveNullableGraphQL(GraphQLCore):
    """This represents the reading version of primitive nullable, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the primitive nullable.
        data_record: The data record of the primitive nullable node.
        boolean: The boolean field.
        date: The date field.
        float_32: The float 32 field.
        float_64: The float 64 field.
        int_32: The int 32 field.
        int_64: The int 64 field.
        json_: The json field.
        text: The text field.
        timestamp: The timestamp field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "PrimitiveNullable", "1")
    boolean: Optional[bool] = None
    date: Optional[datetime.date] = None
    float_32: Optional[float] = Field(None, alias="float32")
    float_64: Optional[float] = Field(None, alias="float64")
    int_32: Optional[int] = Field(None, alias="int32")
    int_64: Optional[int] = Field(None, alias="int64")
    json_: Optional[dict] = Field(None, alias="json")
    text: Optional[str] = None
    timestamp: Optional[datetime.datetime] = None

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> PrimitiveNullable:
        """Convert this GraphQL format of primitive nullable to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return PrimitiveNullable(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            boolean=self.boolean,
            date=self.date,
            float_32=self.float_32,
            float_64=self.float_64,
            int_32=self.int_32,
            int_64=self.int_64,
            json_=self.json_,
            text=self.text,
            timestamp=self.timestamp,
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> PrimitiveNullableWrite:
        """Convert this GraphQL format of primitive nullable to the writing format."""
        return PrimitiveNullableWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            boolean=self.boolean,
            date=self.date,
            float_32=self.float_32,
            float_64=self.float_64,
            int_32=self.int_32,
            int_64=self.int_64,
            json_=self.json_,
            text=self.text,
            timestamp=self.timestamp,
        )

as_read()

Convert this GraphQL format of primitive nullable to the reading format.

Source code in examples/omni/data_classes/_primitive_nullable.py

@no_type_check
def as_read(self) -> PrimitiveNullable:
    """Convert this GraphQL format of primitive nullable to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return PrimitiveNullable(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        boolean=self.boolean,
        date=self.date,
        float_32=self.float_32,
        float_64=self.float_64,
        int_32=self.int_32,
        int_64=self.int_64,
        json_=self.json_,
        text=self.text,
        timestamp=self.timestamp,
    )

as_write()

Convert this GraphQL format of primitive nullable to the writing format.

Source code in examples/omni/data_classes/_primitive_nullable.py

@no_type_check
def as_write(self) -> PrimitiveNullableWrite:
    """Convert this GraphQL format of primitive nullable to the writing format."""
    return PrimitiveNullableWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        boolean=self.boolean,
        date=self.date,
        float_32=self.float_32,
        float_64=self.float_64,
        int_32=self.int_32,
        int_64=self.int_64,
        json_=self.json_,
        text=self.text,
        timestamp=self.timestamp,
    )

PrimitiveNullableList

Bases: DomainModelList[PrimitiveNullable]

List of primitive nullables in the read version.

Source code in examples/omni/data_classes/_primitive_nullable.py

class PrimitiveNullableList(DomainModelList[PrimitiveNullable]):
    """List of primitive nullables in the read version."""

    _INSTANCE = PrimitiveNullable

    def as_write(self) -> PrimitiveNullableWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        return PrimitiveNullableWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> PrimitiveNullableWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/omni/data_classes/_primitive_nullable.py

def as_apply(self) -> PrimitiveNullableWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/omni/data_classes/_primitive_nullable.py

def as_write(self) -> PrimitiveNullableWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    return PrimitiveNullableWriteList([node.as_write() for node in self.data])

PrimitiveNullableListed

Bases: DomainModel

This represents the reading version of primitive nullable listed.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the primitive nullable listed.	required
data_record		The data record of the primitive nullable listed node.	required
boolean		The boolean field.	required
date		The date field.	required
float_32		The float 32 field.	required
float_64		The float 64 field.	required
int_32		The int 32 field.	required
int_64		The int 64 field.	required
json_		The json field.	required
text		The text field.	required
timestamp		The timestamp field.	required

Source code in examples/omni/data_classes/_primitive_nullable_listed.py

class PrimitiveNullableListed(DomainModel):
    """This represents the reading version of primitive nullable listed.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the primitive nullable listed.
        data_record: The data record of the primitive nullable listed node.
        boolean: The boolean field.
        date: The date field.
        float_32: The float 32 field.
        float_64: The float 64 field.
        int_32: The int 32 field.
        int_64: The int 64 field.
        json_: The json field.
        text: The text field.
        timestamp: The timestamp field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "PrimitiveNullableListed", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, None] = None
    boolean: Optional[list[bool]] = None
    date: Optional[list[datetime.date]] = None
    float_32: Optional[list[float]] = Field(None, alias="float32")
    float_64: Optional[list[float]] = Field(None, alias="float64")
    int_32: Optional[list[int]] = Field(None, alias="int32")
    int_64: Optional[list[int]] = Field(None, alias="int64")
    json_: Optional[list[dict]] = Field(None, alias="json")
    text: Optional[list[str]] = None
    timestamp: Optional[list[datetime.datetime]] = None

    def as_write(self) -> PrimitiveNullableListedWrite:
        """Convert this read version of primitive nullable listed to the writing version."""
        return PrimitiveNullableListedWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            boolean=self.boolean,
            date=self.date,
            float_32=self.float_32,
            float_64=self.float_64,
            int_32=self.int_32,
            int_64=self.int_64,
            json_=self.json_,
            text=self.text,
            timestamp=self.timestamp,
        )

    def as_apply(self) -> PrimitiveNullableListedWrite:
        """Convert this read version of primitive nullable listed to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert this read version of primitive nullable listed to the writing version.

Source code in examples/omni/data_classes/_primitive_nullable_listed.py

def as_apply(self) -> PrimitiveNullableListedWrite:
    """Convert this read version of primitive nullable listed to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of primitive nullable listed to the writing version.

Source code in examples/omni/data_classes/_primitive_nullable_listed.py

def as_write(self) -> PrimitiveNullableListedWrite:
    """Convert this read version of primitive nullable listed to the writing version."""
    return PrimitiveNullableListedWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        boolean=self.boolean,
        date=self.date,
        float_32=self.float_32,
        float_64=self.float_64,
        int_32=self.int_32,
        int_64=self.int_64,
        json_=self.json_,
        text=self.text,
        timestamp=self.timestamp,
    )

PrimitiveNullableListedGraphQL

Bases: GraphQLCore

This represents the reading version of primitive nullable listed, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the primitive nullable listed.	required
data_record		The data record of the primitive nullable listed node.	required
boolean		The boolean field.	required
date		The date field.	required
float_32		The float 32 field.	required
float_64		The float 64 field.	required
int_32		The int 32 field.	required
int_64		The int 64 field.	required
json_		The json field.	required
text		The text field.	required
timestamp		The timestamp field.	required

Source code in examples/omni/data_classes/_primitive_nullable_listed.py

class PrimitiveNullableListedGraphQL(GraphQLCore):
    """This represents the reading version of primitive nullable listed, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the primitive nullable listed.
        data_record: The data record of the primitive nullable listed node.
        boolean: The boolean field.
        date: The date field.
        float_32: The float 32 field.
        float_64: The float 64 field.
        int_32: The int 32 field.
        int_64: The int 64 field.
        json_: The json field.
        text: The text field.
        timestamp: The timestamp field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "PrimitiveNullableListed", "1")
    boolean: Optional[list[bool]] = None
    date: Optional[list[datetime.date]] = None
    float_32: Optional[list[float]] = Field(None, alias="float32")
    float_64: Optional[list[float]] = Field(None, alias="float64")
    int_32: Optional[list[int]] = Field(None, alias="int32")
    int_64: Optional[list[int]] = Field(None, alias="int64")
    json_: Optional[list[dict]] = Field(None, alias="json")
    text: Optional[list[str]] = None
    timestamp: Optional[list[datetime.datetime]] = None

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> PrimitiveNullableListed:
        """Convert this GraphQL format of primitive nullable listed to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return PrimitiveNullableListed(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            boolean=self.boolean,
            date=self.date,
            float_32=self.float_32,
            float_64=self.float_64,
            int_32=self.int_32,
            int_64=self.int_64,
            json_=self.json_,
            text=self.text,
            timestamp=self.timestamp,
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> PrimitiveNullableListedWrite:
        """Convert this GraphQL format of primitive nullable listed to the writing format."""
        return PrimitiveNullableListedWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            boolean=self.boolean,
            date=self.date,
            float_32=self.float_32,
            float_64=self.float_64,
            int_32=self.int_32,
            int_64=self.int_64,
            json_=self.json_,
            text=self.text,
            timestamp=self.timestamp,
        )

as_read()

Convert this GraphQL format of primitive nullable listed to the reading format.

Source code in examples/omni/data_classes/_primitive_nullable_listed.py

@no_type_check
def as_read(self) -> PrimitiveNullableListed:
    """Convert this GraphQL format of primitive nullable listed to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return PrimitiveNullableListed(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        boolean=self.boolean,
        date=self.date,
        float_32=self.float_32,
        float_64=self.float_64,
        int_32=self.int_32,
        int_64=self.int_64,
        json_=self.json_,
        text=self.text,
        timestamp=self.timestamp,
    )

as_write()

Convert this GraphQL format of primitive nullable listed to the writing format.

Source code in examples/omni/data_classes/_primitive_nullable_listed.py

@no_type_check
def as_write(self) -> PrimitiveNullableListedWrite:
    """Convert this GraphQL format of primitive nullable listed to the writing format."""
    return PrimitiveNullableListedWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        boolean=self.boolean,
        date=self.date,
        float_32=self.float_32,
        float_64=self.float_64,
        int_32=self.int_32,
        int_64=self.int_64,
        json_=self.json_,
        text=self.text,
        timestamp=self.timestamp,
    )

PrimitiveNullableListedList

Bases: DomainModelList[PrimitiveNullableListed]

List of primitive nullable listeds in the read version.

Source code in examples/omni/data_classes/_primitive_nullable_listed.py

class PrimitiveNullableListedList(DomainModelList[PrimitiveNullableListed]):
    """List of primitive nullable listeds in the read version."""

    _INSTANCE = PrimitiveNullableListed

    def as_write(self) -> PrimitiveNullableListedWriteList:
        """Convert these read versions of primitive nullable listed to the writing versions."""
        return PrimitiveNullableListedWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> PrimitiveNullableListedWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/omni/data_classes/_primitive_nullable_listed.py

def as_apply(self) -> PrimitiveNullableListedWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of primitive nullable listed to the writing versions.

Source code in examples/omni/data_classes/_primitive_nullable_listed.py

def as_write(self) -> PrimitiveNullableListedWriteList:
    """Convert these read versions of primitive nullable listed to the writing versions."""
    return PrimitiveNullableListedWriteList([node.as_write() for node in self.data])

PrimitiveNullableListedWrite

Bases: DomainModelWrite

This represents the writing version of primitive nullable listed.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the primitive nullable listed.	required
data_record		The data record of the primitive nullable listed node.	required
boolean		The boolean field.	required
date		The date field.	required
float_32		The float 32 field.	required
float_64		The float 64 field.	required
int_32		The int 32 field.	required
int_64		The int 64 field.	required
json_		The json field.	required
text		The text field.	required
timestamp		The timestamp field.	required

Source code in examples/omni/data_classes/_primitive_nullable_listed.py

class PrimitiveNullableListedWrite(DomainModelWrite):
    """This represents the writing version of primitive nullable listed.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the primitive nullable listed.
        data_record: The data record of the primitive nullable listed node.
        boolean: The boolean field.
        date: The date field.
        float_32: The float 32 field.
        float_64: The float 64 field.
        int_32: The int 32 field.
        int_64: The int 64 field.
        json_: The json field.
        text: The text field.
        timestamp: The timestamp field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "PrimitiveNullableListed", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = None
    boolean: Optional[list[bool]] = None
    date: Optional[list[datetime.date]] = None
    float_32: Optional[list[float]] = Field(None, alias="float32")
    float_64: Optional[list[float]] = Field(None, alias="float64")
    int_32: Optional[list[int]] = Field(None, alias="int32")
    int_64: Optional[list[int]] = Field(None, alias="int64")
    json_: Optional[list[dict]] = Field(None, alias="json")
    text: Optional[list[str]] = None
    timestamp: Optional[list[datetime.datetime]] = None

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.boolean is not None or write_none:
            properties["boolean"] = self.boolean

        if self.date is not None or write_none:
            properties["date"] = [date.isoformat() for date in self.date or []]

        if self.float_32 is not None or write_none:
            properties["float32"] = self.float_32

        if self.float_64 is not None or write_none:
            properties["float64"] = self.float_64

        if self.int_32 is not None or write_none:
            properties["int32"] = self.int_32

        if self.int_64 is not None or write_none:
            properties["int64"] = self.int_64

        if self.json_ is not None or write_none:
            properties["json"] = self.json_

        if self.text is not None or write_none:
            properties["text"] = self.text

        if self.timestamp is not None or write_none:
            properties["timestamp"] = [
                timestamp.isoformat(timespec="milliseconds") for timestamp in self.timestamp or []
            ]

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        return resources

PrimitiveNullableListedWriteList

Bases: DomainModelWriteList[PrimitiveNullableListedWrite]

List of primitive nullable listeds in the writing version.

Source code in examples/omni/data_classes/_primitive_nullable_listed.py

class PrimitiveNullableListedWriteList(DomainModelWriteList[PrimitiveNullableListedWrite]):
    """List of primitive nullable listeds in the writing version."""

    _INSTANCE = PrimitiveNullableListedWrite

PrimitiveNullableWrite

Bases: DomainModelWrite

This represents the writing version of primitive nullable.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the primitive nullable.	required
data_record		The data record of the primitive nullable node.	required
boolean		The boolean field.	required
date		The date field.	required
float_32		The float 32 field.	required
float_64		The float 64 field.	required
int_32		The int 32 field.	required
int_64		The int 64 field.	required
json_		The json field.	required
text		The text field.	required
timestamp		The timestamp field.	required

Source code in examples/omni/data_classes/_primitive_nullable.py

class PrimitiveNullableWrite(DomainModelWrite):
    """This represents the writing version of primitive nullable.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the primitive nullable.
        data_record: The data record of the primitive nullable node.
        boolean: The boolean field.
        date: The date field.
        float_32: The float 32 field.
        float_64: The float 64 field.
        int_32: The int 32 field.
        int_64: The int 64 field.
        json_: The json field.
        text: The text field.
        timestamp: The timestamp field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "PrimitiveNullable", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = None
    boolean: Optional[bool] = None
    date: Optional[datetime.date] = None
    float_32: Optional[float] = Field(None, alias="float32")
    float_64: Optional[float] = Field(None, alias="float64")
    int_32: Optional[int] = Field(None, alias="int32")
    int_64: Optional[int] = Field(None, alias="int64")
    json_: Optional[dict] = Field(None, alias="json")
    text: Optional[str] = None
    timestamp: Optional[datetime.datetime] = None

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.boolean is not None or write_none:
            properties["boolean"] = self.boolean

        if self.date is not None or write_none:
            properties["date"] = self.date.isoformat() if self.date else None

        if self.float_32 is not None or write_none:
            properties["float32"] = self.float_32

        if self.float_64 is not None or write_none:
            properties["float64"] = self.float_64

        if self.int_32 is not None or write_none:
            properties["int32"] = self.int_32

        if self.int_64 is not None or write_none:
            properties["int64"] = self.int_64

        if self.json_ is not None or write_none:
            properties["json"] = self.json_

        if self.text is not None or write_none:
            properties["text"] = self.text

        if self.timestamp is not None or write_none:
            properties["timestamp"] = self.timestamp.isoformat(timespec="milliseconds") if self.timestamp else None

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        return resources

PrimitiveNullableWriteList

Bases: DomainModelWriteList[PrimitiveNullableWrite]

List of primitive nullables in the writing version.

Source code in examples/omni/data_classes/_primitive_nullable.py

class PrimitiveNullableWriteList(DomainModelWriteList[PrimitiveNullableWrite]):
    """List of primitive nullables in the writing version."""

    _INSTANCE = PrimitiveNullableWrite

PrimitiveRequired

Bases: DomainModel

This represents the reading version of primitive required.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the primitive required.	required
data_record		The data record of the primitive required node.	required
boolean		The boolean field.	required
date		The date field.	required
float_32		The float 32 field.	required
float_64		The float 64 field.	required
int_32		The int 32 field.	required
int_64		The int 64 field.	required
json_		The json field.	required
text		The text field.	required
timestamp		The timestamp field.	required

Source code in examples/omni/data_classes/_primitive_required.py

class PrimitiveRequired(DomainModel):
    """This represents the reading version of primitive required.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the primitive required.
        data_record: The data record of the primitive required node.
        boolean: The boolean field.
        date: The date field.
        float_32: The float 32 field.
        float_64: The float 64 field.
        int_32: The int 32 field.
        int_64: The int 64 field.
        json_: The json field.
        text: The text field.
        timestamp: The timestamp field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "PrimitiveRequired", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, None] = None
    boolean: bool
    date: datetime.date
    float_32: float = Field(alias="float32")
    float_64: float = Field(alias="float64")
    int_32: int = Field(alias="int32")
    int_64: int = Field(alias="int64")
    json_: dict = Field(alias="json")
    text: str
    timestamp: datetime.datetime

    def as_write(self) -> PrimitiveRequiredWrite:
        """Convert this read version of primitive required to the writing version."""
        return PrimitiveRequiredWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            boolean=self.boolean,
            date=self.date,
            float_32=self.float_32,
            float_64=self.float_64,
            int_32=self.int_32,
            int_64=self.int_64,
            json_=self.json_,
            text=self.text,
            timestamp=self.timestamp,
        )

    def as_apply(self) -> PrimitiveRequiredWrite:
        """Convert this read version of primitive required to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert this read version of primitive required to the writing version.

Source code in examples/omni/data_classes/_primitive_required.py

def as_apply(self) -> PrimitiveRequiredWrite:
    """Convert this read version of primitive required to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of primitive required to the writing version.

Source code in examples/omni/data_classes/_primitive_required.py

def as_write(self) -> PrimitiveRequiredWrite:
    """Convert this read version of primitive required to the writing version."""
    return PrimitiveRequiredWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        boolean=self.boolean,
        date=self.date,
        float_32=self.float_32,
        float_64=self.float_64,
        int_32=self.int_32,
        int_64=self.int_64,
        json_=self.json_,
        text=self.text,
        timestamp=self.timestamp,
    )

PrimitiveRequiredGraphQL

Bases: GraphQLCore

This represents the reading version of primitive required, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the primitive required.	required
data_record		The data record of the primitive required node.	required
boolean		The boolean field.	required
date		The date field.	required
float_32		The float 32 field.	required
float_64		The float 64 field.	required
int_32		The int 32 field.	required
int_64		The int 64 field.	required
json_		The json field.	required
text		The text field.	required
timestamp		The timestamp field.	required

Source code in examples/omni/data_classes/_primitive_required.py

class PrimitiveRequiredGraphQL(GraphQLCore):
    """This represents the reading version of primitive required, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the primitive required.
        data_record: The data record of the primitive required node.
        boolean: The boolean field.
        date: The date field.
        float_32: The float 32 field.
        float_64: The float 64 field.
        int_32: The int 32 field.
        int_64: The int 64 field.
        json_: The json field.
        text: The text field.
        timestamp: The timestamp field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "PrimitiveRequired", "1")
    boolean: Optional[bool] = None
    date: Optional[datetime.date] = None
    float_32: Optional[float] = Field(None, alias="float32")
    float_64: Optional[float] = Field(None, alias="float64")
    int_32: Optional[int] = Field(None, alias="int32")
    int_64: Optional[int] = Field(None, alias="int64")
    json_: Optional[dict] = Field(None, alias="json")
    text: Optional[str] = None
    timestamp: Optional[datetime.datetime] = None

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> PrimitiveRequired:
        """Convert this GraphQL format of primitive required to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return PrimitiveRequired(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            boolean=self.boolean,
            date=self.date,
            float_32=self.float_32,
            float_64=self.float_64,
            int_32=self.int_32,
            int_64=self.int_64,
            json_=self.json_,
            text=self.text,
            timestamp=self.timestamp,
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> PrimitiveRequiredWrite:
        """Convert this GraphQL format of primitive required to the writing format."""
        return PrimitiveRequiredWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            boolean=self.boolean,
            date=self.date,
            float_32=self.float_32,
            float_64=self.float_64,
            int_32=self.int_32,
            int_64=self.int_64,
            json_=self.json_,
            text=self.text,
            timestamp=self.timestamp,
        )

as_read()

Convert this GraphQL format of primitive required to the reading format.

Source code in examples/omni/data_classes/_primitive_required.py

@no_type_check
def as_read(self) -> PrimitiveRequired:
    """Convert this GraphQL format of primitive required to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return PrimitiveRequired(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        boolean=self.boolean,
        date=self.date,
        float_32=self.float_32,
        float_64=self.float_64,
        int_32=self.int_32,
        int_64=self.int_64,
        json_=self.json_,
        text=self.text,
        timestamp=self.timestamp,
    )

as_write()

Convert this GraphQL format of primitive required to the writing format.

Source code in examples/omni/data_classes/_primitive_required.py

@no_type_check
def as_write(self) -> PrimitiveRequiredWrite:
    """Convert this GraphQL format of primitive required to the writing format."""
    return PrimitiveRequiredWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        boolean=self.boolean,
        date=self.date,
        float_32=self.float_32,
        float_64=self.float_64,
        int_32=self.int_32,
        int_64=self.int_64,
        json_=self.json_,
        text=self.text,
        timestamp=self.timestamp,
    )

PrimitiveRequiredList

Bases: DomainModelList[PrimitiveRequired]

List of primitive requireds in the read version.

Source code in examples/omni/data_classes/_primitive_required.py

class PrimitiveRequiredList(DomainModelList[PrimitiveRequired]):
    """List of primitive requireds in the read version."""

    _INSTANCE = PrimitiveRequired

    def as_write(self) -> PrimitiveRequiredWriteList:
        """Convert these read versions of primitive required to the writing versions."""
        return PrimitiveRequiredWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> PrimitiveRequiredWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/omni/data_classes/_primitive_required.py

def as_apply(self) -> PrimitiveRequiredWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of primitive required to the writing versions.

Source code in examples/omni/data_classes/_primitive_required.py

def as_write(self) -> PrimitiveRequiredWriteList:
    """Convert these read versions of primitive required to the writing versions."""
    return PrimitiveRequiredWriteList([node.as_write() for node in self.data])

PrimitiveRequiredListed

Bases: DomainModel

This represents the reading version of primitive required listed.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the primitive required listed.	required
data_record		The data record of the primitive required listed node.	required
boolean		The boolean field.	required
date		The date field.	required
float_32		The float 32 field.	required
float_64		The float 64 field.	required
int_32		The int 32 field.	required
int_64		The int 64 field.	required
json_		The json field.	required
text		The text field.	required
timestamp		The timestamp field.	required

Source code in examples/omni/data_classes/_primitive_required_listed.py

class PrimitiveRequiredListed(DomainModel):
    """This represents the reading version of primitive required listed.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the primitive required listed.
        data_record: The data record of the primitive required listed node.
        boolean: The boolean field.
        date: The date field.
        float_32: The float 32 field.
        float_64: The float 64 field.
        int_32: The int 32 field.
        int_64: The int 64 field.
        json_: The json field.
        text: The text field.
        timestamp: The timestamp field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "PrimitiveRequiredListed", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, None] = None
    boolean: list[bool]
    date: list[datetime.date]
    float_32: list[float] = Field(alias="float32")
    float_64: list[float] = Field(alias="float64")
    int_32: list[int] = Field(alias="int32")
    int_64: list[int] = Field(alias="int64")
    json_: list[dict] = Field(alias="json")
    text: list[str]
    timestamp: list[datetime.datetime]

    def as_write(self) -> PrimitiveRequiredListedWrite:
        """Convert this read version of primitive required listed to the writing version."""
        return PrimitiveRequiredListedWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            boolean=self.boolean,
            date=self.date,
            float_32=self.float_32,
            float_64=self.float_64,
            int_32=self.int_32,
            int_64=self.int_64,
            json_=self.json_,
            text=self.text,
            timestamp=self.timestamp,
        )

    def as_apply(self) -> PrimitiveRequiredListedWrite:
        """Convert this read version of primitive required listed to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert this read version of primitive required listed to the writing version.

Source code in examples/omni/data_classes/_primitive_required_listed.py

def as_apply(self) -> PrimitiveRequiredListedWrite:
    """Convert this read version of primitive required listed to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of primitive required listed to the writing version.

Source code in examples/omni/data_classes/_primitive_required_listed.py

def as_write(self) -> PrimitiveRequiredListedWrite:
    """Convert this read version of primitive required listed to the writing version."""
    return PrimitiveRequiredListedWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        boolean=self.boolean,
        date=self.date,
        float_32=self.float_32,
        float_64=self.float_64,
        int_32=self.int_32,
        int_64=self.int_64,
        json_=self.json_,
        text=self.text,
        timestamp=self.timestamp,
    )

PrimitiveRequiredListedGraphQL

Bases: GraphQLCore

This represents the reading version of primitive required listed, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the primitive required listed.	required
data_record		The data record of the primitive required listed node.	required
boolean		The boolean field.	required
date		The date field.	required
float_32		The float 32 field.	required
float_64		The float 64 field.	required
int_32		The int 32 field.	required
int_64		The int 64 field.	required
json_		The json field.	required
text		The text field.	required
timestamp		The timestamp field.	required

Source code in examples/omni/data_classes/_primitive_required_listed.py

class PrimitiveRequiredListedGraphQL(GraphQLCore):
    """This represents the reading version of primitive required listed, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the primitive required listed.
        data_record: The data record of the primitive required listed node.
        boolean: The boolean field.
        date: The date field.
        float_32: The float 32 field.
        float_64: The float 64 field.
        int_32: The int 32 field.
        int_64: The int 64 field.
        json_: The json field.
        text: The text field.
        timestamp: The timestamp field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "PrimitiveRequiredListed", "1")
    boolean: Optional[list[bool]] = None
    date: Optional[list[datetime.date]] = None
    float_32: Optional[list[float]] = Field(None, alias="float32")
    float_64: Optional[list[float]] = Field(None, alias="float64")
    int_32: Optional[list[int]] = Field(None, alias="int32")
    int_64: Optional[list[int]] = Field(None, alias="int64")
    json_: Optional[list[dict]] = Field(None, alias="json")
    text: Optional[list[str]] = None
    timestamp: Optional[list[datetime.datetime]] = None

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> PrimitiveRequiredListed:
        """Convert this GraphQL format of primitive required listed to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return PrimitiveRequiredListed(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            boolean=self.boolean,
            date=self.date,
            float_32=self.float_32,
            float_64=self.float_64,
            int_32=self.int_32,
            int_64=self.int_64,
            json_=self.json_,
            text=self.text,
            timestamp=self.timestamp,
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> PrimitiveRequiredListedWrite:
        """Convert this GraphQL format of primitive required listed to the writing format."""
        return PrimitiveRequiredListedWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            boolean=self.boolean,
            date=self.date,
            float_32=self.float_32,
            float_64=self.float_64,
            int_32=self.int_32,
            int_64=self.int_64,
            json_=self.json_,
            text=self.text,
            timestamp=self.timestamp,
        )

as_read()

Convert this GraphQL format of primitive required listed to the reading format.

Source code in examples/omni/data_classes/_primitive_required_listed.py

@no_type_check
def as_read(self) -> PrimitiveRequiredListed:
    """Convert this GraphQL format of primitive required listed to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return PrimitiveRequiredListed(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        boolean=self.boolean,
        date=self.date,
        float_32=self.float_32,
        float_64=self.float_64,
        int_32=self.int_32,
        int_64=self.int_64,
        json_=self.json_,
        text=self.text,
        timestamp=self.timestamp,
    )

as_write()

Convert this GraphQL format of primitive required listed to the writing format.

Source code in examples/omni/data_classes/_primitive_required_listed.py

@no_type_check
def as_write(self) -> PrimitiveRequiredListedWrite:
    """Convert this GraphQL format of primitive required listed to the writing format."""
    return PrimitiveRequiredListedWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        boolean=self.boolean,
        date=self.date,
        float_32=self.float_32,
        float_64=self.float_64,
        int_32=self.int_32,
        int_64=self.int_64,
        json_=self.json_,
        text=self.text,
        timestamp=self.timestamp,
    )

PrimitiveRequiredListedList

Bases: DomainModelList[PrimitiveRequiredListed]

List of primitive required listeds in the read version.

Source code in examples/omni/data_classes/_primitive_required_listed.py

class PrimitiveRequiredListedList(DomainModelList[PrimitiveRequiredListed]):
    """List of primitive required listeds in the read version."""

    _INSTANCE = PrimitiveRequiredListed

    def as_write(self) -> PrimitiveRequiredListedWriteList:
        """Convert these read versions of primitive required listed to the writing versions."""
        return PrimitiveRequiredListedWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> PrimitiveRequiredListedWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/omni/data_classes/_primitive_required_listed.py

def as_apply(self) -> PrimitiveRequiredListedWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of primitive required listed to the writing versions.

Source code in examples/omni/data_classes/_primitive_required_listed.py

def as_write(self) -> PrimitiveRequiredListedWriteList:
    """Convert these read versions of primitive required listed to the writing versions."""
    return PrimitiveRequiredListedWriteList([node.as_write() for node in self.data])

PrimitiveRequiredListedWrite

Bases: DomainModelWrite

This represents the writing version of primitive required listed.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the primitive required listed.	required
data_record		The data record of the primitive required listed node.	required
boolean		The boolean field.	required
date		The date field.	required
float_32		The float 32 field.	required
float_64		The float 64 field.	required
int_32		The int 32 field.	required
int_64		The int 64 field.	required
json_		The json field.	required
text		The text field.	required
timestamp		The timestamp field.	required

Source code in examples/omni/data_classes/_primitive_required_listed.py

class PrimitiveRequiredListedWrite(DomainModelWrite):
    """This represents the writing version of primitive required listed.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the primitive required listed.
        data_record: The data record of the primitive required listed node.
        boolean: The boolean field.
        date: The date field.
        float_32: The float 32 field.
        float_64: The float 64 field.
        int_32: The int 32 field.
        int_64: The int 64 field.
        json_: The json field.
        text: The text field.
        timestamp: The timestamp field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "PrimitiveRequiredListed", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = None
    boolean: list[bool]
    date: list[datetime.date]
    float_32: list[float] = Field(alias="float32")
    float_64: list[float] = Field(alias="float64")
    int_32: list[int] = Field(alias="int32")
    int_64: list[int] = Field(alias="int64")
    json_: list[dict] = Field(alias="json")
    text: list[str]
    timestamp: list[datetime.datetime]

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.boolean is not None:
            properties["boolean"] = self.boolean

        if self.date is not None:
            properties["date"] = [date.isoformat() for date in self.date or []]

        if self.float_32 is not None:
            properties["float32"] = self.float_32

        if self.float_64 is not None:
            properties["float64"] = self.float_64

        if self.int_32 is not None:
            properties["int32"] = self.int_32

        if self.int_64 is not None:
            properties["int64"] = self.int_64

        if self.json_ is not None:
            properties["json"] = self.json_

        if self.text is not None:
            properties["text"] = self.text

        if self.timestamp is not None:
            properties["timestamp"] = [
                timestamp.isoformat(timespec="milliseconds") for timestamp in self.timestamp or []
            ]

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        return resources

PrimitiveRequiredListedWriteList

Bases: DomainModelWriteList[PrimitiveRequiredListedWrite]

List of primitive required listeds in the writing version.

Source code in examples/omni/data_classes/_primitive_required_listed.py

class PrimitiveRequiredListedWriteList(DomainModelWriteList[PrimitiveRequiredListedWrite]):
    """List of primitive required listeds in the writing version."""

    _INSTANCE = PrimitiveRequiredListedWrite

PrimitiveRequiredWrite

Bases: DomainModelWrite

This represents the writing version of primitive required.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the primitive required.	required
data_record		The data record of the primitive required node.	required
boolean		The boolean field.	required
date		The date field.	required
float_32		The float 32 field.	required
float_64		The float 64 field.	required
int_32		The int 32 field.	required
int_64		The int 64 field.	required
json_		The json field.	required
text		The text field.	required
timestamp		The timestamp field.	required

Source code in examples/omni/data_classes/_primitive_required.py

class PrimitiveRequiredWrite(DomainModelWrite):
    """This represents the writing version of primitive required.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the primitive required.
        data_record: The data record of the primitive required node.
        boolean: The boolean field.
        date: The date field.
        float_32: The float 32 field.
        float_64: The float 64 field.
        int_32: The int 32 field.
        int_64: The int 64 field.
        json_: The json field.
        text: The text field.
        timestamp: The timestamp field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "PrimitiveRequired", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = None
    boolean: bool
    date: datetime.date
    float_32: float = Field(alias="float32")
    float_64: float = Field(alias="float64")
    int_32: int = Field(alias="int32")
    int_64: int = Field(alias="int64")
    json_: dict = Field(alias="json")
    text: str
    timestamp: datetime.datetime

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.boolean is not None:
            properties["boolean"] = self.boolean

        if self.date is not None:
            properties["date"] = self.date.isoformat() if self.date else None

        if self.float_32 is not None:
            properties["float32"] = self.float_32

        if self.float_64 is not None:
            properties["float64"] = self.float_64

        if self.int_32 is not None:
            properties["int32"] = self.int_32

        if self.int_64 is not None:
            properties["int64"] = self.int_64

        if self.json_ is not None:
            properties["json"] = self.json_

        if self.text is not None:
            properties["text"] = self.text

        if self.timestamp is not None:
            properties["timestamp"] = self.timestamp.isoformat(timespec="milliseconds") if self.timestamp else None

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        return resources

PrimitiveRequiredWriteList

Bases: DomainModelWriteList[PrimitiveRequiredWrite]

List of primitive requireds in the writing version.

Source code in examples/omni/data_classes/_primitive_required.py

class PrimitiveRequiredWriteList(DomainModelWriteList[PrimitiveRequiredWrite]):
    """List of primitive requireds in the writing version."""

    _INSTANCE = PrimitiveRequiredWrite

PrimitiveWithDefaults

Bases: DomainModel

This represents the reading version of primitive with default.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the primitive with default.	required
data_record		The data record of the primitive with default node.	required
auto_increment_int_32		The auto increment int 32 field.	required
default_boolean		The default boolean field.	required
default_float_32		The default float 32 field.	required
default_object		The default object field.	required
default_string		The default string field.	required

Source code in examples/omni/data_classes/_primitive_with_defaults.py

class PrimitiveWithDefaults(DomainModel):
    """This represents the reading version of primitive with default.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the primitive with default.
        data_record: The data record of the primitive with default node.
        auto_increment_int_32: The auto increment int 32 field.
        default_boolean: The default boolean field.
        default_float_32: The default float 32 field.
        default_object: The default object field.
        default_string: The default string field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "PrimitiveWithDefaults", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, None] = None
    auto_increment_int_32: int = Field(alias="autoIncrementInt32")
    default_boolean: Optional[bool] = Field(None, alias="defaultBoolean")
    default_float_32: Optional[float] = Field(None, alias="defaultFloat32")
    default_object: Optional[dict] = Field(None, alias="defaultObject")
    default_string: Optional[str] = Field(None, alias="defaultString")

    def as_write(self) -> PrimitiveWithDefaultsWrite:
        """Convert this read version of primitive with default to the writing version."""
        return PrimitiveWithDefaultsWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            auto_increment_int_32=self.auto_increment_int_32,
            default_boolean=self.default_boolean,
            default_float_32=self.default_float_32,
            default_object=self.default_object,
            default_string=self.default_string,
        )

    def as_apply(self) -> PrimitiveWithDefaultsWrite:
        """Convert this read version of primitive with default to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert this read version of primitive with default to the writing version.

Source code in examples/omni/data_classes/_primitive_with_defaults.py

def as_apply(self) -> PrimitiveWithDefaultsWrite:
    """Convert this read version of primitive with default to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of primitive with default to the writing version.

Source code in examples/omni/data_classes/_primitive_with_defaults.py

def as_write(self) -> PrimitiveWithDefaultsWrite:
    """Convert this read version of primitive with default to the writing version."""
    return PrimitiveWithDefaultsWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        auto_increment_int_32=self.auto_increment_int_32,
        default_boolean=self.default_boolean,
        default_float_32=self.default_float_32,
        default_object=self.default_object,
        default_string=self.default_string,
    )

PrimitiveWithDefaultsGraphQL

Bases: GraphQLCore

This represents the reading version of primitive with default, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the primitive with default.	required
data_record		The data record of the primitive with default node.	required
auto_increment_int_32		The auto increment int 32 field.	required
default_boolean		The default boolean field.	required
default_float_32		The default float 32 field.	required
default_object		The default object field.	required
default_string		The default string field.	required

Source code in examples/omni/data_classes/_primitive_with_defaults.py

class PrimitiveWithDefaultsGraphQL(GraphQLCore):
    """This represents the reading version of primitive with default, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the primitive with default.
        data_record: The data record of the primitive with default node.
        auto_increment_int_32: The auto increment int 32 field.
        default_boolean: The default boolean field.
        default_float_32: The default float 32 field.
        default_object: The default object field.
        default_string: The default string field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "PrimitiveWithDefaults", "1")
    auto_increment_int_32: Optional[int] = Field(None, alias="autoIncrementInt32")
    default_boolean: Optional[bool] = Field(None, alias="defaultBoolean")
    default_float_32: Optional[float] = Field(None, alias="defaultFloat32")
    default_object: Optional[dict] = Field(None, alias="defaultObject")
    default_string: Optional[str] = Field(None, alias="defaultString")

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> PrimitiveWithDefaults:
        """Convert this GraphQL format of primitive with default to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return PrimitiveWithDefaults(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            auto_increment_int_32=self.auto_increment_int_32,
            default_boolean=self.default_boolean,
            default_float_32=self.default_float_32,
            default_object=self.default_object,
            default_string=self.default_string,
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> PrimitiveWithDefaultsWrite:
        """Convert this GraphQL format of primitive with default to the writing format."""
        return PrimitiveWithDefaultsWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            auto_increment_int_32=self.auto_increment_int_32,
            default_boolean=self.default_boolean,
            default_float_32=self.default_float_32,
            default_object=self.default_object,
            default_string=self.default_string,
        )

as_read()

Convert this GraphQL format of primitive with default to the reading format.

Source code in examples/omni/data_classes/_primitive_with_defaults.py

@no_type_check
def as_read(self) -> PrimitiveWithDefaults:
    """Convert this GraphQL format of primitive with default to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return PrimitiveWithDefaults(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        auto_increment_int_32=self.auto_increment_int_32,
        default_boolean=self.default_boolean,
        default_float_32=self.default_float_32,
        default_object=self.default_object,
        default_string=self.default_string,
    )

as_write()

Convert this GraphQL format of primitive with default to the writing format.

Source code in examples/omni/data_classes/_primitive_with_defaults.py

@no_type_check
def as_write(self) -> PrimitiveWithDefaultsWrite:
    """Convert this GraphQL format of primitive with default to the writing format."""
    return PrimitiveWithDefaultsWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        auto_increment_int_32=self.auto_increment_int_32,
        default_boolean=self.default_boolean,
        default_float_32=self.default_float_32,
        default_object=self.default_object,
        default_string=self.default_string,
    )

PrimitiveWithDefaultsList

Bases: DomainModelList[PrimitiveWithDefaults]

List of primitive with defaults in the read version.

Source code in examples/omni/data_classes/_primitive_with_defaults.py

class PrimitiveWithDefaultsList(DomainModelList[PrimitiveWithDefaults]):
    """List of primitive with defaults in the read version."""

    _INSTANCE = PrimitiveWithDefaults

    def as_write(self) -> PrimitiveWithDefaultsWriteList:
        """Convert these read versions of primitive with default to the writing versions."""
        return PrimitiveWithDefaultsWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> PrimitiveWithDefaultsWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/omni/data_classes/_primitive_with_defaults.py

def as_apply(self) -> PrimitiveWithDefaultsWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of primitive with default to the writing versions.

Source code in examples/omni/data_classes/_primitive_with_defaults.py

def as_write(self) -> PrimitiveWithDefaultsWriteList:
    """Convert these read versions of primitive with default to the writing versions."""
    return PrimitiveWithDefaultsWriteList([node.as_write() for node in self.data])

PrimitiveWithDefaultsWrite

Bases: DomainModelWrite

This represents the writing version of primitive with default.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the primitive with default.	required
data_record		The data record of the primitive with default node.	required
auto_increment_int_32		The auto increment int 32 field.	required
default_boolean		The default boolean field.	required
default_float_32		The default float 32 field.	required
default_object		The default object field.	required
default_string		The default string field.	required

Source code in examples/omni/data_classes/_primitive_with_defaults.py

class PrimitiveWithDefaultsWrite(DomainModelWrite):
    """This represents the writing version of primitive with default.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the primitive with default.
        data_record: The data record of the primitive with default node.
        auto_increment_int_32: The auto increment int 32 field.
        default_boolean: The default boolean field.
        default_float_32: The default float 32 field.
        default_object: The default object field.
        default_string: The default string field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "PrimitiveWithDefaults", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = None
    auto_increment_int_32: int = Field(alias="autoIncrementInt32")
    default_boolean: Optional[bool] = Field(True, alias="defaultBoolean")
    default_float_32: Optional[float] = Field(0.42, alias="defaultFloat32")
    default_object: Optional[dict] = Field({"foo": "bar"}, alias="defaultObject")
    default_string: Optional[str] = Field("my default text", alias="defaultString")

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.auto_increment_int_32 is not None:
            properties["autoIncrementInt32"] = self.auto_increment_int_32

        if self.default_boolean is not None or write_none:
            properties["defaultBoolean"] = self.default_boolean

        if self.default_float_32 is not None or write_none:
            properties["defaultFloat32"] = self.default_float_32

        if self.default_object is not None or write_none:
            properties["defaultObject"] = self.default_object

        if self.default_string is not None or write_none:
            properties["defaultString"] = self.default_string

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        return resources

PrimitiveWithDefaultsWriteList

Bases: DomainModelWriteList[PrimitiveWithDefaultsWrite]

List of primitive with defaults in the writing version.

Source code in examples/omni/data_classes/_primitive_with_defaults.py

class PrimitiveWithDefaultsWriteList(DomainModelWriteList[PrimitiveWithDefaultsWrite]):
    """List of primitive with defaults in the writing version."""

    _INSTANCE = PrimitiveWithDefaultsWrite

SubInterface

Bases: MainInterface

This represents the reading version of sub interface.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the sub interface.	required
data_record		The data record of the sub interface node.	required
main_value		The main value field.	required
sub_value		The sub value field.	required

Source code in examples/omni/data_classes/_sub_interface.py

class SubInterface(MainInterface):
    """This represents the reading version of sub interface.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the sub interface.
        data_record: The data record of the sub interface node.
        main_value: The main value field.
        sub_value: The sub value field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "SubInterface", "1")

    node_type: Union[dm.DirectRelationReference, None] = None
    sub_value: Optional[str] = Field(None, alias="subValue")

    def as_write(self) -> SubInterfaceWrite:
        """Convert this read version of sub interface to the writing version."""
        return SubInterfaceWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            main_value=self.main_value,
            sub_value=self.sub_value,
        )

    def as_apply(self) -> SubInterfaceWrite:
        """Convert this read version of sub interface to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert this read version of sub interface to the writing version.

Source code in examples/omni/data_classes/_sub_interface.py

def as_apply(self) -> SubInterfaceWrite:
    """Convert this read version of sub interface to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of sub interface to the writing version.

Source code in examples/omni/data_classes/_sub_interface.py

def as_write(self) -> SubInterfaceWrite:
    """Convert this read version of sub interface to the writing version."""
    return SubInterfaceWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        main_value=self.main_value,
        sub_value=self.sub_value,
    )

SubInterfaceGraphQL

Bases: GraphQLCore

This represents the reading version of sub interface, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the sub interface.	required
data_record		The data record of the sub interface node.	required
main_value		The main value field.	required
sub_value		The sub value field.	required

Source code in examples/omni/data_classes/_sub_interface.py

class SubInterfaceGraphQL(GraphQLCore):
    """This represents the reading version of sub interface, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the sub interface.
        data_record: The data record of the sub interface node.
        main_value: The main value field.
        sub_value: The sub value field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "SubInterface", "1")
    main_value: Optional[str] = Field(None, alias="mainValue")
    sub_value: Optional[str] = Field(None, alias="subValue")

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> SubInterface:
        """Convert this GraphQL format of sub interface to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return SubInterface(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            main_value=self.main_value,
            sub_value=self.sub_value,
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> SubInterfaceWrite:
        """Convert this GraphQL format of sub interface to the writing format."""
        return SubInterfaceWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            main_value=self.main_value,
            sub_value=self.sub_value,
        )

as_read()

Convert this GraphQL format of sub interface to the reading format.

Source code in examples/omni/data_classes/_sub_interface.py

@no_type_check
def as_read(self) -> SubInterface:
    """Convert this GraphQL format of sub interface to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return SubInterface(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        main_value=self.main_value,
        sub_value=self.sub_value,
    )

as_write()

Convert this GraphQL format of sub interface to the writing format.

Source code in examples/omni/data_classes/_sub_interface.py

@no_type_check
def as_write(self) -> SubInterfaceWrite:
    """Convert this GraphQL format of sub interface to the writing format."""
    return SubInterfaceWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        main_value=self.main_value,
        sub_value=self.sub_value,
    )

SubInterfaceList

Bases: DomainModelList[SubInterface]

List of sub interfaces in the read version.

Source code in examples/omni/data_classes/_sub_interface.py

class SubInterfaceList(DomainModelList[SubInterface]):
    """List of sub interfaces in the read version."""

    _INSTANCE = SubInterface

    def as_write(self) -> SubInterfaceWriteList:
        """Convert these read versions of sub interface to the writing versions."""
        return SubInterfaceWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> SubInterfaceWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/omni/data_classes/_sub_interface.py

def as_apply(self) -> SubInterfaceWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of sub interface to the writing versions.

Source code in examples/omni/data_classes/_sub_interface.py

def as_write(self) -> SubInterfaceWriteList:
    """Convert these read versions of sub interface to the writing versions."""
    return SubInterfaceWriteList([node.as_write() for node in self.data])

SubInterfaceWrite

Bases: MainInterfaceWrite

This represents the writing version of sub interface.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the sub interface.	required
data_record		The data record of the sub interface node.	required
main_value		The main value field.	required
sub_value		The sub value field.	required

Source code in examples/omni/data_classes/_sub_interface.py

class SubInterfaceWrite(MainInterfaceWrite):
    """This represents the writing version of sub interface.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the sub interface.
        data_record: The data record of the sub interface node.
        main_value: The main value field.
        sub_value: The sub value field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("pygen-models", "SubInterface", "1")

    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = None
    sub_value: Optional[str] = Field(None, alias="subValue")

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.main_value is not None or write_none:
            properties["mainValue"] = self.main_value

        if self.sub_value is not None or write_none:
            properties["subValue"] = self.sub_value

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        return resources

SubInterfaceWriteList

Bases: DomainModelWriteList[SubInterfaceWrite]

List of sub interfaces in the writing version.

Source code in examples/omni/data_classes/_sub_interface.py

class SubInterfaceWriteList(DomainModelWriteList[SubInterfaceWrite]):
    """List of sub interfaces in the writing version."""

    _INSTANCE = SubInterfaceWrite