Base

cognite.pygen._core.models.fields.base

This module contains the base class for all fields.

Field dataclass

A field represents a pydantic field in the generated pydantic class.

Parameters:

Name	Type	Description	Default
name	str	The name of the field. This is used in the generated Python code.	required
doc_name	str	The name of the field in the documentation.	required
prop_name	str	The name of the property in the data model. This is used when reading and writing to CDF.	required
pydantic_field	Literal['Field', 'pydantic.Field']	The name to use for the import 'from pydantic import Field'. This is used in the edge case when the name 'Field' name clashes with the data model class name.	required

Source code in cognite/pygen/_core/models/fields/base.py

@dataclass(frozen=True)
class Field:
    """
    A field represents a pydantic field in the generated pydantic class.

    Args:
        name: The name of the field. This is used in the generated Python code.
        doc_name: The name of the field in the documentation.
        prop_name: The name of the property in the data model. This is used when reading and writing to CDF.
        pydantic_field: The name to use for the import 'from pydantic import Field'. This is used in the edge case
                        when the name 'Field' name clashes with the data model class name.

    """

    name: str
    doc_name: str
    prop_name: str
    description: str | None
    pydantic_field: Literal["Field", "pydantic.Field"]

    @property
    def need_alias(self) -> bool:
        return self.name != self.prop_name

    @classmethod
    def from_property(
        cls,
        prop_id: str,
        prop: ViewProperty,
        node_class_by_view_id: dict[dm.ViewId, NodeDataClass],
        edge_class_by_view_id: dict[dm.ViewId, EdgeDataClass],
        config: pygen_config.PygenConfig,
        view_id: dm.ViewId,
        pydantic_field: Literal["Field", "pydantic.Field"],
        has_default_instance_space: bool,
        direct_relations_by_view_id: dict[dm.ViewId, set[str]],
        view_property_by_container_direct_relation: dict[tuple[dm.ContainerId, str], set[dm.PropertyId]],
        view_by_id: dict[dm.ViewId, dm.View],
    ) -> Field | None:
        from .cdf_reference import CDFExternalField
        from .connections import BaseConnectionField
        from .primitive import BasePrimitiveField

        field_naming = config.naming.field
        name = create_name(prop_id, field_naming.name)
        if is_reserved_word(name, "field", view_id, prop_id):
            name = f"{name}_"

        doc_name = to_words(name, singularize=True)
        variable = create_name(prop_id, field_naming.variable)
        if is_reserved_word(variable, "variable", view_id, prop_id):
            variable = f"{variable}_"

        description: str | None = None
        if hasattr(prop, "description") and isinstance(prop.description, str):
            # This is a workaround for the fact that the description can contain curly quotes
            # which is ruff will complain about. (These comes from che Core model)
            description = prop.description.replace("‘", "'").replace("’", "'")  # noqa: RUF001

        base = cls(
            name=name,
            doc_name=doc_name,
            prop_name=prop_id,
            description=description,
            pydantic_field=pydantic_field,
        )
        if isinstance(prop, dm.ConnectionDefinition) or (
            isinstance(prop, dm.MappedProperty) and isinstance(prop.type, dm.DirectRelation)
        ):
            return BaseConnectionField.load(
                base,
                prop,
                variable,
                node_class_by_view_id,
                edge_class_by_view_id,
                has_default_instance_space,
                view_id,
                direct_relations_by_view_id,
                view_property_by_container_direct_relation,
                view_by_id,
            )
        elif isinstance(prop, dm.MappedProperty) and isinstance(prop.type, dm.CDFExternalIdReference):
            return CDFExternalField.load(base, prop, variable)
        elif isinstance(prop, dm.MappedProperty) and isinstance(prop.type, dm.PropertyType):
            return BasePrimitiveField.load(base, prop, variable)
        else:
            warnings.warn(
                f"Unsupported property type: {type(prop)}. Skipping field {prop_id}", UserWarning, stacklevel=2
            )
            return None

    def as_read_type_hint(self) -> str:
        raise NotImplementedError()

    def as_write_type_hint(self) -> str:
        raise NotImplementedError()

    def as_graphql_type_hint(self) -> str:
        raise NotImplementedError()

    def as_typed_hint(self, operation: Literal["write", "read"] = "write") -> str:
        raise NotImplementedError()

    def as_typed_init_set(self) -> str:
        return self.name

    def as_write(self) -> str:
        """Used in the .as_write() method for the read version of the data class."""
        raise NotImplementedError

    def as_write_graphql(self) -> str:
        """Used in the .as_write() method for the graphQL version of the data class."""
        return self.as_write()

    def as_read_graphql(self) -> str:
        """Used in the .as_read() method for the graphQL version of the data class."""
        raise NotImplementedError

    def as_value(self) -> str:
        """Used in the ._to_instances_write() method to write the value of the field to the node instance.
        This should only be implemented for container fields, i.e., fields that store their value in a container."""
        raise NotImplementedError

    @property
    def argument_documentation(self) -> str:
        if self.description:
            return self.description
        else:
            return f"The {self.doc_name} field."

    # The properties below are overwritten in the child classes
    @property
    def is_connection(self) -> bool:
        return False

    @property
    def is_time_field(self) -> bool:
        return False

    @property
    def is_timestamp(self) -> bool:
        return False

    @property
    def is_time_series(self) -> bool:
        return False

    @property
    def is_list(self) -> bool:
        return False

    @property
    def is_text_field(self) -> bool:
        return False

    @property
    def is_write_field(self) -> bool:
        return True

as_read_graphql()

Used in the .as_read() method for the graphQL version of the data class.

Source code in cognite/pygen/_core/models/fields/base.py

def as_read_graphql(self) -> str:
    """Used in the .as_read() method for the graphQL version of the data class."""
    raise NotImplementedError

as_value()

Used in the ._to_instances_write() method to write the value of the field to the node instance. This should only be implemented for container fields, i.e., fields that store their value in a container.

Source code in cognite/pygen/_core/models/fields/base.py

def as_value(self) -> str:
    """Used in the ._to_instances_write() method to write the value of the field to the node instance.
    This should only be implemented for container fields, i.e., fields that store their value in a container."""
    raise NotImplementedError

as_write()

Used in the .as_write() method for the read version of the data class.

Source code in cognite/pygen/_core/models/fields/base.py

def as_write(self) -> str:
    """Used in the .as_write() method for the read version of the data class."""
    raise NotImplementedError

as_write_graphql()

Used in the .as_write() method for the graphQL version of the data class.

Source code in cognite/pygen/_core/models/fields/base.py

def as_write_graphql(self) -> str:
    """Used in the .as_write() method for the graphQL version of the data class."""
    return self.as_write()