in elasticsearch/dsl/document_base.py [0:0]
def __init__(self, name: str, bases: Tuple[type, ...], attrs: Dict[str, Any]):
meta = attrs.pop("Meta", None)
# create the mapping instance
self.mapping: Mapping = getattr(meta, "mapping", Mapping())
# register the document's fields, which can be given in a few formats:
#
# class MyDocument(Document):
# # required field using native typing
# # (str, int, float, bool, datetime, date)
# field1: str
#
# # optional field using native typing
# field2: Optional[datetime]
#
# # array field using native typing
# field3: list[int]
#
# # sub-object, same as Object(MyInnerDoc)
# field4: MyInnerDoc
#
# # nested sub-objects, same as Nested(MyInnerDoc)
# field5: list[MyInnerDoc]
#
# # use typing, but override with any stock or custom field
# field6: bool = MyCustomField()
#
# # best mypy and pyright support and dataclass-like behavior
# field7: M[date]
# field8: M[str] = mapped_field(MyCustomText(), default="foo")
#
# # legacy format without Python typing
# field9 = Text()
#
# # ignore attributes
# field10: ClassVar[string] = "a regular class variable"
annotations = attrs.get("__annotations__", {})
fields = set([n for n in attrs if isinstance(attrs[n], Field)])
fields.update(annotations.keys())
field_defaults = {}
for name in fields:
value: Any = None
required = None
multi = None
if name in annotations:
# the field has a type annotation, so next we try to figure out
# what field type we can use
type_ = annotations[name]
skip = False
required = True
multi = False
while hasattr(type_, "__origin__"):
if type_.__origin__ == ClassVar:
skip = True
break
elif type_.__origin__ == Mapped:
# M[type] -> extract the wrapped type
type_ = type_.__args__[0]
elif type_.__origin__ == Union:
if len(type_.__args__) == 2 and type_.__args__[1] is type(None):
# Optional[type] -> mark instance as optional
required = False
type_ = type_.__args__[0]
else:
raise TypeError("Unsupported union")
elif type_.__origin__ in [list, List]:
# List[type] -> mark instance as multi
multi = True
required = False
type_ = type_.__args__[0]
else:
break
if skip or type_ == ClassVar:
# skip ClassVar attributes
continue
if type(type_) is UnionType:
# a union given with the pipe syntax
args = get_args(type_)
if len(args) == 2 and args[1] is type(None):
required = False
type_ = type_.__args__[0]
else:
raise TypeError("Unsupported union")
field = None
field_args: List[Any] = []
field_kwargs: Dict[str, Any] = {}
if isinstance(type_, type) and issubclass(type_, InnerDoc):
# object or nested field
field = Nested if multi else Object
field_args = [type_]
elif type_ in self.type_annotation_map:
# use best field type for the type hint provided
field, field_kwargs = self.type_annotation_map[type_] # type: ignore[assignment]
if field:
field_kwargs = {
"multi": multi,
"required": required,
**field_kwargs,
}
value = field(*field_args, **field_kwargs)
if name in attrs:
# this field has a right-side value, which can be field
# instance on its own or wrapped with mapped_field()
attr_value = attrs[name]
if isinstance(attr_value, dict):
# the mapped_field() wrapper function was used so we need
# to look for the field instance and also record any
# dataclass-style defaults
attr_value = attrs[name].get("_field")
default_value = attrs[name].get("default") or attrs[name].get(
"default_factory"
)
if default_value:
field_defaults[name] = default_value
if attr_value:
value = attr_value
if required is not None:
value._required = required
if multi is not None:
value._multi = multi
if value is None:
raise TypeError(f"Cannot map field {name}")
self.mapping.field(name, value)
if name in attrs:
del attrs[name]
# store dataclass-style defaults for ObjectBase.__init__ to assign
attrs["_defaults"] = field_defaults
# add all the mappings for meta fields
for name in dir(meta):
if isinstance(getattr(meta, name, None), MetaField):
params = getattr(meta, name)
self.mapping.meta(name, *params.args, **params.kwargs)
# document inheritance - include the fields from parents' mappings
for b in bases:
if hasattr(b, "_doc_type") and hasattr(b._doc_type, "mapping"):
self.mapping.update(b._doc_type.mapping, update_only=True)