Files
2025-07-23 11:38:50 +02:00

278 lines
9.0 KiB
Python

#
# This file is distributed under the MIT License. See LICENSE.md for details.
#
import copy
from abc import ABC
from typing import Dict, List
from .definition import Definition
from .enum import EnumDefinition
from .reference import ReferenceDefinition
from .sequence import SequenceDefinition
from .upcastable import UpcastableDefinition
# TODO: we should return the fields in a stable order (in all_fields, all_optional_fields, etc)
def is_enum(resolved_type):
return isinstance(resolved_type, EnumDefinition)
class StructField(ABC):
def __init__(self, *, name, doc=None, optional=False, const=False, is_guid=False):
self.name = name
self.doc = doc
self.optional = optional
self.const = const
self.is_guid = is_guid
self.resolved_type = None
@staticmethod
def from_yaml(source_dict: Dict):
if source_dict.get("type"):
return SimpleStructField(**source_dict)
if source_dict.get("sequence"):
sequence_def = source_dict["sequence"]
args = copy.copy(source_dict)
del args["sequence"]
args.update(
{
"sequence_type": sequence_def["type"],
"element_type": sequence_def["elementType"],
"upcastable": sequence_def.get("upcastable", False),
}
)
return SequenceStructField(**args)
if source_dict.get("reference"):
reference = source_dict["reference"]
args = copy.copy(source_dict)
del args["reference"]
args.update(
{
"pointee_type": reference["pointeeType"],
"root_type": reference["rootType"],
}
)
return ReferenceStructField(**args)
raise ValueError("Invalid struct field")
def resolve_references(self, schema):
raise NotImplementedError()
class UpcastableStructField(StructField):
def __init__(self, underlying):
super().__init__(
name=underlying.name,
doc=underlying.doc,
optional=underlying.optional,
const=underlying.const,
is_guid=underlying.is_guid,
)
self.underlying = underlying
def resolve_references(self, schema):
self.underlying.resolve_references(schema)
self.resolved_type = UpcastableDefinition(self.underlying.resolved_type)
assert self.resolved_type
class SimpleStructField(StructField):
def __init__(
self,
*,
name,
type, # noqa: A002
doc=None,
optional=False,
const=False,
upcastable=False,
is_guid=False,
):
super().__init__(name=name, doc=doc, optional=optional, const=const, is_guid=is_guid)
self.type = type
self.upcastable = upcastable
def resolve_references(self, schema):
self.resolved_type = schema.get_definition_for(self.type)
if self.upcastable:
self.resolved_type = UpcastableDefinition(self.resolved_type)
assert self.resolved_type
class SequenceStructField(StructField):
def __init__(
self,
*,
name,
sequence_type,
element_type,
upcastable=False,
doc=None,
optional=False,
const=False,
):
super().__init__(name=name, doc=doc, optional=optional, const=const)
self.sequence_type = sequence_type
self.element_type = element_type
self.upcastable = upcastable
self.resolved_element_type = None
def resolve_references(self, schema):
self.resolved_element_type = schema.get_definition_for(self.element_type)
if self.upcastable:
self.resolved_element_type = UpcastableDefinition(self.resolved_element_type)
self.resolved_type = SequenceDefinition(self.sequence_type, self.resolved_element_type)
class ReferenceStructField(StructField):
def __init__(self, *, name, pointee_type, root_type, doc=None, optional=False, const=False):
super().__init__(name=name, doc=doc, optional=optional, const=const)
self.pointee_type = pointee_type
self.root_type = root_type
def resolve_references(self, schema):
resolved_pointee_type = schema.get_definition_for(self.pointee_type)
resolved_root_type = schema.get_definition_for(self.root_type)
self.resolved_type = ReferenceDefinition(resolved_pointee_type, resolved_root_type)
class StructDefinition(Definition):
def __init__(
self,
*,
name,
fields,
inherits=None,
doc=None,
abstract=False,
_key=None,
):
super().__init__(False, name)
self.doc = doc
# Names of types on which this definition depends on.
# Not necessarily names defined by the user
self.autogenerated = False
self.fields: List[StructField] = fields
self.abstract = abstract
self._key = _key
self._inherits = inherits
# These fields will be populated by resolve_references()
self.inherits = None
self._key_kind_index = None
# Type containing the key fields (might be self or a base class)
self.key_definition = None
# None, "simple" or "composite"
self.keytype = None
self.key_fields: List[SimpleStructField] = []
self.emit_full_constructor = None
self.children: List[StructDefinition] = []
self._refs_resolved = False
def resolve_references(self, schema):
if self._refs_resolved:
return
if self._inherits:
self.inherits = schema.get_definition_for(self._inherits)
self.dependencies.add(self._inherits)
for field in self.fields:
field.resolve_references(schema)
if isinstance(field, SimpleStructField):
self.dependencies.add(field.type)
elif isinstance(field, SequenceStructField):
self.dependencies.add(field.element_type)
elif isinstance(field, ReferenceStructField):
self.dependencies.add(field.pointee_type)
# TODO: if we add this dependency we generate circular dependencies
# self.dependencies.add(field.root_type) # noqa: E800
else:
raise ValueError()
if self._key:
self.key_definition = self
elif self.inherits and self.inherits._key:
# TODO: support indirect inheritance
self.key_definition = self.inherits
# TODO: mark key fields as const
if self.key_definition:
for key_field_index, key_field_name in enumerate(self.key_definition._key):
key_field = next(f for f in self.key_definition.fields if f.name == key_field_name)
assert isinstance(key_field, SimpleStructField)
self.key_fields.append(key_field)
if key_field.name == "Kind":
assert self._key_kind_index is None, "Multiple kind fields in the key"
self._key_kind_index = key_field_index
if len(self.key_definition.key_fields) == 0:
self.keytype = None
elif len(self.key_definition.key_fields) == 1:
self.keytype = "simple"
else:
self.keytype = "composite"
self._refs_resolved = True
key_fields_names = {f.name for f in self.key_fields}
all_field_names = {f.name for f in self.all_fields}
self.emit_full_constructor = key_fields_names != all_field_names
if (self.inherits or self.abstract) and self.keytype:
assert self._key_kind_index is not None, "A polymorphic type without kind in the key"
@staticmethod
def from_dict(source_dict: Dict):
args = {
"name": source_dict["name"],
"doc": source_dict.get("doc"),
"fields": [StructField.from_yaml(d) for d in source_dict["fields"]],
"_key": source_dict.get("key"),
"inherits": source_dict.get("inherits"),
"abstract": source_dict.get("abstract", False),
}
return StructDefinition(**args)
@property
def optional_fields(self):
for field in self.fields:
if field.optional:
yield field
@property
def required_fields(self):
for field in self.fields:
if not field.optional:
yield field
@property
def all_optional_fields(self):
for field in self.all_fields:
if field.optional:
yield field
@property
def all_required_fields(self):
for field in self.all_fields:
if not field.optional:
yield field
@property
def all_fields(self):
if not self._refs_resolved:
raise RuntimeError("all_fields not available before resolving references")
if self.inherits:
yield from self.inherits.fields
yield from self.fields