@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
@property
def argument_documentation(self) -> str:
if self.description:
msg = self.description
else:
msg = f"The {self.doc_name} field."
line_width = 120 - 10 - len(self.name) - 1
return ("\n" + " " * 12).join(textwrap.wrap(msg, width=line_width))
@property
def argument_documentation_method(self) -> str:
if self.description:
msg = self.description
else:
msg = f"The {self.doc_name} field."
line_width = 120 - 14 - len(self.name) - 1
return ("\n" + " " * 16).join(textwrap.wrap(msg, width=line_width))
# 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