Commit 2bedcf87 authored by ozzeh's avatar ozzeh
Browse files

Add object reference support

parent b39385af
schema_registry.egg-info/
\ No newline at end of file
schema_registry.egg-info/
.python-version
\ No newline at end of file
PRIMITIVE_TYPES = {
"array": list,
"boolean": bool,
......@@ -7,5 +5,5 @@ PRIMITIVE_TYPES = {
"number": int,
"null": None,
"object": type,
"string": str
}
\ No newline at end of file
"string": str,
}
......@@ -8,6 +8,7 @@ class _ReflectedModel(BaseModel):
class Config:
alias_generator = lambda x: "fields_" if x == "fields" else x
class SchemaReflector:
def __init__(self, schema):
self.schema = schema
......@@ -18,35 +19,43 @@ class SchemaReflector:
def _resolve_reference(self, ref_value):
if ref_value.startswith("#"):
return resolve_pointer(self.__dict__, ref_value.split("#")[1])
raise TypeError("Cannot resolve a reference that's not a json pointer. (Reference value {})".format(ref_value))
def _resolve_array(self, name, property_info, required) -> tuple:
raise TypeError(
"Cannot resolve a reference that's not a json pointer. (Reference value {})".format(
ref_value
)
)
def _resolve_array(self, property_info, required) -> tuple:
items = property_info.get("items")
item_type, item_value = next(iter(items.items()))
if item_type == "$ref":
referenced_type = self._resolve_reference(item_value)
return (referenced_type, ... if required else None)
return (referenced_type if required else Optional[referenced_type], ... if required else None)
raise NotImplementedError("Cannot reflect type: {}".format(item_type))
def _resolve_property(self, name, property_info, required=True) -> tuple:
type_ = property_info.get("type")
if "type" in property_info:
type_ = property_info.get("type")
if type_ == "string":
return (str, ... if required else None)
if type_ == "string":
return (str if required else Optional[bool], ... if required else None)
elif type_ == "boolean":
return (bool, ... if required else None)
elif type_ == "boolean":
return (bool if required else Optional[bool], ... if required else None)
elif type_ in ("number", "integer"):
return (int, ... if required else None)
elif type_ in ("number", "integer"):
return (int if required else Optional[int], ... if required else None)
elif type_ == "array":
return self._resolve_array(name, property_info, required)
elif type_ == "array":
return self._resolve_array(property_info, required)
else:
if "$ref" in property_info:
return (self._resolve_reference(property_info["$ref"]), ... if required else None)
raise NotImplementedError(f"Not able to reflect the type: {type_}")
raise NotImplementedError(f"Not able to reflect the type: {property_info}")
def _resolve_properties(self):
if "properties" not in self.schema:
......@@ -60,7 +69,6 @@ class SchemaReflector:
)
def _resolve_definition(self, d_name, d_info):
debug(d_name, d_info)
if d_info["type"] != "object":
raise TypeError("Cannot reflect a non-object")
......@@ -83,4 +91,6 @@ class SchemaReflector:
self._resolve_properties()
return create_model(self.root_model_name, __base__=_ReflectedModel, **self.fields)
return create_model(
self.root_model_name, __base__=_ReflectedModel, **self.fields
)
......@@ -68,6 +68,33 @@ def complex_model():
yield item
@pytest.fixture(scope="module")
def complex_referenced_model():
item = {
"title": "ComplexReferencedModel",
"description": "Hi mom",
"type": "object",
"properties": {
"name": {"title": "Name", "type": "string"},
"description": {"title": "Description", "type": "string"},
"group": {"$ref": "#/definitions/ReferencedGroup"},
},
"required": ["name", "group"],
"definitions": {
"ReferencedGroup": {
"title": "ReferencedGroup",
"type": "object",
"properties": {
"id": {"title": "Id", "type": "integer"},
"name": {"title": "Name", "type": "string"},
},
"required": ["id", "name"],
}
},
}
yield item
@pytest.fixture(scope="module")
def reflected_simple_model(simple_model, named_registry):
from schema_registry.reflection import SchemaReflector
......@@ -88,6 +115,16 @@ def reflected_complex_model(complex_model, named_registry):
yield model
@pytest.fixture(scope="module")
def reflected_complex_referenced_model(complex_referenced_model, named_registry):
from schema_registry.reflection import SchemaReflector
model = SchemaReflector(complex_referenced_model).create_model_for_jsonschema()
debug(model.__fields__)
named_registry.register_model("com.pleaseignore.tvm.test", model)
yield model
def test_simple_reflection(reflected_simple_model):
pass
......@@ -95,6 +132,10 @@ def test_simple_reflection(reflected_simple_model):
def test_complex_reflection(reflected_complex_model):
pass
def test_complex_referenced_model_reflection(reflected_complex_referenced_model):
pass
def test_simple_reflection_registration(reflected_simple_model, named_registry):
debug(named_registry.schema_for_model(reflected_simple_model))
......@@ -61,6 +61,22 @@ def complex_model():
yield ComplexModel
@pytest.fixture(scope="module")
def complex_model_with_references():
class ReferencedGroup(BaseModel):
id: int
name: str
class ComplexReferencedModel(BaseModel):
"""Hi mom"""
name: str
description: Optional[str]
group: ReferencedGroup
yield ComplexReferencedModel
def test_load_named_schemas(test_model, named_registry):
pass
......@@ -76,9 +92,17 @@ def test_registered_model(test_model, named_registry):
def test_complex_model(complex_model, named_registry):
named_registry.register_model("com.pleaseignore.tvm.test", complex_model)
def test_nullable_model(nullable_model, named_registry):
named_registry.register_model("com.pleaseignore.tvm.test", nullable_model)
def test_complex_model_with_references(complex_model_with_references, named_registry):
named_registry.register_model(
"com.pleaseignore.tvm.test", complex_model_with_references
)
def test_send_simple_message(test_model, named_registry):
instance = test_model(name="Test", description="Hello!")
log.debug(instance)
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment