Commit aa2c5618 authored by ozzeh's avatar ozzeh
Browse files

Now with even less sys.modules hacking

parent 5eaa35fe
from pydantic import BaseModel, create_model, Field
from devtools import debug
from typing import Optional, List
class SchemaReflector:
def __init__(self, schema):
self.schema = schema
self.fields = {}
self.references = {}
def _resolve_reference(self, ref_value, property_info):
accessors = ref_value.rsplit("/", 1)
return self.references[accessors[1]]
def _resolve_array(self, name, property_info, required) -> tuple:
debug(name, property_info, required)
items = property_info.get("items")
item_type, item_value = next(iter(items.items()))
if item_type == "$ref":
referenced_type = self._resolve_reference(item_value, property_info)
return (referenced_type, ... if required else None)
def _resolve_property(self, name, property_info, required=True) -> tuple:
type_ = property_info.get("type")
if type_ == "string":
return (str, ... if required else None)
elif type_ == "boolean":
return (bool, ... if required else None)
elif type_ in ("number", "integer"):
return (int, ... if required else None)
elif type_ == "array":
return self._resolve_array(name, property_info, required)
raise NotImplementedError(f"Not able to reflect the type: {type_}")
def _resolve_properties(self):
if "properties" not in self.schema:
raise TypeError("No properties are defined for this schema")
required_fields = self.schema.get("required")
for name, info in self.schema.get("properties").items():
self.fields[name] = self._resolve_property(
name, info, required=name in required_fields
)
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")
model = SchemaReflector(d_info).create_model_for_jsonschema()
self.references[d_name] = model
def create_model_for_jsonschema(self):
if "title" not in self.schema:
raise TypeError("Schema needs a title field")
if "type" not in self.schema:
raise TypeError("Schema needs a type field")
self.root_model_name = self.schema.get("title")
if "definitions" in self.schema:
for name, d_info in self.schema.get("definitions").items():
self._resolve_definition(name, d_info)
self._resolve_properties()
return create_model(self.root_model_name, **self.fields)
from typing import Optional
import pytest
import boto3
from typing import List
from schema_registry import SchemaRegistry
from pydantic import BaseModel
import logging
from devtools import debug
logging.basicConfig()
log = logging.getLogger("schema_registry")
log.setLevel(logging.WARNING)
log.propagate = True
@pytest.fixture(scope="module")
def named_registry():
_registry = SchemaRegistry(registry_name="TAPI-TEST")
yield _registry
@pytest.fixture(scope="module")
def simple_model():
item = {
"title": "TestingModel",
"type": "object",
"properties": {
"name": {"title": "Name", "type": "string"},
"description": {"title": "Description", "type": "string"},
},
"required": ["name"],
}
return item
@pytest.fixture(scope="module")
def complex_model():
item = {
"title": "ComplexModel",
"description": "Hi mom",
"type": "object",
"properties": {
"name": {"title": "Name", "type": "string"},
"description": {"title": "Description", "type": "string"},
"groups": {
"title": "Groups",
"type": "array",
"items": {"$ref": "#/definitions/Group"},
},
},
"required": ["name", "groups"],
"definitions": {
"Group": {
"title": "Group",
"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
model = SchemaReflector(simple_model).create_model_for_jsonschema()
debug(model.__fields__)
named_registry.register_model("com.pleaseignore.tvm.test", model)
yield model
@pytest.fixture(scope="module")
def reflected_complex_model(complex_model, named_registry):
from schema_registry.reflection import SchemaReflector
model = SchemaReflector(complex_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
def test_complex_reflection(reflected_complex_model):
pass
def test_simple_reflection_registration(reflected_simple_model, named_registry):
debug(named_registry.schema_for_model(reflected_simple_model))
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