Skip to content

datarobot_genai.drtools.panels.schema_registry

schema_registry

Schema registry for Json panel validation (ported from wren-mcp).

A process-wide registry of namespaced Pydantic schemas (e.g. "cuopt.VRPData") that agents can discover (:meth:SchemaRegistry.list_schemas), introspect (:meth:SchemaRegistry.describe), and validate data against (:meth:SchemaRegistry.validate) before storing it on a Json panel. Domains register their schemas at import time::

from datarobot_genai.drtools.panels.schema_registry import SchemaRegistry

SchemaRegistry.register("myapp.MyModel", MyModel)

SchemaValidationError

Bases: ValueError

Raised when data fails schema validation.

Source code in datarobot_genai/drtools/panels/schema_registry.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
class SchemaValidationError(ValueError):
    """Raised when data fails schema validation."""

    def __init__(self, schema_name: str, data: dict[str, Any], errors: list[dict[str, Any]]):
        self.schema_name = schema_name
        self.data = data
        self.errors = errors

        error_lines = [f"Schema validation failed for '{schema_name}':"]
        for err in errors:
            loc = " -> ".join(str(x) for x in err.get("loc", []))
            msg = err.get("msg", "Unknown error")
            err_type = err.get("type", "")
            error_lines.append(f"  • {loc}: {msg}" if loc else f"  • {msg}")
            if err_type:
                error_lines.append(f"    (error type: {err_type})")

        super().__init__("\n".join(error_lines))

SchemaRegistry

Central registry of Pydantic schemas for Json panel validation.

Source code in datarobot_genai/drtools/panels/schema_registry.py 
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
class SchemaRegistry:
    """Central registry of Pydantic schemas for Json panel validation."""

    _schemas: dict[str, type[BaseModel]] = {}

    @classmethod
    def register(cls, name: str, schema: type[BaseModel]) -> None:
        """Register a Pydantic model class under a namespaced ``name``."""
        if not isinstance(schema, type) or not issubclass(schema, BaseModel):
            raise TypeError(f"Schema must be a Pydantic BaseModel class, got {type(schema)}")
        cls._schemas[name] = schema
        logger.debug("Registered schema: %s", name)

    @classmethod
    def unregister(cls, name: str) -> bool:
        """Remove a schema from the registry. Returns True if it existed."""
        return cls._schemas.pop(name, None) is not None

    @classmethod
    def get(cls, name: str) -> type[BaseModel]:
        """Get a schema by name; raises ``KeyError`` if not registered."""
        if name not in cls._schemas:
            available = ", ".join(sorted(cls._schemas.keys()))
            raise KeyError(
                f"Schema '{name}' not found. Available schemas: {available or '(none registered)'}"
            )
        return cls._schemas[name]

    @classmethod
    def exists(cls, name: str) -> bool:
        """Check if a schema is registered."""
        return name in cls._schemas

    @classmethod
    def list_schemas(cls, namespace: str | None = None) -> dict[str, dict[str, Any]]:
        """List registered schemas (optionally filtered by ``namespace``) with summaries."""
        result = {}
        for name, schema in sorted(cls._schemas.items()):
            if namespace and not name.startswith(f"{namespace}."):
                continue
            result[name] = {
                "name": name,
                "description": (schema.__doc__ or "").strip().split("\n")[0],
                "required_fields": cls._get_required_fields(schema),
                "optional_fields": cls._get_optional_fields(schema),
            }
        return result

    @classmethod
    def describe(cls, name: str) -> dict[str, Any]:
        """Detailed schema description: fields, full JSON Schema, and an example."""
        schema = cls.get(name)
        return {
            "name": name,
            "description": (schema.__doc__ or "").strip(),
            "fields": cls._describe_fields(schema),
            "json_schema": schema.model_json_schema(),
            "example": cls._generate_example(schema),
        }

    @classmethod
    def validate(cls, name: str, data: dict[str, Any]) -> dict[str, Any]:
        """Validate ``data`` and return it normalized; raises :class:`SchemaValidationError`."""
        schema = cls.get(name)
        try:
            validated = schema.model_validate(data)
            return validated.model_dump(mode="python", exclude_none=True)
        except ValidationError as e:
            raise SchemaValidationError(
                schema_name=name,
                data=data,
                errors=[dict(err) for err in e.errors()],
            ) from e

    @classmethod
    def _describe_fields(cls, schema: type[BaseModel]) -> dict[str, dict[str, Any]]:
        fields = {}
        for name, field_info in schema.model_fields.items():
            annotation = field_info.annotation
            field_desc: dict[str, Any] = {
                "type": cls._type_to_string(annotation),
                "required": field_info.is_required(),
            }

            if field_info.default is not None and field_info.default is not PydanticUndefined:
                default_val = field_info.default
                try:
                    json.dumps(default_val)
                    field_desc["default"] = default_val
                except (TypeError, ValueError):
                    field_desc["default"] = str(default_val)

            if field_info.description:
                field_desc["description"] = field_info.description

            origin = get_origin(annotation)
            if origin is list:
                args = get_args(annotation)
                if args and isinstance(args[0], type) and issubclass(args[0], BaseModel):
                    field_desc["items_schema"] = args[0].__name__
            elif isinstance(annotation, type) and issubclass(annotation, BaseModel):
                field_desc["nested_schema"] = annotation.__name__

            fields[name] = field_desc

        return fields

    @classmethod
    def _get_required_fields(cls, schema: type[BaseModel]) -> list[str]:
        return [name for name, info in schema.model_fields.items() if info.is_required()]

    @classmethod
    def _get_optional_fields(cls, schema: type[BaseModel]) -> list[str]:
        return [name for name, info in schema.model_fields.items() if not info.is_required()]

    @classmethod
    def _type_to_string(cls, annotation: Any) -> str:
        if annotation is None:
            return "None"

        origin = get_origin(annotation)
        args = get_args(annotation)

        if origin is None:
            if hasattr(annotation, "__name__"):
                return str(annotation.__name__)
            return str(annotation)

        origin_name = getattr(origin, "__name__", str(origin))
        if args:
            args_str = ", ".join(cls._type_to_string(arg) for arg in args)
            return f"{origin_name}[{args_str}]"
        return origin_name

    @classmethod
    def _generate_example(cls, schema: type[BaseModel]) -> dict[str, Any] | None:
        try:
            return {
                name: cls._example_for_type(info.annotation, name)
                for name, info in schema.model_fields.items()
                if info.is_required()
            }
        except Exception:  # noqa: BLE001 - examples are best-effort decoration
            return None

    @classmethod
    def _example_for_type(cls, annotation: Any, field_name: str = "") -> Any:
        origin = get_origin(annotation)
        args = get_args(annotation)

        if origin is Union:
            non_none = [arg for arg in args if arg is not type(None)]
            return cls._example_for_type(non_none[0], field_name) if non_none else None
        if origin is list:
            return [cls._example_for_type(args[0], field_name)] if args else []
        if origin is dict:
            return {}
        if origin is tuple:
            return tuple(
                cls._example_for_type(arg, f"{field_name}[{i}]") for i, arg in enumerate(args)
            )
        if isinstance(annotation, type) and issubclass(annotation, BaseModel):
            return cls._generate_example(annotation)

        scalar_examples: dict[Any, Any] = {
            str: f"example_{field_name}" if field_name else "example",
            int: 1,
            float: 1.0,
            bool: True,
        }
        return scalar_examples.get(annotation)

register classmethod 

register(name: str, schema: type[BaseModel]) -> None

Register a Pydantic model class under a namespaced name.

Source code in datarobot_genai/drtools/panels/schema_registry.py 
69
70
71
72
73
74
75
@classmethod
def register(cls, name: str, schema: type[BaseModel]) -> None:
    """Register a Pydantic model class under a namespaced ``name``."""
    if not isinstance(schema, type) or not issubclass(schema, BaseModel):
        raise TypeError(f"Schema must be a Pydantic BaseModel class, got {type(schema)}")
    cls._schemas[name] = schema
    logger.debug("Registered schema: %s", name)

unregister classmethod 

unregister(name: str) -> bool

Remove a schema from the registry. Returns True if it existed.

Source code in datarobot_genai/drtools/panels/schema_registry.py 
77
78
79
80
@classmethod
def unregister(cls, name: str) -> bool:
    """Remove a schema from the registry. Returns True if it existed."""
    return cls._schemas.pop(name, None) is not None

get classmethod 

get(name: str) -> type[BaseModel]

Get a schema by name; raises KeyError if not registered.

Source code in datarobot_genai/drtools/panels/schema_registry.py 
82
83
84
85
86
87
88
89
90
@classmethod
def get(cls, name: str) -> type[BaseModel]:
    """Get a schema by name; raises ``KeyError`` if not registered."""
    if name not in cls._schemas:
        available = ", ".join(sorted(cls._schemas.keys()))
        raise KeyError(
            f"Schema '{name}' not found. Available schemas: {available or '(none registered)'}"
        )
    return cls._schemas[name]

exists classmethod 

exists(name: str) -> bool

Check if a schema is registered.

Source code in datarobot_genai/drtools/panels/schema_registry.py 
92
93
94
95
@classmethod
def exists(cls, name: str) -> bool:
    """Check if a schema is registered."""
    return name in cls._schemas

list_schemas classmethod 

list_schemas(namespace: str | None = None) -> dict[str, dict[str, Any]]

List registered schemas (optionally filtered by namespace) with summaries.

Source code in datarobot_genai/drtools/panels/schema_registry.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
@classmethod
def list_schemas(cls, namespace: str | None = None) -> dict[str, dict[str, Any]]:
    """List registered schemas (optionally filtered by ``namespace``) with summaries."""
    result = {}
    for name, schema in sorted(cls._schemas.items()):
        if namespace and not name.startswith(f"{namespace}."):
            continue
        result[name] = {
            "name": name,
            "description": (schema.__doc__ or "").strip().split("\n")[0],
            "required_fields": cls._get_required_fields(schema),
            "optional_fields": cls._get_optional_fields(schema),
        }
    return result

describe classmethod 

describe(name: str) -> dict[str, Any]

Detailed schema description: fields, full JSON Schema, and an example.

Source code in datarobot_genai/drtools/panels/schema_registry.py 
112
113
114
115
116
117
118
119
120
121
122
@classmethod
def describe(cls, name: str) -> dict[str, Any]:
    """Detailed schema description: fields, full JSON Schema, and an example."""
    schema = cls.get(name)
    return {
        "name": name,
        "description": (schema.__doc__ or "").strip(),
        "fields": cls._describe_fields(schema),
        "json_schema": schema.model_json_schema(),
        "example": cls._generate_example(schema),
    }

validate classmethod 

validate(name: str, data: dict[str, Any]) -> dict[str, Any]

Validate data and return it normalized; raises :class:SchemaValidationError.

Source code in datarobot_genai/drtools/panels/schema_registry.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
@classmethod
def validate(cls, name: str, data: dict[str, Any]) -> dict[str, Any]:
    """Validate ``data`` and return it normalized; raises :class:`SchemaValidationError`."""
    schema = cls.get(name)
    try:
        validated = schema.model_validate(data)
        return validated.model_dump(mode="python", exclude_none=True)
    except ValidationError as e:
        raise SchemaValidationError(
            schema_name=name,
            data=data,
            errors=[dict(err) for err in e.errors()],
        ) from e