Commit 526983cd authored by ozzeh's avatar ozzeh
Browse files

Add register_reflected_model instead

parent 4c5d108f
[tool.poetry]
name = "schema-registry"
version = "0.1.2"
version = "0.1.3"
description = ""
authors = ["ozzeh <ozzeh@pleaseignore.com>"]
......
......@@ -105,6 +105,54 @@ class SchemaRegistry:
def get_schema(self, name) -> Schema:
return self._schemas[name]
def _create_schema_for_model(self, schema_name: str, model: Type[BaseModel]):
opts = dict(
Content=model.schema_json(),
RegistryName=self.registry_name,
SchemaName=schema_name,
Type="JSONSchemaDraft4",
)
response = self.schema_client.create_schema(**opts)
schema_info = _SchemaCreateUpdateModel.parse_obj(response)
return schema_info
def _update_schema_for_model(self, schema_name: str, model: Type[BaseModel]):
opts = dict(
Content=model.schema_json(),
RegistryName=self.registry_name,
SchemaName=schema_name,
Type="JSONSchemaDraft4",
)
try:
response = self.schema_client.update_schema(**opts)
schema_info = _SchemaCreateUpdateModel.parse_obj(response)
return schema_info
except self.schema_client.exceptions.ConflictException:
return None
def _schema_exists(self, schema_name: str) -> bool:
schema_name = "{}.{}".format(namespace, model.__name__)
try:
response = self.schema_client.describe_schema(RegistryName=self.registry_name, SchemaName=schema_name)
return True
except self.schema_client.exceptions.NotFoundException:
return False
except:
raise
def register_reflected_model(self, namespace, model: Type[BaseModel]) -> _SchemaCreateUpdateModel:
schema_name = "{}.{}".format(namespace, model.__name__)
exists = self._schema_exists(schema_name)
if not exists:
schema = self._create_schema_for_model(schema_name, model)
else:
schema = self._update_schema_for_model(schema_name, model)
self._model_schemas[model] = schema
return schema
def register_model(
self, namespace, model: Type[BaseModel]
) -> _SchemaCreateUpdateModel:
......@@ -135,6 +183,7 @@ class SchemaRegistry:
self._model_schemas[model] = schema_info
return schema_info
def schema_for_model(self, model: Type[BaseModel]) -> dict:
if model not in self._model_schemas:
raise ModelNotRegisteredError(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