Fix use.schema.id, broken by signature change to sr.lookup_schema(). (#2104)

* Fix use.schema.id, broken by signature change to sr.lookup_schema().

* Fix PEP-8 violations.

Author: insightfuls

README.md

@@ -106,7 +106,7 @@ p = Producer({'bootstrap.servers': 'mybroker1,mybroker2'})
 
 def delivery_report(err, msg):
     """ Called once for each message produced to indicate delivery result.
-        Triggered by poll() or flush(). """
+        Triggered by poll() or flush()."""
     if err is not None:
         print('Message delivery failed: {}'.format(err))
     else:

src/confluent_kafka/schema_registry/_async/avro.py

@@ -323,11 +323,11 @@ async def __serialize(self, obj: object, ctx: Optional[SerializationContext] = N
                 # a schema without a subject so we set the schema_id here to handle
                 # the initial registration.
                 registered_schema = await self._registry.register_schema_full_response(
-                    subject, self._schema, self._normalize_schemas)
+                    subject, self._schema, normalize_schemas=self._normalize_schemas)
                 self._schema_id = SchemaId(AVRO_TYPE, registered_schema.schema_id, registered_schema.guid)
             else:
                 registered_schema = await self._registry.lookup_schema(
-                    subject, self._schema, self._normalize_schemas)
+                    subject, self._schema, normalize_schemas=self._normalize_schemas)
                 self._schema_id = SchemaId(AVRO_TYPE, registered_schema.schema_id, registered_schema.guid)
 
             self._known_subjects.add(subject)

src/confluent_kafka/schema_registry/_async/json_schema.py

@@ -332,11 +332,11 @@ async def __serialize(self, obj: object, ctx: Optional[SerializationContext] = N
                 # a schema without a subject so we set the schema_id here to handle
                 # the initial registration.
                 registered_schema = await self._registry.register_schema_full_response(
-                    subject, self._schema, self._normalize_schemas)
+                    subject, self._schema, normalize_schemas=self._normalize_schemas)
                 self._schema_id = SchemaId(JSON_TYPE, registered_schema.schema_id, registered_schema.guid)
             else:
                 registered_schema = await self._registry.lookup_schema(
-                    subject, self._schema, self._normalize_schemas)
+                    subject, self._schema, normalize_schemas=self._normalize_schemas)
                 self._schema_id = SchemaId(JSON_TYPE, registered_schema.schema_id, registered_schema.guid)
 
             self._known_subjects.add(subject)

src/confluent_kafka/schema_registry/_async/mock_schema_registry_client.py

@@ -94,8 +94,8 @@ def get_latest_version(self, subject_name: str) -> Optional[RegisteredSchema]:
             return None
 
     def get_latest_with_metadata(
-        self, subject_name: str,
-        metadata: Dict[str, str]
+        self, subject_name: str, metadata: Dict[str, str],
+        deleted: bool = False, fmt: Optional[str] = None
     ) -> Optional[RegisteredSchema]:
         with self.lock:
             if subject_name in self.subject_schemas:
@@ -155,7 +155,8 @@ async def register_schema(
         self, subject_name: str, schema: 'Schema',
         normalize_schemas: bool = False
     ) -> int:
-        registered_schema = await self.register_schema_full_response(subject_name, schema, normalize_schemas)
+        registered_schema = await self.register_schema_full_response(
+            subject_name, schema, normalize_schemas=normalize_schemas)
         return registered_schema.schema_id
 
     async def register_schema_full_response(
@@ -202,7 +203,7 @@ async def get_schema_by_guid(
 
     async def lookup_schema(
         self, subject_name: str, schema: 'Schema',
-        normalize_schemas: bool = False, deleted: bool = False
+        normalize_schemas: bool = False, fmt: Optional[str] = None, deleted: bool = False
     ) -> 'RegisteredSchema':
 
         registered_schema = self._store.get_registered_schema_by_schema(subject_name, schema)

src/confluent_kafka/schema_registry/_async/protobuf.py

@@ -407,11 +407,11 @@ async def __serialize(self, message: Message, ctx: Optional[SerializationContext
 
             if self._auto_register:
                 registered_schema = await self._registry.register_schema_full_response(
-                    subject, self._schema, self._normalize_schemas)
+                    subject, self._schema, normalize_schemas=self._normalize_schemas)
                 self._schema_id = SchemaId(PROTOBUF_TYPE, registered_schema.schema_id, registered_schema.guid)
             else:
                 registered_schema = await self._registry.lookup_schema(
-                    subject, self._schema, self._normalize_schemas)
+                    subject, self._schema, normalize_schemas=self._normalize_schemas)
                 self._schema_id = SchemaId(PROTOBUF_TYPE, registered_schema.schema_id, registered_schema.guid)
 
             self._known_subjects.add(subject)

src/confluent_kafka/schema_registry/_async/schema_registry_client.py

@@ -637,7 +637,8 @@ async def register_schema(
             `POST Subject Version API Reference <https://docs.confluent.io/current/schema-registry/develop/api.html#post--subjects-(string-%20subject)-versions>`_
         """  # noqa: E501
 
-        registered_schema = await self.register_schema_full_response(subject_name, schema, normalize_schemas)
+        registered_schema = await self.register_schema_full_response(
+            subject_name, schema, normalize_schemas=normalize_schemas)
         return registered_schema.schema_id
 
     async def register_schema_full_response(

src/confluent_kafka/schema_registry/_async/serde.py

@@ -47,10 +47,11 @@ class AsyncBaseSerde(object):
     async def _get_reader_schema(self, subject: str, fmt: Optional[str] = None) -> Optional[RegisteredSchema]:
         if self._use_schema_id is not None:
             schema = await self._registry.get_schema(self._use_schema_id, subject, fmt)
-            return await self._registry.lookup_schema(subject, schema, False, True)
+            return await self._registry.lookup_schema(
+                subject, schema, normalize_schemas=False, deleted=True)
         if self._use_latest_with_metadata is not None:
             return await self._registry.get_latest_with_metadata(
-                subject, self._use_latest_with_metadata, True, fmt)
+                subject, self._use_latest_with_metadata, deleted=True, fmt=fmt)
         if self._use_latest_version:
             return await self._registry.get_latest_version(subject, fmt)
         return None
@@ -238,7 +239,8 @@ async def _get_migrations(
         self, subject: str, source_info: Schema,
         target: RegisteredSchema, fmt: Optional[str]
     ) -> List[Migration]:
-        source = await self._registry.lookup_schema(subject, source_info, False, True)
+        source = await self._registry.lookup_schema(
+            subject, source_info, normalize_schemas=False, deleted=True)
         migrations = []
         if source.version < target.version:
             migration_mode = RuleMode.UPGRADE

src/confluent_kafka/schema_registry/_sync/avro.py

@@ -323,11 +323,11 @@ def __serialize(self, obj: object, ctx: Optional[SerializationContext] = None) -
                 # a schema without a subject so we set the schema_id here to handle
                 # the initial registration.
                 registered_schema = self._registry.register_schema_full_response(
-                    subject, self._schema, self._normalize_schemas)
+                    subject, self._schema, normalize_schemas=self._normalize_schemas)
                 self._schema_id = SchemaId(AVRO_TYPE, registered_schema.schema_id, registered_schema.guid)
             else:
                 registered_schema = self._registry.lookup_schema(
-                    subject, self._schema, self._normalize_schemas)
+                    subject, self._schema, normalize_schemas=self._normalize_schemas)
                 self._schema_id = SchemaId(AVRO_TYPE, registered_schema.schema_id, registered_schema.guid)
 
             self._known_subjects.add(subject)

src/confluent_kafka/schema_registry/_sync/json_schema.py

@@ -332,11 +332,11 @@ def __serialize(self, obj: object, ctx: Optional[SerializationContext] = None) -
                 # a schema without a subject so we set the schema_id here to handle
                 # the initial registration.
                 registered_schema = self._registry.register_schema_full_response(
-                    subject, self._schema, self._normalize_schemas)
+                    subject, self._schema, normalize_schemas=self._normalize_schemas)
                 self._schema_id = SchemaId(JSON_TYPE, registered_schema.schema_id, registered_schema.guid)
             else:
                 registered_schema = self._registry.lookup_schema(
-                    subject, self._schema, self._normalize_schemas)
+                    subject, self._schema, normalize_schemas=self._normalize_schemas)
                 self._schema_id = SchemaId(JSON_TYPE, registered_schema.schema_id, registered_schema.guid)
 
             self._known_subjects.add(subject)

src/confluent_kafka/schema_registry/_sync/mock_schema_registry_client.py

@@ -94,8 +94,8 @@ def get_latest_version(self, subject_name: str) -> Optional[RegisteredSchema]:
             return None
 
     def get_latest_with_metadata(
-        self, subject_name: str,
-        metadata: Dict[str, str]
+        self, subject_name: str, metadata: Dict[str, str],
+        deleted: bool = False, fmt: Optional[str] = None
     ) -> Optional[RegisteredSchema]:
         with self.lock:
             if subject_name in self.subject_schemas:
@@ -155,7 +155,8 @@ def register_schema(
         self, subject_name: str, schema: 'Schema',
         normalize_schemas: bool = False
     ) -> int:
-        registered_schema = self.register_schema_full_response(subject_name, schema, normalize_schemas)
+        registered_schema = self.register_schema_full_response(
+            subject_name, schema, normalize_schemas=normalize_schemas)
         return registered_schema.schema_id
 
     def register_schema_full_response(
@@ -202,7 +203,7 @@ def get_schema_by_guid(
 
     def lookup_schema(
         self, subject_name: str, schema: 'Schema',
-        normalize_schemas: bool = False, deleted: bool = False
+        normalize_schemas: bool = False, fmt: Optional[str] = None, deleted: bool = False
     ) -> 'RegisteredSchema':
 
         registered_schema = self._store.get_registered_schema_by_schema(subject_name, schema)