Basic support of references Karapace API calls.

.gitignore

@@ -16,3 +16,4 @@ __pycache__/
 /kafka_*/
 venv
 /karapace/version.py
+.run

karapace/schema_models.py

@@ -12,8 +12,9 @@
     SchemaParseException as ProtobufSchemaParseException,
 )
 from karapace.protobuf.schema import ProtobufSchema
+from karapace.typing import JsonData
 from karapace.utils import json_encode
-from typing import Any, Dict, Union
+from typing import Any, Dict, Optional, Union
 
 import json
 
@@ -67,6 +68,10 @@ class InvalidSchema(Exception):
     pass
 
 
+class InvalidReferences(Exception):
+    pass
+
+
 @unique
 class SchemaType(str, Enum):
     AVRO = "AVRO"
@@ -75,7 +80,7 @@ class SchemaType(str, Enum):
 
 
 class TypedSchema:
-    def __init__(self, schema_type: SchemaType, schema_str: str):
+    def __init__(self, schema_type: SchemaType, schema_str: str, references: Optional["References"] = None):
         """Schema with type information
 
         Args:
@@ -84,6 +89,7 @@ def __init__(self, schema_type: SchemaType, schema_str: str):
         """
         self.schema_type = schema_type
         self.schema_str = schema_str
+        self.references = references
 
     def to_dict(self) -> Dict[str, Any]:
         if self.schema_type is SchemaType.PROTOBUF:
@@ -100,17 +106,42 @@ def __repr__(self) -> str:
             return f"TypedSchema(type={self.schema_type}, schema={str(self)})"
         return f"TypedSchema(type={self.schema_type}, schema={json_encode(self.to_dict())})"
 
+    def get_references(self) -> Optional["References"]:
+        return self.references
+
     def __eq__(self, other: Any) -> bool:
-        return isinstance(other, TypedSchema) and self.__str__() == other.__str__() and self.schema_type is other.schema_type
+        a = isinstance(other, TypedSchema) and self.schema_type is other.schema_type and self.__str__() == other.__str__()
+
+        if not a:
+            return False
+        x = self.get_references()
+        y = other.get_references()
+        if x:
+            if y:
+                if x == y:
+                    return True
+            else:
+                return False
+        else:
+            if y:
+                return False
+
+        return True
 
 
 class ValidatedTypedSchema(TypedSchema):
-    def __init__(self, schema_type: SchemaType, schema_str: str, schema: Union[Draft7Validator, AvroSchema, ProtobufSchema]):
-        super().__init__(schema_type=schema_type, schema_str=schema_str)
+    def __init__(
+        self,
+        schema_type: SchemaType,
+        schema_str: str,
+        schema: Union[Draft7Validator, AvroSchema, ProtobufSchema],
+        references: Optional["References"] = None,
+    ):
+        super().__init__(schema_type=schema_type, schema_str=schema_str, references=references)
         self.schema = schema
 
     @staticmethod
-    def parse(schema_type: SchemaType, schema_str: str) -> "ValidatedTypedSchema":
+    def parse(schema_type: SchemaType, schema_str: str, references: Optional["References"] = None) -> "ValidatedTypedSchema":
         if schema_type not in [SchemaType.AVRO, SchemaType.JSONSCHEMA, SchemaType.PROTOBUF]:
             raise InvalidSchema(f"Unknown parser {schema_type} for {schema_str}")
 
@@ -146,9 +177,32 @@ def parse(schema_type: SchemaType, schema_str: str) -> "ValidatedTypedSchema":
         else:
             raise InvalidSchema(f"Unknown parser {schema_type} for {schema_str}")
 
-        return ValidatedTypedSchema(schema_type=schema_type, schema_str=schema_str, schema=parsed_schema)
+        return ValidatedTypedSchema(
+            schema_type=schema_type, schema_str=schema_str, schema=parsed_schema, references=references
+        )
 
     def __str__(self) -> str:
         if self.schema_type == SchemaType.PROTOBUF:
             return str(self.schema)
         return super().__str__()
+
+
+class References:
+    def __init__(self, schema_type: SchemaType, references: JsonData):
+        """Schema with type information
+
+        Args:
+            schema_type (SchemaType): The type of the schema
+            references (str): The original schema string
+        """
+        self.schema_type = schema_type
+        self.references = references
+
+    def val(self) -> JsonData:
+        return self.references
+
+    def json(self) -> str:
+        return str(json_encode(self.references, sort_keys=True))
+
+    def __eq__(self, other: Any) -> bool:
+        return self.json() == other.json()

karapace/schema_reader.py

@@ -15,7 +15,7 @@
 from karapace.statsd import StatsClient
 from karapace.utils import KarapaceKafkaClient
 from threading import Event, Lock, Thread
-from typing import Any, Dict, Optional
+from typing import Any, Dict, List, Optional
 
 import json
 import logging
@@ -26,6 +26,7 @@
 Schema = Dict[str, Any]
 # Container type for a subject, with configuration settings and all the schemas
 SubjectData = Dict[str, Any]
+Referents = List
 
 # The value `0` is a valid offset and it represents the first message produced
 # to a topic, therefore it can not be used.
@@ -139,6 +140,7 @@ def __init__(
         self.config = config
         self.subjects: Dict[Subject, SubjectData] = {}
         self.schemas: Dict[int, TypedSchema] = {}
+        self.referenced_by: Dict[str, Referents] = {}
         self.global_schema_id = 0
         self.admin_client: Optional[KafkaAdminClient] = None
         self.topic_replication_factor = self.config["replication_factor"]
@@ -291,6 +293,7 @@ def _handle_msg_config(self, key: dict, value: Optional[dict]) -> None:
             else:
                 LOG.info("Setting subject: %r config to: %r, value: %r", subject, value["compatibilityLevel"], value)
                 self.subjects[subject]["compatibility"] = value["compatibilityLevel"]
+
         elif value is not None:
             LOG.info("Setting global config to: %r, value: %r", value["compatibilityLevel"], value)
             self.config["compatibility"] = value["compatibilityLevel"]
@@ -334,6 +337,7 @@ def _handle_msg_schema(self, key: dict, value: Optional[dict]) -> None:
         schema_version = value["version"]
         schema_deleted = value.get("deleted", False)
 
+        schema_references = value.get("references", None)
         try:
             schema_type_parsed = SchemaType(schema_type)
         except ValueError:
@@ -366,20 +370,33 @@ def _handle_msg_schema(self, key: dict, value: Optional[dict]) -> None:
             typed_schema = TypedSchema(
                 schema_type=schema_type_parsed,
                 schema_str=schema_str,
+                references=schema_references,
             )
             schema = {
                 "schema": typed_schema,
                 "version": schema_version,
                 "id": schema_id,
                 "deleted": schema_deleted,
             }
+            if schema_references:
+                schema["references"] = schema_references
 
             if schema_version in subjects_schemas:
                 LOG.info("Updating entry subject: %r version: %r id: %r", schema_subject, schema_version, schema_id)
             else:
                 LOG.info("Adding entry subject: %r version: %r id: %r", schema_subject, schema_version, schema_id)
 
             subjects_schemas[schema_version] = schema
+            if schema_references:
+                for ref in schema_references:
+                    ref_str = str(ref["subject"]) + "_" + str(ref["version"])
+                    referents = self.referenced_by.get(ref_str, None)
+                    if referents:
+                        LOG.info("Adding entry subject referenced_by : %r", ref_str)
+                        referents.append(schema_id)
+                    else:
+                        LOG.info("Adding entry subject referenced_by : %r", ref_str)
+                        self.referenced_by[ref_str] = [schema_id]
 
             subjects_schemas[schema_version] = {
                 "schema": typed_schema,