caikit__caikit-233

{
  "status": "ok",
  "stdout": "Caught expected exception: All DataObjectBase classes must follow dataclass syntax\n",
  "stderr": "<function register_backend_type at 0x704f057ae170> is still in the BETA phase and subject to change!\nTraceback (most recent call last):\n  File \"<stdin>\", line 11, in <module>\n  File \"/workspace/caikit/core/data_model/dataobject.py\", line 88, in __new__\n    raise TypeError(\nTypeError: All DataObjectBase classes must follow dataclass syntax\n",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "caikit/core/data_model/dataobject.py",
  "content": "# Copyright The Caikit Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\"\"\"This module defines the @schema decorator which can be used to declare data\nmodel objects inline without manually defining the protobufs representation\n\"\"\"\n\n\n# Standard\nfrom enum import Enum\nfrom typing import Any, Callable, List, Type, Union, get_args, get_origin\nimport dataclasses\n\n# Third Party\nfrom google.protobuf import descriptor as _descriptor\nfrom google.protobuf import message as _message\nfrom google.protobuf import struct_pb2\nfrom google.protobuf.internal.enum_type_wrapper import EnumTypeWrapper\nimport numpy as np\n\n# First Party\nfrom py_to_proto.dataclass_to_proto import PY_TO_PROTO_TYPES, DataclassConverter\nimport alog\nimport py_to_proto\n\n# Local\nfrom ..toolkit.errors import error_handler\nfrom . import enums\nfrom .base import DataBase, _DataBaseMetaClass\nfrom .json_dict import JsonDict\n\n## Globals #####################################################################\n\nlog = alog.use_channel(\"SCHEMA\")\nerror = error_handler.get(log)\n\n# Type mapping for type hints in @dataobject classes\nDATAOBJECT_PY_TO_PROTO_TYPES = {\n    JsonDict: struct_pb2.Struct,\n    np.int32: _descriptor.FieldDescriptor.TYPE_INT32,\n    np.int64: _descriptor.FieldDescriptor.TYPE_INT64,\n    np.uint32: _descriptor.FieldDescriptor.TYPE_UINT32,\n    np.uint64: _descriptor.FieldDescriptor.TYPE_UINT64,\n    np.float32: _descriptor.FieldDescriptor.TYPE_FLOAT,\n    np.float64: _descriptor.FieldDescriptor.TYPE_DOUBLE,\n    **PY_TO_PROTO_TYPES,\n}\n\n# Common package prefix\nCAIKIT_DATA_MODEL = \"caikit_data_model\"\n\n# Registry of auto-generated protos so that they can be rendered to .proto\n_AUTO_GEN_PROTO_CLASSES = []\n\n# Special attribute used to indicate which defaults are user provided\n_USER_DEFINED_DEFAULTS = \"__user_defined_defaults__\"\n\n## Public ######################################################################\n\n\nclass _DataObjectBaseMetaClass(_DataBaseMetaClass):\n    \"\"\"This metaclass is used for the DataObject base class so that all data\n    objects can delay the creation of their proto class until after the\n    metaclass has been instantiated.\n    \"\"\"\n\n    def __new__(mcs, name, bases, attrs):\n        \"\"\"When instantiating a new DataObject class, the proto class will not\n        yet have been generated, but the set of fields will be known since the\n        class will be the raw input representation of a @dataclass\n        \"\"\"\n\n        # Get the annotations that will go into the dataclass\n        if name != \"DataObjectBase\":\n            field_names = attrs.get(\"__annotations__\")\n            if field_names is None:\n                raise TypeError(\n                    \"All DataObjectBase classes must follow dataclass syntax\"\n                )\n            attrs[_DataBaseMetaClass._FWD_DECL_FIELDS] = field_names\n\n        # Delegate to the base metaclass\n        return super().__new__(mcs, name, bases, attrs)\n\n\nclass DataObjectBase(DataBase, metaclass=_DataObjectBaseMetaClass):\n    \"\"\"A DataObject is a data model class that is backed by a @dataclass.\n\n    Data model classes that use the @dataobject decorator must derive from this\n    base class.\n    \"\"\"\n\n\ndef dataobject(*args, **kwargs) -> Callable[[Type], Type[DataBase]]:\n    \"\"\"The @dataobject decorator can be used to define a Data Model object's\n    schema inline with the definition of the python class rather than needing to\n    bind to a pre-compiled protobufs class. For example:\n\n    @dataobject(\"foo.bar\")\n    @dataclass\n    class MyDataObject(DataObjectBase):\n        '''My Custom Data Object'''\n        foo: str\n        bar: int\n\n    NOTE: The wrapped class must NOT inherit directly from DataBase. That\n        inheritance will be added by this decorator, but if it is written\n        directly, the metaclass that links protobufs to the class will be called\n        before this decorator can auto-gen the protobufs class.\n\n    Args:\n        package:  str\n            The package name to use for the generated protobufs class\n\n    Returns:\n        decorator:  Callable[[Type], Type[DataBase]]\n            The decorator function that will wrap the given class\n    \"\"\"\n\n    def decorator(cls: Type) -> Type[DataBase]:\n        # Make sure that the wrapped class does NOT inherit from DataBase\n        error.value_check(\n            \"<COR95184230E>\",\n            issubclass(cls, (DataObjectBase, Enum)),\n            \"{} must inherit from DataObjectBase/Enum when using @dataobject\",\n            cls.__name__,\n        )\n\n        # Add the package to the kwargs\n        kwargs.setdefault(\"package\", package)\n\n        # If it's not an enum, fill in any missing field defaults as None\n        # and make sure it's a dataclass\n        if not issubclass(cls, Enum):\n            log.debug2(\"Wrapping data class %s\", cls)\n            user_defined_defaults = {}\n            for annotation in getattr(cls, \"__annotations__\", {}):\n                user_defined_default = getattr(cls, annotation, dataclasses.MISSING)\n                if user_defined_default == dataclasses.MISSING:",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 386,
  "is_truncated": true
}

{
  "status": "ok",
  "filepath": "caikit/core/data_model/dataobject.py",
  "content": "                    log.debug3(\"Filling in None default for %s.%s\", cls, annotation)\n                    setattr(cls, annotation, None)\n                else:\n                    user_defined_defaults[annotation] = user_defined_default\n            # If the current __init__ is auto-generated by dataclass, remove\n            # it so that a new one is created with the new defaults\n            if _has_dataclass_init(cls):\n                log.debug3(\"Resetting default dataclass init\")\n                delattr(cls, \"__init__\")\n            cls = dataclasses.dataclass(cls)\n            setattr(cls, _USER_DEFINED_DEFAULTS, user_defined_defaults)\n\n        descriptor = _dataobject_to_proto(dataclass_=cls, **kwargs)\n\n        # Create the message class from the dataclass\n        proto_class = py_to_proto.descriptor_to_message_class(descriptor)\n        _AUTO_GEN_PROTO_CLASSES.append(proto_class)\n\n        # Add enums to the global enums module\n        for enum_class in _get_all_enums(proto_class):\n            log.debug2(\"Importing enum [%s]\", enum_class.DESCRIPTOR.name)\n            enums.import_enum(enum_class)\n\n        # Declare the merged class that binds DataBase to the wrapped class with\n        # this generated proto class\n        if isinstance(proto_class, type):\n            setattr(cls, \"_proto_class\", proto_class)\n            cls = _make_data_model_class(proto_class, cls)\n\n            # If this was a default-generated dataclass __init__ and there are\n            # any oneofs, we need to augment the __init__ to support kwargs for\n            # the individual fields\n            if _has_dataclass_init(cls) and cls._fields_oneofs_map:\n                setattr(cls, \"__init__\", _make_oneof_init(cls))\n\n        else:\n            enums.import_enum(proto_class, cls)\n            setattr(cls, \"_proto_enum\", proto_class)\n\n        # Return the decorated class\n        return cls\n\n    # If called without the function invocation, fill in the default argument\n    if args and callable(args[0]):\n        assert not kwargs, \"This shouldn't happen!\"\n        package = CAIKIT_DATA_MODEL\n        return decorator(args[0])\n\n    # Pull the package as an arg or a keyword arg\n    if args:\n        package = args[0]\n        if \"package\" in kwargs:\n            raise TypeError(\"Got multiple values for argument 'package'\")\n    else:\n        package = kwargs.get(\"package\", CAIKIT_DATA_MODEL)\n    return decorator\n\n\ndef render_dataobject_protos(interfaces_dir: str):\n    \"\"\"Write out protobufs files for all proto classes generated from dataobjects\n    to the target interfaces directory\n\n    Args:\n        interfaces_dir:  str\n            The target directory (must already exist)\n    \"\"\"\n    for proto_class in _AUTO_GEN_PROTO_CLASSES:\n        proto_class.write_proto_file(interfaces_dir)\n\n\n## Implementation Details ######################################################\n\n\ndef _dataobject_to_proto(*args, **kwargs):\n    kwargs.setdefault(\"type_mapping\", DATAOBJECT_PY_TO_PROTO_TYPES)\n    return _DataobjectConverter(*args, **kwargs).descriptor\n\n\nclass _DataobjectConverter(DataclassConverter):\n    \"\"\"Augment the dataclass converter to be able to pull descriptors from\n    existing data objects\n    \"\"\"\n\n    def get_concrete_type(self, entry: Any) -> Any:\n        \"\"\"Also include data model classes and enums as concrete types\"\"\"\n        unwrapped = self._resolve_wrapped_type(entry)\n        if (\n            isinstance(unwrapped, type)\n            and issubclass(unwrapped, DataBase)\n            and unwrapped._proto_class is not None\n        ) or hasattr(unwrapped, \"_proto_enum\"):\n            return entry\n        return super().get_concrete_type(entry)\n\n    def get_descriptor(self, entry: Any) -> Any:\n        \"\"\"Unpack data model classes and enums to their descriptors\"\"\"\n        entry = self._resolve_wrapped_type(entry)\n        if isinstance(entry, type) and issubclass(entry, DataBase):\n            return entry._proto_class.DESCRIPTOR\n        proto_enum = getattr(entry, \"_proto_enum\", None)\n        if proto_enum is not None:\n            return proto_enum.DESCRIPTOR\n        return super().get_descriptor(entry)\n\n    def get_optional_field_names(self, entry: Any) -> List[str]:\n        \"\"\"Get the names of any fields which are optional. This will be any\n        field that has a user-defined default or is marked as Optional[]\n        \"\"\"\n        optional_fields = list(getattr(entry, _USER_DEFINED_DEFAULTS, {}))\n        for field_name, field in entry.__dataclass_fields__.items():\n            if (\n                field_name not in optional_fields\n                and self._is_python_optional(field.type) is not None\n            ):\n                optional_fields.append(field_name)\n        return optional_fields\n\n    @staticmethod\n    def _is_python_optional(entry: Any) -> Any:\n        \"\"\"Detect if this type is a python optional\"\"\"\n        if get_origin(entry) is Union:\n            args = get_args(entry)\n            return type(None) in args\n\n\ndef _get_all_enums(\n    proto_class: Union[_message.Message, EnumTypeWrapper],\n) -> List[EnumTypeWrapper]:\n    \"\"\"Given a generated proto class, recursively extract all enums\"\"\"\n    all_enums = []\n    if isinstance(proto_class, EnumTypeWrapper):\n        all_enums.append(proto_class)\n    else:\n        for enum_descriptor in proto_class.DESCRIPTOR.enum_types:\n            all_enums.append(getattr(proto_class, enum_descriptor.name))\n        for nested_proto_descriptor in proto_class.DESCRIPTOR.nested_types:\n            all_enums.extend(\n                _get_all_enums(getattr(proto_class, nested_proto_descriptor.name))\n            )\n\n    return all_enums\n\n\ndef _make_data_model_class(proto_class, cls):\n    if issubclass(cls, DataObjectBase):\n        _DataBaseMetaClass.parse_proto_descriptor(cls)\n\n    # Recursively make all nested message wrappers\n    for nested_message_descriptor in proto_class.DESCRIPTOR.nested_types:\n        nested_message_name = nested_message_descriptor.name",
  "start_line": 151,
  "end_line": 300,
  "total_lines": 386,
  "is_truncated": true
}

{
  "status": "ok",
  "filepath": "caikit/core/data_model/dataobject.py",
  "content": "        nested_proto_class = getattr(proto_class, nested_message_name)\n        setattr(\n            cls,\n            nested_message_name,\n            _make_data_model_class(\n                nested_proto_class,\n                _DataBaseMetaClass.__new__(\n                    _DataBaseMetaClass,\n                    name=nested_message_name,\n                    bases=(DataBase,),\n                    attrs={\"_proto_class\": getattr(proto_class, nested_message_name)},\n                ),\n            ),\n        )\n    for nested_enum_descriptor in proto_class.DESCRIPTOR.enum_types:\n        setattr(\n            cls,\n            nested_enum_descriptor.name,\n            getattr(enums, nested_enum_descriptor.name),\n        )\n\n    return cls\n\n\ndef _make_oneof_init(cls):\n    \"\"\"Helper to augment a defaulted dataclass __init__ to support kwargs for\n    oneof fields\n    \"\"\"\n    original_init = cls.__init__\n    fields_to_oneofs = cls._fields_to_oneof\n    oneofs_to_fields = cls._fields_oneofs_map\n\n    def __init__(self, *args, **kwargs):\n        new_kwargs = {}\n        to_remove = []\n        which_oneof = {}\n        for field_name, val in kwargs.items():\n            if oneof_name := fields_to_oneofs.get(field_name):\n                oneof_pos_idx = list(cls.__dataclass_fields__.keys()).index(oneof_name)\n                has_pos_val = len(args) > oneof_pos_idx\n                if has_pos_val:\n                    error(\n                        \"<COR09282193E>\",\n                        TypeError(\n                            \"Received conflicting oneof args/kwargs for {}/{}\".format(\n                                oneof_name,\n                                field_name,\n                            )\n                        ),\n                    )\n\n                other_oneof_fields = (\n                    field\n                    for field in [oneof_name] + oneofs_to_fields[oneof_name]\n                    if field != field_name\n                )\n                if any(field in kwargs for field in other_oneof_fields):\n                    error(\n                        \"<COR59933157E>\",\n                        TypeError(\n                            \"Received multiple keyword arguments for oneof {}\".format(\n                                oneof_name,\n                            )\n                        ),\n                    )\n                new_kwargs[oneof_name] = val\n                to_remove.append(field_name)\n                which_oneof[oneof_name] = field_name\n\n        for kwarg in to_remove:\n            del kwargs[kwarg]\n        kwargs.update(new_kwargs)\n        original_init(self, *args, **kwargs)\n        setattr(self, _DataBaseMetaClass._WHICH_ONEOF_ATTR, which_oneof)\n\n    return __init__\n\n\ndef _has_dataclass_init(cls) -> bool:\n    \"\"\"When the dataclass decorator adds an __init__ to a class, it adds\n    __annotations__ to the init function itself. This function uses that fact to\n    detect if the class's __init__ function was generated by @dataclass\n    \"\"\"\n    return bool(getattr(cls.__init__, \"__annotations__\", None)) and not any(\n        cls.__init__ is base.__init__ for base in cls.__bases__\n    )",
  "start_line": 301,
  "end_line": 386,
  "total_lines": 386,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "caikit/core/data_model/base.py",
  "content": "# Copyright The Caikit Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\n\"\"\"Base classes and functionality for all data structures.\n\"\"\"\n\n# metaclass-generated field members cannot be detected by pylint\n# pylint: disable=no-member\n\n# Standard\nfrom dataclasses import dataclass\nfrom enum import Enum\nfrom typing import Any, Dict, Optional, Type, Union\nimport base64\nimport json\n\n# Third Party\nfrom google.protobuf import json_format\nfrom google.protobuf.descriptor import Descriptor\nfrom google.protobuf.internal import type_checkers as proto_type_checkers\nfrom google.protobuf.message import Message as ProtoMessageType\n\n# First Party\nimport alog\n\n# Local\nfrom ..toolkit.errors import error_handler\nfrom . import enums, json_dict\n\nlog = alog.use_channel(\"DATAM\")\nerror = error_handler.get(log)\n\n\nclass _DataBaseMetaClass(type):\n    \"\"\"Meta class for all structures in the data model.\"\"\"\n\n    # store a registry of all classes that use this metaclass, i.e.,\n    # all classes that extend DataBase. This is used for constructing new\n    # instances by name without having to introspect all modules in data_model.\n    class_registry = {}\n\n    # This sentinel value is used to determine whether a given attribute is\n    # present on a class without doing `getattr` twice in the case where the\n    # attribute does exist.\n    _MISSING_ATTRIBUTE = \"missing attribute\"\n\n    # Special attribute used to communicate that the proto fields are forward\n    # declared and will be populated after the metaclass has completed\n    # construction.\n    _FWD_DECL_FIELDS = \"__fwd_decl_fields__\"\n\n    # Special instance attributes that an instance of a class derived from\n    # DataBase may have. These are added to __slots__.\n    _BACKEND_ATTR = \"_backend\"\n    _WHICH_ONEOF_ATTR = \"_which_oneof\"\n\n    def __new__(mcs, name, bases, attrs):\n        \"\"\"When constructing a new data model class, we set the 'fields' class variable from the\n        protobufs descriptor and then set the '__slots__' magic class attribute to fields.  This\n        provides two benefits: (a) performance is improved since the classes only need to know\n        about these attributes (b) it helps to enforce that all member variables in these classes\n        are described in the protobufs.\n\n        Note:  If you want to add a variable for internal use that is not described in the\n        protobufs, it can be named in the tuple class variable _private_slots and will\n        automatically be added to __slots__.\n        \"\"\"\n\n        # Protobufs fields can be divided into these categories, which are used\n        # to automatically determine appropriate behavior in a number of methods\n        attrs[\"full_name\"] = name\n        attrs[\"fields_enum_map\"] = {}\n        attrs[\"fields_enum_rev\"] = {}\n        attrs[\"_fields_oneofs_map\"] = {}\n        attrs[\"_fields_to_oneof\"] = {}\n        attrs[\"_fields_map\"] = ()\n        attrs[\"_fields_message\"] = ()\n        attrs[\"_fields_message_repeated\"] = ()\n        attrs[\"_fields_enum\"] = ()\n        attrs[\"_fields_enum_repeated\"] = ()\n        attrs[\"_fields_primitive\"] = ()\n        attrs[\"_fields_primitive_repeated\"] = ()\n\n        # Look for the set of fields either from a predefined protobuf class or\n        # from a forward declaration from @dataobject\n        fields = ()\n        proto_class = None\n        if name not in [\"DataBase\", \"DataObjectBase\"]:\n            # Look for a precompiled proto class and if found, parse its\n            # descriptor\n            proto_class = attrs.get(\"_proto_class\")\n            if proto_class is not None:\n                all_oneof_fields = [\n                    field.name\n                    for oneof in proto_class.DESCRIPTOR.oneofs\n                    for field in oneof.fields\n                ]\n                fields = tuple(\n                    (\n                        field\n                        for field in proto_class.DESCRIPTOR.fields_by_name\n                        if field not in all_oneof_fields\n                    )\n                ) + tuple(proto_class.DESCRIPTOR.oneofs_by_name)\n\n            # Otherwise, we need to get the fields from a \"special\" attribute\n            else:\n                fields = attrs.pop(mcs._FWD_DECL_FIELDS, None)\n                log.debug4(\n                    \"Using dataclass forward declaration fields %s for %s\", fields, name\n                )\n                error.value_check(\n                    \"<COR49310991E>\",\n                    fields is not None,\n                    \"No proto class found for {}\",\n                    name,\n                )\n        attrs[\"fields\"] = fields\n        attrs[\"_proto_class\"] = proto_class\n\n        # Look if any private slots are declared as class variables\n        private_slots = attrs.setdefault(\"_private_slots\", ())\n\n        # Class slots are fields + private slots, this prevents other\n        # member attributes from being set and also improves performance\n        attrs[\"__slots__\"] = tuple(\n            [f\"_{field}\" for field in fields]\n            + list(private_slots)\n            + [mcs._BACKEND_ATTR, mcs._WHICH_ONEOF_ATTR]\n        )\n\n        # Create the instance of the type\n        instance = super().__new__(mcs, name, bases, attrs)\n\n        # If there's a valid proto class, perform proto descriptor parsing\n        if proto_class is not None:\n            mcs.parse_proto_descriptor(instance)\n",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 990,
  "is_truncated": true
}

{
  "status": "ok",
  "stdout": "Base instance: Base(foo=1)\nDerived instance: Derived(foo=1, bar=2)\n",
  "stderr": "<function register_backend_type at 0x70a3e88e9240> is still in the BETA phase and subject to change!\n",
  "exit_code": 0
}

{
  "status": "ok",
  "stdout": "155:    def parse_proto_descriptor(mcs, cls):\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "caikit/core/data_model/base.py",
  "content": "    def parse_proto_descriptor(mcs, cls):\n        \"\"\"Encapsulate the logic for parsing the protobuf descriptor here. This\n        allows the parsing to be done as a post-process after metaclass\n        initialization\n        \"\"\"\n\n        # use the fully qualified protobuf name to avoid conflicts with\n        # nested messages that have matching names\n        cls.full_name = cls._proto_class.DESCRIPTOR.full_name\n\n        # preserve old fields for _make_property_getter later\n        old_fields = cls.fields\n\n        # overwrite to only have proto-specific fields present\n        cls.fields = tuple(cls._proto_class.DESCRIPTOR.fields_by_name)\n\n        # map from all enum fields to their enum classes\n        # note: enums are also primitives, these overlap\n        cls.fields_enum_map = {\n            field.name: getattr(enums, field.enum_type.name)\n            for field in cls._proto_class.DESCRIPTOR.fields\n            if field.enum_type is not None\n        }\n\n        cls.fields_enum_rev = {\n            field.name: getattr(enums, field.enum_type.name + \"Rev\")\n            for field in cls._proto_class.DESCRIPTOR.fields\n            if field.enum_type is not None\n        }\n\n        # mapping of all oneofs and the fields that are part of them\n        # NOTE: protobuf makes an interesting use of oneof to wrap types that\n        #   should be explicitly optional. We don't want to consider these\n        #   oneofs in the general oneof handling.\n        cls._fields_oneofs_map = {\n            oneof_name: [field.name for field in oneof.fields]\n            for oneof_name, oneof in cls._proto_class.DESCRIPTOR.oneofs_by_name.items()\n            if len(oneof.fields) != 1 or oneof.name != f\"_{oneof.fields[0].name}\"\n        }\n        cls._fields_to_oneof = {\n            field_name: oneof_name\n            for (oneof_name, oneof_fields) in cls._fields_oneofs_map.items()\n            for field_name in oneof_fields\n        }\n\n        # all repeated fields\n        fields_repeated = tuple(\n            field.name\n            for field in cls._proto_class.DESCRIPTOR.fields\n            if field.label == field.LABEL_REPEATED\n        )\n\n        # all messages, repeated or not\n        _fields_message_all = tuple(\n            field.name\n            for field in cls._proto_class.DESCRIPTOR.fields\n            if field.type == field.TYPE_MESSAGE\n        )\n\n        # all enums, repeated or not\n        _fields_enum_all = tuple(\n            field.name\n            for field in cls._proto_class.DESCRIPTOR.fields\n            if field.enum_type is not None\n        )\n\n        # all fields of type map\n        cls._fields_map = tuple(\n            field.name\n            for field in cls._proto_class.DESCRIPTOR.fields\n            if field.message_type and field.message_type.GetOptions().map_entry\n        )\n\n        # all primitives, repeated or not\n        _fields_primitive_all = (\n            frozenset(cls.fields)\n            .difference(cls._fields_map)\n            .difference(_fields_message_all)\n            .difference(_fields_enum_all)\n        )\n\n        # messages that are not repeated\n        cls._fields_message = frozenset(_fields_message_all).difference(fields_repeated)\n\n        # messages that are repeated\n        cls._fields_message_repeated = frozenset(fields_repeated).intersection(\n            _fields_message_all\n        )\n\n        # enums that are not repeated\n        cls._fields_enum = frozenset(_fields_enum_all).difference(fields_repeated)\n\n        # enums that are repeated\n        cls._fields_enum_repeated = frozenset(_fields_enum_all).intersection(\n            fields_repeated\n        )\n\n        # primitives that are not repeated\n        cls._fields_primitive = frozenset(_fields_primitive_all).difference(\n            fields_repeated\n        )\n\n        # primitives that are repeated\n        cls._fields_primitive_repeated = frozenset(fields_repeated).intersection(\n            _fields_primitive_all\n        )\n\n        # Update the global class and proto registries\n        # NOTE: Explicitly not respecting metaclass inheritance so single\n        #   registry shared for all\n        _DataBaseMetaClass.class_registry[cls.full_name] = cls\n\n        # Add properties that use the underlying backend. Also add fields that\n        # existed in old_fields for supporting oneofs\n        # see https://github.com/caikit/caikit/pull/107 for details\n        for field in set(cls.fields + tuple(old_fields)):\n            # If the field is the name of a field within a oneof and it was not\n            # in the old fields, the data is held under the oneof's name if this\n            # is the set value for the oneof\n            if oneof_name := cls._fields_to_oneof.get(field):\n                setattr(cls, field, mcs._make_property_getter(field, oneof_name))\n\n            # If the field is a plain field or the name of a oneof, it will be\n            # accessed directly\n            else:\n                setattr(cls, field, mcs._make_property_getter(field))\n\n        # If there is not already an __init__ function defined, make one\n        current_init = cls.__init__\n        if current_init is None or current_init is DataBase.__init__:\n            setattr(cls, \"__init__\", mcs._make_init(cls.fields))\n\n    @classmethod\n    def _make_property_getter(mcs, field, oneof_name=None):\n        \"\"\"This helper creates an @property attribute getter for the given field\n\n        NOTE: This needs to live as a standalone function in order for the given\n            field name to be properly bound to the closure for the attrs\n        \"\"\"\n        private_name = f\"_{field}\" if oneof_name is None else oneof_name\n\n        def _property_getter(self):\n            # Check to see if the private name is defined and just return it if\n            # it is\n            current = getattr(self, private_name, mcs._MISSING_ATTRIBUTE)\n            if current is not mcs._MISSING_ATTRIBUTE:\n                return current\n\n            # If not currently set, delegate to the backend\n            backend = self.backend",
  "start_line": 155,
  "end_line": 304,
  "total_lines": 990,
  "is_truncated": true
}

{
  "status": "ok",
  "stdout": "A annotations: {'x': <class 'int'>}\nB annotations: {}\nClass A: attrs has __annotations__ = {'x': <class 'int'>}\nClass B: attrs has __annotations__ = None\n",
  "stderr": "",
  "exit_code": 0
}