Add get_by_version to CachedSchemaRegistryClient (#867)

* Add get_by_version to cached schema registry client

* update docstring

* Fix tests and docker context

Author: garrett92895

src/confluent_kafka/avro/cached_schema_registry_client.py

@@ -321,7 +321,7 @@ def get_by_id(self, schema_id):
 
     def get_latest_schema(self, subject):
         """
-        GET /subjects/(string: subject)/versions/(versionId: version)
+        GET /subjects/(string: subject)/versions/latest
 
         Return the latest 3-tuple of:
         (the schema id, the parsed avro schema, the schema version)
@@ -334,7 +334,25 @@ def get_latest_schema(self, subject):
         :returns: (schema_id, schema, version)
         :rtype: (string, schema, int)
         """
-        url = '/'.join([self.url, 'subjects', subject, 'versions', 'latest'])
+        return self.get_by_version(subject, 'latest')
+
+    def get_by_version(self, subject, version):
+        """
+        GET /subjects/(string: subject)/versions/(versionId: version)
+
+        Return the 3-tuple of:
+        (the schema id, the parsed avro schema, the schema version)
+        for a particular subject and version.
+
+        This call always contacts the registry.
+
+        If the subject is not found, (None,None,None) is returned.
+        :param str subject: subject name
+        :param int version: version number
+        :returns: (schema_id, schema, version)
+        :rtype: (string, schema, int)
+        """
+        url = '/'.join([self.url, 'subjects', subject, 'versions', str(version)])
 
         result, code = self._send_request(url)
         if code == 404:

tests/avro/mock_registry.py

@@ -59,6 +59,7 @@ def __init__(self, *args, **kwargs):
         self.all_routes = {
             'GET': [
                 (r"/schemas/ids/(\d+)", 'get_schema_by_id'),
+                (r"/subjects/(\w+)/versions/(\d+)", 'get_by_version'),
                 (r"/subjects/(\w+)/versions/latest", 'get_latest')
             ],
             'POST': [
@@ -170,6 +171,21 @@ def get_latest(self, req, groups):
         }
         return (200, result)
 
+    def get_by_version(self, req, groups):
+        subject = groups[0]
+        version = int(groups[1])
+
+        schema_id, avro_schema, version = self.registry.get_by_version(subject, version)
+        if schema_id is None:
+            return self._create_error("Not found", 404)
+        result = {
+            "schema": json.dumps(avro_schema.to_json()),
+            "subject": subject,
+            "id": schema_id,
+            "version": version
+        }
+        return (200, result)
+
     def add_count(self, path):
         if path not in self.counts:
             self.counts[path] = 0

tests/avro/mock_schema_registry_client.py

@@ -135,6 +135,28 @@ def get_by_id(self, schema_id):
         """Retrieve a parsed avro schema by id or None if not found"""
         return self.id_to_schema.get(schema_id, None)
 
+    def get_by_version(self, subject, version):
+        """
+        Return the 3-tuple of:
+        (the schema id, the parsed avro schema, the schema version)
+        for a particular subject and version.
+
+        If the subject or version are not found, (None,None,None) is returned.
+        """
+        schema_versions = self.subject_to_schema_versions.get(subject) or []
+        # get schema+version by version
+        schema_versions = [x for x in schema_versions.items() if x[1] == version]
+        schema = None
+        schema_version = None
+        if len(schema_versions) > 0:
+            schema = schema_versions[0][0]
+            schema_version = schema_versions[0][1]
+
+        # get schema_id by schema
+        schema_id = self.subject_to_schema_ids.get(subject).get(schema)
+
+        return (schema_id, schema, schema_version)
+
     def get_latest_schema(self, subject):
         """
         Return the latest 3-tuple of:

tests/avro/test_cached_client.py

@@ -129,6 +129,12 @@ def test_multi_register(self):
         # latest should not change with a re-reg
         self.assertEqual(latest2, latest3)
 
+        schema1 = client.get_by_version(subject, v1)
+        schema2 = client.get_by_version(subject, v2)
+        self.assertEqual(schema1, latest1)
+        self.assertEqual(schema2, latest2)
+        self.assertEqual(schema2, latest3)
+
     def hash_func(self):
         return hash(str(self))