Commit 12536f6e authored by ozzeh's avatar ozzeh
Browse files

Blacken

parent 064cc126
schema_registry.egg-info/
.python-version
\ No newline at end of file
.python-version
.idea/
.vscode/
\ No newline at end of file
......@@ -3,7 +3,6 @@ import sys
import logging
import json
from typing import List, Optional, Dict, Type
from datetime import datetime
......@@ -70,7 +69,6 @@ class Schema:
else:
return self._versions[version]
def __repr__(self):
return f"Schema<{self.schema_name}, versions: {len(self._versions)}, default version: {self.default_version}>"
......@@ -107,7 +105,6 @@ class SchemaRegistry:
def get_schema(self, name) -> Schema:
return self._schemas[name]
def register_model(
self, namespace, model: Type[BaseModel]
) -> _SchemaCreateUpdateModel:
......@@ -144,9 +141,8 @@ class SchemaRegistry:
raise ModelNotRegisteredError(model)
return self._model_schemas[model].dict(
include=set(["schema_arn", "schema_name", "schema_version"])
include={"schema_arn", "schema_name", "schema_version"}
)
def send_event(
self, event_bus, sender, model: BaseModel, extra_resources: List[str] = None
......@@ -166,7 +162,7 @@ class SchemaRegistry:
event_data = {
"schema": schema_info.dict(
include=set(["schema_arn", "schema_name", "schema_version"])
include={"schema_arn", "schema_name", "schema_version"}
),
"event": model.dict(),
}
......
......@@ -66,7 +66,7 @@ class Event(BaseModel):
detail_type: str = Field(..., alias="detail-type")
detail: dict
source: str
account: str
region: str
......@@ -83,4 +83,4 @@ class Event(BaseModel):
@property
def schema_name(self) -> str:
return self.detail_type.split("/")[1].split(":")[0]
\ No newline at end of file
return self.detail_type.split("/")[1].split(":")[0]
......@@ -25,13 +25,11 @@ class _ReflectedModel(BaseModel):
@property
def detail_type(self) -> Optional[str]:
return self.__detail_type__ or None
class Config:
alias_generator = lambda x: "fields_" if x == "fields" else x
class SchemaReflector:
def __init__(self, schema, registry=None):
self.schema = schema
......
......@@ -8,4 +8,4 @@ def test_loading_schemas():
registry = SchemaRegistry("TAPI-TEST")
schema: Schema = registry._schemas["schema_registry.test.TestingModel"]
debug(schema.get())
reflector: SchemaReflector = SchemaReflector(schema.get().content_dict)
\ No newline at end of file
reflector: SchemaReflector = SchemaReflector(schema.get().content_dict)
......@@ -148,10 +148,11 @@ def test_complex_referenced_model_reflection(reflected_complex_referenced_model)
def test_simple_reflection_registration(reflected_simple_model, named_registry):
named_registry.schema_for_model(reflected_simple_model)
def test_event_reflection():
event_json = '{"version": "0", "id": "d944d595-b186-4b86-43fe-b096d7e13bb3", "detail-type": "TAPI-TEST/schema_registry.test.TestingModel:1", "source": "com.pleaseignore.tvm.test", "account": "740218546536", "time": "2020-11-27T16:53:00Z", "region": "eu-west-1", "resources": ["pydantic-schema-registry"], "detail": {"name": "ozzeh", "description": "big willy johnston"}}'
event = Event.parse_raw(event_json)
model = reflect_event(json.loads(event_json))
debug(model)
debug(model.__fields__)
debug(model.__detail_type__)
\ No newline at end of file
debug(model.__detail_type__)
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