Feature/explicit exception type #47
5
jambo/exceptions/__init__.py
Normal file
5
jambo/exceptions/__init__.py
Normal file
@@ -0,0 +1,5 @@
|
|||||||
|
from .invalid_schema_exception import InvalidSchemaException
|
||||||
|
from .unsupported_schema_exception import UnsupportedSchemaException
|
||||||
|
|
||||||
|
|
||||||
|
__all__ = ["InvalidSchemaException", "UnsupportedSchemaException"]
|
||||||
27
jambo/exceptions/invalid_schema_exception.py
Normal file
27
jambo/exceptions/invalid_schema_exception.py
Normal file
@@ -0,0 +1,27 @@
|
|||||||
|
from typing_extensions import Optional
|
||||||
|
|
||||||
|
|
||||||
|
class InvalidSchemaException(ValueError):
|
||||||
|
"""Exception raised for invalid JSON schemas."""
|
||||||
|
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
message: str,
|
||||||
|
invalid_field: Optional[str] = None,
|
||||||
|
cause: Optional[BaseException] = None,
|
||||||
|
) -> None:
|
||||||
|
# Normalize message by stripping redundant prefix if present
|
||||||
|
message = message.removeprefix("Invalid JSON Schema: ")
|
||||||
|
self.invalid_field = invalid_field
|
||||||
|
self.cause = cause
|
||||||
|
super().__init__(message)
|
||||||
|
|
||||||
|
def __str__(self) -> str:
|
||||||
|
base_msg = f"Invalid JSON Schema: {super().__str__()}"
|
||||||
|
if self.invalid_field:
|
||||||
|
return f"{base_msg} (invalid field: {self.invalid_field})"
|
||||||
|
if self.cause:
|
||||||
|
return (
|
||||||
|
f"{base_msg} (caused by {self.cause.__class__.__name__}: {self.cause})"
|
||||||
|
)
|
||||||
|
return base_msg
|
||||||
23
jambo/exceptions/unsupported_schema_exception.py
Normal file
23
jambo/exceptions/unsupported_schema_exception.py
Normal file
@@ -0,0 +1,23 @@
|
|||||||
|
from typing_extensions import Optional
|
||||||
|
|
||||||
|
|
||||||
|
class UnsupportedSchemaException(ValueError):
|
||||||
|
"""Exception raised for unsupported JSON schemas."""
|
||||||
|
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
message: str,
|
||||||
|
unsupported_field: Optional[str] = None,
|
||||||
|
cause: Optional[BaseException] = None,
|
||||||
|
) -> None:
|
||||||
|
# Normalize message by stripping redundant prefix if present
|
||||||
|
message = message.removeprefix("Unsupported JSON Schema: ")
|
||||||
|
self.unsupported_field = unsupported_field
|
||||||
|
self.cause = cause
|
||||||
|
super().__init__(message)
|
||||||
|
|
||||||
|
def __str__(self) -> str:
|
||||||
|
base_msg = f"Unsupported JSON Schema: {super().__str__()}"
|
||||||
|
if self.unsupported_field:
|
||||||
|
return f"{base_msg} (unsupported field: {self.unsupported_field})"
|
||||||
|
return base_msg
|
||||||
@@ -1,5 +1,6 @@
|
|||||||
|
from jambo.exceptions import InvalidSchemaException, UnsupportedSchemaException
|
||||||
from jambo.parser import ObjectTypeParser, RefTypeParser
|
from jambo.parser import ObjectTypeParser, RefTypeParser
|
||||||
from jambo.types.json_schema_type import JSONSchema
|
from jambo.types import JSONSchema
|
||||||
|
|
||||||
from jsonschema.exceptions import SchemaError
|
from jsonschema.exceptions import SchemaError
|
||||||
from jsonschema.validators import validator_for
|
from jsonschema.validators import validator_for
|
||||||
@@ -26,11 +27,15 @@ class SchemaConverter:
|
|||||||
try:
|
try:
|
||||||
validator = validator_for(schema)
|
validator = validator_for(schema)
|
||||||
validator.check_schema(schema) # type: ignore
|
validator.check_schema(schema) # type: ignore
|
||||||
except SchemaError as e:
|
except SchemaError as err:
|
||||||
raise ValueError(f"Invalid JSON Schema: {e}")
|
raise InvalidSchemaException(
|
||||||
|
"Validation of JSON Schema failed.", cause=err
|
||||||
|
) from err
|
||||||
|
|
||||||
if "title" not in schema:
|
if "title" not in schema:
|
||||||
raise ValueError("JSON Schema must have a title.")
|
raise InvalidSchemaException(
|
||||||
|
"Schema must have a title.", invalid_field="title"
|
||||||
|
)
|
||||||
|
|
||||||
schema_type = SchemaConverter._get_schema_type(schema)
|
schema_type = SchemaConverter._get_schema_type(schema)
|
||||||
|
|
||||||
@@ -55,7 +60,13 @@ class SchemaConverter:
|
|||||||
)
|
)
|
||||||
return parsed_model
|
return parsed_model
|
||||||
case _:
|
case _:
|
||||||
raise TypeError(f"Unsupported schema type: {schema_type}")
|
unsupported_type = (
|
||||||
|
f"type:{schema_type}" if schema_type else "missing type"
|
||||||
|
)
|
||||||
|
raise UnsupportedSchemaException(
|
||||||
|
"Only object and $ref schema types are supported.",
|
||||||
|
unsupported_field=unsupported_type,
|
||||||
|
)
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def _get_schema_type(schema: JSONSchema) -> str | None:
|
def _get_schema_type(schema: JSONSchema) -> str | None:
|
||||||
|
|||||||
@@ -0,0 +1,16 @@
|
|||||||
|
from .json_schema_type import (
|
||||||
|
JSONSchema,
|
||||||
|
JSONSchemaNativeTypes,
|
||||||
|
JSONSchemaType,
|
||||||
|
JSONType,
|
||||||
|
)
|
||||||
|
from .type_parser_options import TypeParserOptions
|
||||||
|
|
||||||
|
|
||||||
|
__all__ = [
|
||||||
|
"JSONSchemaType",
|
||||||
|
"JSONSchemaNativeTypes",
|
||||||
|
"JSONType",
|
||||||
|
"JSONSchema",
|
||||||
|
"TypeParserOptions",
|
||||||
|
]
|
||||||
|
|||||||
Reference in New Issue
Block a user