caikit__caikit-233

{
  "status": "ok",
  "stdout": "./caikit/core/data_model/dataobject.py:89:                    \"All DataObjectBase classes must follow dataclass syntax\"\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "caikit/core/data_model/dataobject.py",
  "content": "    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.",
  "start_line": 50,
  "end_line": 120,
  "total_lines": 386,
  "is_truncated": false
}

{
  "status": "ok",
  "stdout": "Caught expected error: All DataObjectBase classes must follow dataclass syntax\n",
  "stderr": "<function register_backend_type at 0x730a9a9ae170> is still in the BETA phase and subject to change!\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",
  "start_line": 1,
  "end_line": 80,
  "total_lines": 386,
  "is_truncated": false
}

{
  "status": "ok",
  "stdout": "caikit/core/data_model/base.py\n",
  "stderr": "",
  "exit_code": 0
}

{
  "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",
  "filepath": "caikit/core/data_model/dataobject.py",
  "content": "        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:\n                    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)",
  "start_line": 123,
  "end_line": 272,
  "total_lines": 386,
  "is_truncated": true
}

{
  "status": "ok",
  "stdout": "",
  "stderr": "<function register_backend_type at 0x736f494e9240> is still in the BETA phase and subject to change!\n",
  "exit_code": 0
}