[Python Client] Python client support using custom Avro schema definition

[gaoran10]

Motivation

Currently, the Python client didn't support using schema definition to generate AvroSchema, so users couldn't use the schema definition file in the Python client.

Modifications

Add a new init-param schema_definition for AvroSchema to support initializing the AvroSchema by an Avro schema definition.

class AvroSchema(Schema):
    def __init__(self, record_cls, schema_definition=None):
        if record_cls is None and schema_definition is None:
            raise AssertionError("The param record_cls and schema_definition shouldn't be both None.")

        if record_cls is not None:
            self._schema = record_cls.schema()
        else:
            self._schema = schema_definition
        super(AvroSchema, self).__init__(record_cls, _pulsar.SchemaType.AVRO, self._schema, 'AVRO')

How to use

Assume that there is a company Avro schema definition file company.avsc like this.

{
    "doc": "this is doc",
    "namespace": "example.avro",
    "type": "record",
    "name": "Company",
    "fields": [
        {"name": "name", "type": ["null", "string"]},
        {"name": "address", "type": ["null", "string"]},
        {"name": "employees", "type": ["null", {"type": "array", "items": {
            "type": "record",
            "name": "Employee",
            "fields": [
                {"name": "name", "type": ["null", "string"]},
                {"name": "age", "type": ["null", "int"]}
            ]
        }}]},
        {"name": "labels", "type": ["null", {"type": "map", "values": "string"}]}
    ]
}

Users could load schema from file by avro.schema or fastavro.schema

refer to load_schema or Avro Schema

schema_definition = load_schema("examples/company.avsc")
# schema_definition = avro.schema.parse(open("examples/company.avsc", "rb").read()).to_json()
avro_schema = AvroSchema(None, schema_definition=schema_definition)

producer = client.create_producer(
    topic=topic,
    schema=avro_schema)
consumer = client.subscribe(topic, 'test', schema=avro_schema)

company = {
    "name": "company-name" + str(i),
    "address": 'xxx road xxx street ' + str(i),
    "employees": [
        {"name": "user" + str(i), "age": 20 + i},
        {"name": "user" + str(i), "age": 30 + i},
        {"name": "user" + str(i), "age": 35 + i},
    ],
    "labels": {
        "industry": "software" + str(i),
        "scale": ">100",
        "funds": "1000000.0"
    }
}
producer.send(company)

msg = consumer.receive()
# Users could get a dict object by `value()` method.
msg.value()

Verifying this change

Add new tests to verify encode and decode using AvroSchema generated by the Avro schema definition.

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

• Dependencies (does it add or upgrade a dependency): (no)
• The public API: (no)
• The schema: (no)
• The default values of configurations: (no)
• The wire protocol: (no)
• The rest endpoints: (no)
• The admin cli options: (no)
• Anything that affects deployment: (no)

Documentation

Check the box below and label this PR (if you have committer privilege).

Need to update docs?

• doc-required

  (If you need help on updating docs, create a doc issue)

• no-need-doc

  (Please explain why)

• doc

  (If this PR contains doc changes)

[github-actions]

@gaoran10:Thanks for your contribution. For this PR, do we need to update docs?
(The PR template contains info about doc, which helps others know more about the changes. Can you provide doc-related info in this and future PR descriptions? Thanks)