Rework client wrapper and specific exceptions (#1)

Author: JohnPreston

.pre-commit-config.yaml

@@ -1,6 +1,6 @@
 repos:
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v4.4.0
+    rev: v4.5.0
     hooks:
       - id: end-of-file-fixer
       - id: trailing-whitespace
@@ -16,18 +16,18 @@ repos:
       - id: fix-byte-order-marker
 
   - repo: https://github.com/asottile/pyupgrade
-    rev: v3.10.1
+    rev: v3.15.1
     hooks:
       - id: pyupgrade
         args: [ "--py38-plus", "--keep-runtime-typing" ]
 
   - repo: https://github.com/psf/black
-    rev: 23.7.0
+    rev: 24.2.0
     hooks:
       - id: black
 
   - repo: https://github.com/asottile/blacken-docs
-    rev: 1.15.0
+    rev: 1.16.0
     hooks:
       - id: blacken-docs
         additional_dependencies: [ black ]
@@ -37,7 +37,7 @@ repos:
     hooks:
       - id: seed-isort-config
   - repo: https://github.com/pycqa/isort
-    rev: 5.12.0
+    rev: 5.13.2
     hooks:
       - id: isort
         args: [ "--profile", "black", "--filter-files" ]

LICENSE

@@ -2,15 +2,22 @@
 Kafka schema registry admin
 ===========================
 
-Pure HTTP client library (using requests) to manipulate schemas and definitions into Schema Registry"
+Simple / light HTTP client library (using requests) to manipulate schemas and definitions into Schema Registry.
 
-API specification is documented `here <https://docs.confluent.io/platform/current/schema-registry/develop/api.html#overview>`__
+* Confluent API specification is documented `here <https://docs.confluent.io/platform/current/schema-registry/develop/api.html#overview>`__
 
+* RedPanda API specification is documented `here <https://docs.redpanda.com/current/manage/schema-reg/schema-reg-api/>`__
 
-Credits
--------
 
-This package was created with Cookiecutter_ and the `audreyr/cookiecutter-pypackage`_ project template.
+Usage
+======
 
-.. _Cookiecutter: https://github.com/audreyr/cookiecutter
-.. _`audreyr/cookiecutter-pypackage`: https://github.com/audreyr/cookiecutter-pypackage
+Very simple example to manipulate the schema registry and its resources.
+
+.. code-block::
+
+    from kafka_schema_registry_admin import SchemaRegistry
+
+    registry = SchemaRegistry("http://localhost:8081")
+    subjects = registry.get_all_subjects()
+    schemas = registry.get_all_schemas()

README.rst

@@ -0,0 +1,100 @@
+---
+version: '2'
+services:
+  broker:
+    image: apache/kafka:3.7.0
+    hostname: broker
+    container_name: broker
+    ports:
+      - "9092:9092"
+    environment:
+      KAFKA_NODE_ID: 1
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT'
+      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT_HOST://localhost:9092,PLAINTEXT://broker:19092'
+      KAFKA_PROCESS_ROLES: 'broker,controller'
+      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@broker:29093'
+      KAFKA_LISTENERS: 'CONTROLLER://:29093,PLAINTEXT_HOST://:9092,PLAINTEXT://:19092'
+      KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
+      KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
+      CLUSTER_ID: '4L6g3nShT-eMCtK--X86sw'
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+      KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs'
+
+  schema-registry:
+    image: confluentinc/cp-schema-registry:7.5.0
+    hostname: schema-registry
+    container_name: schema-registry
+    depends_on:
+      - broker
+    ports:
+      - "8081:8081"
+    environment:
+      SCHEMA_REGISTRY_HOST_NAME: schema-registry
+      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: 'broker:19092'
+      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081
+
+  connect:
+    image: cricketeerone/apache-kafka-connect
+    hostname: connect
+    container_name: connect
+    depends_on:
+      - broker
+      - schema-registry
+    ports:
+      - "8083:8083"
+    environment:
+      CONNECT_BOOTSTRAP_SERVERS: 'broker:19092'
+      CONNECT_REST_ADVERTISED_HOST_NAME: connect
+      CONNECT_GROUP_ID: compose-connect-group
+      CONNECT_CONFIG_STORAGE_TOPIC: docker-connect-configs
+      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
+      CONNECT_OFFSET_FLUSH_INTERVAL_MS: 10000
+      CONNECT_OFFSET_STORAGE_TOPIC: docker-connect-offsets
+      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
+      CONNECT_STATUS_STORAGE_TOPIC: docker-connect-status
+      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
+      CONNECT_KEY_CONVERTER: org.apache.kafka.connect.storage.StringConverter
+      CONNECT_VALUE_CONVERTER: org.apache.kafka.connect.storage.StringConverter
+      CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: http://schema-registry:8081
+      CONNECT_PLUGIN_PATH: "/usr/share/java,/usr/share/confluent-hub-components"
+      CONNECT_LOG4J_LOGGERS: org.reflections=ERROR
+
+  postgresql:
+    image: postgres:14
+    hostname: postgresql
+    volumes:
+      - pg_data:/var/lib/postgresql/data
+    environment:
+      POSTGRES_DB: "conduktor-console"
+      POSTGRES_USER: "conduktor"
+      POSTGRES_PASSWORD: "change_me"
+      POSTGRES_HOST_AUTH_METHOD: "scram-sha-256"
+
+  conduktor-console:
+    image: conduktor/conduktor-console:1.21.1
+    depends_on:
+      - postgresql
+      - broker
+    ports:
+      - "8080:8080"
+    volumes:
+      - conduktor_data:/var/conduktor
+    environment:
+      CDK_DATABASE_URL: "postgresql://conduktor:change_me@postgresql:5432/conduktor-console"
+      CDK_MONITORING_CORTEX-URL: http://conduktor-monitoring:9009/
+      CDK_MONITORING_ALERT-MANAGER-URL: http://conduktor-monitoring:9010/
+      CDK_MONITORING_CALLBACK-URL: http://conduktor-platform:8080/monitoring/api/
+      CDK_MONITORING_NOTIFICATIONS-CALLBACK-URL: http://localhost:8080
+
+  conduktor-monitoring:
+    image: conduktor/conduktor-console-cortex:1.21.1
+    environment:
+      CDK_CONSOLE-URL: "http://conduktor-console:8080"
+    depends_on:
+      - conduktor-console
+volumes:
+  pg_data: {}
+  conduktor_data: {}

docker-compose.yml

@@ -1,8 +1,12 @@
-# SPDX-License-Identifier: LGPL-3.0-only
+# SPDX-License-Identifier: Apache License 2.0
 # Copyright 2021 John Mille <john@ews-network.net>
 
 """Top-level package for Kafka schema registry admin."""
 
 __author__ = """JohnPreston"""
 __email__ = "john@ews-network.net"
 __version__ = "0.2.6"
+
+from .kafka_schema_registry_admin import SchemaRegistry
+
+__all__ = ["SchemaRegistry"]

kafka_schema_registry_admin/__init__.py

@@ -0,0 +1,78 @@
+# SPDX-License-Identifier: Apache License 2.0
+# Copyright 2021 John Mille <john@ews-network.net>
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from requests import Response
+
+from urllib.parse import urlparse
+
+import requests
+
+from .errors import evaluate_api_return
+
+
+class Client:
+    """API Client wrapper around the requests"""
+
+    def __init__(self, base_url: str):
+        self._base_url = base_url
+
+        self._default_headers: dict = {
+            "Accept": "application/json",
+        }
+        self._post_headers: dict = {
+            "Accept": "application/json",
+            "Content-Type": "application/vnd.schemaregistry.v1+json",
+        }
+
+    @evaluate_api_return
+    def get(self, api_path: str, *args, **kwargs) -> Response:
+        """Get the data from the api_path"""
+        headers = kwargs.get("headers", {})
+        if not headers:
+            kwargs["headers"] = headers
+        headers.update(self._default_headers)
+        url: str = urlparse(self._base_url + api_path).geturl()
+
+        response = requests.get(url, *args, **kwargs)
+        return response
+
+    @evaluate_api_return
+    def post(self, api_path: str, *args, **kwargs) -> Response:
+        """POST the data from the api_path"""
+        headers = kwargs.get("headers", {})
+        if not headers:
+            kwargs["headers"] = headers
+        headers.update(self._default_headers)
+        headers.update(self._post_headers)
+        url: str = urlparse(self._base_url + api_path).geturl()
+        response = requests.post(url, *args, **kwargs)
+        return response
+
+    @evaluate_api_return
+    def put(self, api_path: str, *args, **kwargs) -> Response:
+        """PUT the data from the api_path"""
+        headers = kwargs.get("headers", {})
+        if not headers:
+            kwargs["headers"] = headers
+        headers.update(self._default_headers)
+        url: str = urlparse(self._base_url + api_path).geturl()
+
+        response = requests.put(url, *args, **kwargs)
+        return response
+
+    @evaluate_api_return
+    def delete(self, api_path: str, *args, **kwargs) -> Response:
+        """DELETE the data from the api_path"""
+        headers = kwargs.get("headers", {})
+        if not headers:
+            kwargs["headers"] = headers
+        headers.update(self._default_headers)
+
+        url: str = urlparse(self._base_url + api_path).geturl()
+        response = requests.delete(url, *args, **kwargs)
+        return response