Added registry management functions. Schema recovery working (#6)

Author: JohnPreston

kafka_schema_registry_admin/__init__.py

@@ -7,6 +7,6 @@
 __email__ = "john@ews-network.net"
 __version__ = "0.4.0"
 
-from .kafka_schema_registry_admin import SchemaRegistry
+from .kafka_schema_registry_admin import CompatibilityMode, RegistryMode, SchemaRegistry
 
-__all__ = ["SchemaRegistry"]
+__all__ = ["SchemaRegistry", "RegistryMode", "CompatibilityMode"]

kafka_schema_registry_admin/client_wrapper/errors.py

@@ -44,8 +44,9 @@ def __init__(self, code, details):
                 if error_message.startswith(
                     "Schema being registered is incompatible with an earlier schema for subject"
                 ):
-                    raise IncompatibleSchema(code, details)
-            super().__init__(details, code, details[1:])
+                    raise IncompatibleSchema(code, details[1:])
+            else:
+                super().__init__(details, code, details[1:])
 
 
 class IncompatibleSchema(ApiGenericException):
@@ -84,6 +85,16 @@ def __init__(self, code, details):
         super().__init__("Forbidden", code, details)
 
 
+class UnprocessableEntity(ApiGenericException):
+    def __init__(self, code, details):
+        super().__init__("Unprocessable Entity", code, details)
+
+
+class UnexpectedException(ApiGenericException):
+    def __init__(self, code, details):
+        super().__init__("Unexpected Error", code, details)
+
+
 class SchemaRegistryApiException(ApiGenericException):
     """
     Top class for DatabaseUser exceptions
@@ -94,11 +105,18 @@ class SchemaRegistryApiException(ApiGenericException):
         404: NotFoundException,
         401: UnauthorizedException,
         403: ForbiddenException,
+        422: UnprocessableEntity,
     }
 
     def __init__(self, code, details):
-        exception_class = self.EXCEPTION_CLASSES.get(code, ApiGenericException)
-        super().__init__("Api Error", code, details)
+        exception_class = self.EXCEPTION_CLASSES.get(code, UnexpectedException)
+        super().__init__(
+            details[0] is isinstance(details[0], str)
+            and details[0]
+            or "SchemaRegistry Api Error",
+            code,
+            details,
+        )
         self.exception_instance = exception_class(code, details)
 
 

kafka_schema_registry_admin/kafka_schema_registry_admin.py

@@ -30,6 +30,22 @@ class Type(Enum):
     PROTOBUFF = "PROTOBUF"
 
 
+class RegistryMode(Enum):
+    IMPORT = "IMPORT"
+    READONLY = "READONLY"
+    READWRITE = "READWRITE"
+
+
+class CompatibilityMode(Enum):
+    BACKWARD = "BACKWARD"
+    BACKWARD_TRANSITIVE = "BACKWARD_TRANSITIVE"
+    FORWARD = "FORWARD"
+    FORWARD_TRANSITIVE = "FORWARD_TRANSITIVE"
+    FULL = "FULL"
+    FULL_TRANSITIVE = "FULL_TRANSITIVE"
+    NONE = "NONE"
+
+
 class SchemaRegistry:
 
     def __init__(self, base_url: str, *args, **kwargs):
@@ -133,7 +149,13 @@ def post_subject_schema(
         return req
 
     def post_subject_schema_version(
-        self, subject_name, definition, normalize: bool = False, schema_type=None
+        self,
+        subject_name,
+        definition,
+        normalize: bool = False,
+        schema_type=None,
+        version_id: int = None,
+        schema_id: int = None,
     ) -> Response:
         """
         `API Doc <https://docs.confluent.io/platform/current/schema-registry/develop/api.html#post--subjects-(string-%20subject)-versions>`__
@@ -154,6 +176,11 @@ def post_subject_schema_version(
             url = f"/subjects/{subject_name}/versions"
             if normalize:
                 url = f"{url}?normalize=true"
+            """When trying to do recovery, SR must be in import mode either globally or for the subject itself."""
+            if version_id and schema_id:
+                payload["version"] = version_id
+                payload["id"] = schema_id
+
             req = self.client.post(url, json=payload)
             return req
 
@@ -215,9 +242,10 @@ def post_compatibility_subject_versions(
         references: list = None,
     ) -> Response:
         url = f"/compatibility/subjects/{subject_name}/versions"
-        return self.validate_subject_compatibility(
+        payload = self.set_subject_validity_payload(
             url, definition, schema_type, verbose=verbose, references=references
         )
+        return self.client.post(url, json=payload)
 
     def post_compatibility_subject_version_id(
         self,
@@ -229,18 +257,19 @@ def post_compatibility_subject_version_id(
         references: list = None,
     ) -> Response:
         url = f"/compatibility/subjects/{subject_name}/versions/{version_id}"
-        return self.validate_subject_compatibility(
+        payload = self.set_subject_validity_payload(
             url, definition, schema_type, verbose=verbose, references=references
         )
+        return self.client.post(url, json=payload)
 
-    def validate_subject_compatibility(
-        self,
+    @staticmethod
+    def set_subject_validity_payload(
         url: str,
         definition,
         schema_type,
         verbose: bool = False,
         references: list = None,
-    ) -> Response:
+    ) -> dict:
         if verbose:
             url = f"{url}?verbose=true"
         LOG.debug(url)
@@ -254,9 +283,7 @@ def validate_subject_compatibility(
         payload = {"schema": definition, "schemaType": schema_type}
         if references and isinstance(references, list):
             payload["references"] = references
-
-        req = self.client.post(url, json=payload)
-        return req
+        return payload
 
     def get_compatibility_subject_config(self, subject_name) -> Response:
         url = f"/config/{subject_name}/"
@@ -268,3 +295,130 @@ def put_compatibility_subject_config(self, subject_name, compatibility) -> Respo
         payload = {"compatibility": compatibility}
         req = self.client.put(url, json=payload)
         return req
+
+    def get_mode(self, as_str: bool = False) -> Response | str:
+        """
+        `API Doc <https://docs.confluent.io/platform/current/schema-registry/develop/api.html#mode>`__
+        """
+        url_path: str = "/mode"
+        req = self.client.get(url_path)
+        if as_str:
+            return RegistryMode[req.json().get("mode")].value
+        return req
+
+    def put_mode(self, mode: str | RegistryMode, force: bool = False) -> Response:
+        """
+        `API Doc <https://docs.confluent.io/platform/current/schema-registry/develop/api.html#put--mode>`__
+        """
+        url_path: str = "/mode"
+        if force:
+            url_path += "?force=true"
+        req = self.client.put(url_path, json={"mode": mode})
+        return req
+
+    def get_subject_mode(self, subject_name: str) -> Response:
+        """
+        `API Doc <https://docs.confluent.io/platform/current/schema-registry/develop/api.html#get--mode-(string-%20subject)>`__
+        """
+        url_path: str = f"/mode/{subject_name}"
+        req = self.client.get(url_path)
+        return req
+
+    def put_subject_mode(
+        self, subject_name: str, mode: str | RegistryMode, force: bool = False
+    ) -> Response:
+        """
+        `API Doc <https://docs.confluent.io/platform/current/schema-registry/develop/api.html#put--mode-(string-%20subject)>`__
+        """
+        url_path: str = f"/mode/{subject_name}"
+        if force:
+            url_path += "?force=true"
+        req = self.client.put(url_path, json={"mode": mode})
+        return req
+
+    def get_config(self):
+        """
+        `API Doc <https://docs.confluent.io/platform/current/schema-registry/develop/api.html#get--config>`__
+        """
+        url_path: str = "/config"
+        req = self.client.get(url_path)
+        return req
+
+    def put_config(
+        self,
+        alias: str = None,
+        normalize: bool = False,
+        compatibility: str | CompatibilityMode = "NONE",
+    ):
+        """
+        `API Doc <https://docs.confluent.io/platform/current/schema-registry/develop/api.html#put--config>`__
+        """
+        url_path: str = "/config"
+        payload: dict = {}
+        if compatibility:
+            payload["compatibility"] = compatibility
+        if alias:
+            payload["alias"] = alias
+        if normalize:
+            payload["normalize"] = normalize
+        req = self.client.put(url_path, json=payload)
+        return req
+
+    def get_subject_config(
+        self,
+        subject: str,
+        default: bool = False,
+        alias: str = None,
+        normalize: bool = False,
+        compatibility: str | CompatibilityMode = "NONE",
+    ):
+        """
+        `API Doc <https://docs.confluent.io/platform/current/schema-registry/develop/api.html#get--config>`__
+        """
+        url_path: str = f"/config/{subject}"
+        if default:
+            url_path += "?defaultToGlobal=true"
+        payload: dict = {}
+        if compatibility:
+            payload["compatibility"] = compatibility
+        if alias:
+            payload["alias"] = alias
+        if normalize:
+            payload["normalize"] = normalize
+        req = self.client.get(url_path, json=payload)
+        return req
+
+    def put_subject_config(
+        self,
+        subject: str,
+        alias: str = None,
+        normalize: bool = False,
+        compatibility: str | CompatibilityMode = "NONE",
+    ):
+        """
+        `API Doc <https://docs.confluent.io/platform/current/schema-registry/develop/api.html#put--config>`__
+        """
+        url_path: str = f"/config/{subject}"
+        payload: dict = {}
+        if compatibility:
+            payload["compatibility"] = compatibility
+        if alias:
+            payload["alias"] = alias
+        if normalize:
+            payload["normalize"] = normalize
+        req = self.client.put(url_path, json=payload)
+        return req
+
+    def delete_subject_config(
+        self,
+        subject: str,
+        alias: str = None,
+        normalize: bool = False,
+        compatibility: str | CompatibilityMode = "NONE",
+    ):
+        """
+        `API Doc <https://docs.confluent.io/platform/current/schema-registry/develop/api.html#delete--config-(string-%20subject)>`__
+        """
+        url_path: str = f"/config/{subject}"
+        req = self.client.delete(url_path)
+        return req

tests/conftest.py

@@ -2,16 +2,42 @@
 
 from os import path
 
+import pytest
 from testcontainers.compose import DockerCompose
 
+from kafka_schema_registry_admin import SchemaRegistry
+
 HERE = path.abspath(path.dirname(__file__))
 
-compose = DockerCompose(
-    path.abspath(f"{HERE}/.."), compose_file_name="docker-compose.yaml", wait=True
+
+docker_compose = DockerCompose(
+    path.abspath(f"{HERE}/.."),
+    compose_file_name="docker-compose.yaml",
+    wait=True,
+    pull=True,
 )
 
+docker_compose.stop(down=True)
+docker_compose.start()
+sr_port = int(docker_compose.get_service_port("schema-registry", 8081))
+base_url: str = f"http://localhost:{sr_port}"
+docker_compose.wait_for(f"{base_url}/subjects")
+
+
+@pytest.fixture(scope="session")
+def local_registry():
+    return SchemaRegistry(base_url)
+
+
+@pytest.fixture(scope="session")
+def authed_local_registry():
+    return SchemaRegistry(
+        base_url,
+        **{"basic_auth.username": "confluent", "basic_auth.password": "confluent"},
+    )
+
 
 def pytest_sessionfinish(session, exitstatus):
-    # compose.stop()
+    docker_compose.stop()
     print("Testing session has finished")
     print(f"Exit status: {exitstatus}")

tests/test_registry_config.py

@@ -0,0 +1,32 @@
+#  SPDX-License-Identifier: Apache License 2.0
+#  Copyright 2020-2021 John Mille <john@ews-network.net>
+import json
+from copy import deepcopy
+from os import path
+
+import pytest
+
+from kafka_schema_registry_admin import CompatibilityMode, RegistryMode, SchemaRegistry
+
+
+def test_changing_compatibility(authed_local_registry):
+    config = authed_local_registry.get_config().json()
+    print("CONFIG 1?", config)
+    authed_local_registry.put_config(
+        compatibility=CompatibilityMode.FULL.value, normalize=True
+    )
+    config = authed_local_registry.get_config().json()
+    print("CONFIG 2?", config)
+
+
+def test_changing_mode(authed_local_registry):
+    mode = authed_local_registry.get_mode(as_str=True)
+    assert mode == RegistryMode.READWRITE.value
+
+    authed_local_registry.put_mode(RegistryMode.READONLY.value)
+    assert (
+        authed_local_registry.get_mode().json()["mode"] == RegistryMode.READONLY.value
+    )
+
+    authed_local_registry.put_mode(RegistryMode.IMPORT.value)
+    assert authed_local_registry.get_mode().json()["mode"] == RegistryMode.IMPORT.value