brads-test-repo-124u9432


Namebrads-test-repo-124u9432 JSON
Version 2.4.2 PyPI version JSON
download
home_pagehttps://github.com/marcosschroh/python-schema-registry-client
SummaryPython Rest Client to interact against Schema Registry Confluent Server to manage Avro Schemas
upload_time2023-04-19 00:58:56
maintainer
docs_urlNone
authorMarcos Schroh
requires_python
licenseMIT
keywords schema registry python avro apache apache avro json json schema
VCS
bugtrack_url
requirements No requirements were recorded.
Travis-CI No Travis.
coveralls test coverage No coveralls.
            # Python Rest Client Schema Registry

[![Build Status](https://img.shields.io/endpoint.svg?url=https%3A%2F%2Factions-badge.atrox.dev%2Fmarcosschroh%2Fpython-schema-registry-client%2Fbadge%3Fref%3Dmaster&style=flat)](https://actions-badge.atrox.dev/marcosschroh/python-schema-registry-client/goto?ref=master)
[![GitHub license](https://img.shields.io/github/license/marcosschroh/python-schema-registry-client.svg)](https://github.com/marcosschroh/python-schema-registry-client/blob/master/LICENSE)
[![codecov](https://codecov.io/gh/marcosschroh/python-schema-registry-client/branch/master/graph/badge.svg)](https://codecov.io/gh/marcosschroh/python-schema-registry-client)
[![Python Version](https://img.shields.io/badge/python-3.7+-blue.svg)](https://img.shields.io/badge/python-3.7+-blue.svg)

Python Rest Client to interact against [schema-registry](https://docs.confluent.io/current/schema-registry/index.html) confluent server to manage [Avro](https://docs.oracle.com/database/nosql-12.1.3.1/GettingStartedGuide/avroschemas.html) and [JSON](https://json-schema.org/) schemas resources.

## Requirements

python 3.7+

## Installation

```bash
pip install python-schema-registry-client
```

If you want the `Faust` functionality:

```bash
pip install python-schema-registry-client[faust]
```

Note that this will automatically add a dependency on the [faust-streaming](https://github.com/faust-streaming/faust) fork of faust. If you want to use the
old faust version, simply install it manually and then install `python-schema-registry-client` without the `faust` extra enabled, the functionality will
be the same.

## Client API, Serializer, Faust Integration and Schema Server description

**Documentation**: [https://marcosschroh.github.io/python-schema-registry-client.io](https://marcosschroh.github.io/python-schema-registry-client)

## Avro Schema Usage

```python
from schema_registry.client import SchemaRegistryClient, schema

client = SchemaRegistryClient(url="http://127.0.0.1:8081")

deployment_schema = {
    "type": "record",
    "namespace": "com.kubertenes",
    "name": "AvroDeployment",
    "fields": [
        {"name": "image", "type": "string"},
        {"name": "replicas", "type": "int"},
        {"name": "port", "type": "int"},
    ],
}

avro_schema = schema.AvroSchema(deployment_schema)

schema_id = client.register("test-deployment", avro_schema)
```

or async

```python
from schema_registry.client import AsyncSchemaRegistryClient, schema

async_client = AsyncSchemaRegistryClient(url="http://127.0.0.1:8081")

deployment_schema = {
    "type": "record",
    "namespace": "com.kubertenes",
    "name": "AvroDeployment",
    "fields": [
        {"name": "image", "type": "string"},
        {"name": "replicas", "type": "int"},
        {"name": "port", "type": "int"},
    ],
}

avro_schema = schema.AvroSchema(deployment_schema)

schema_id = await async_client.register("test-deployment", avro_schema)
```

## JSON Schema Usage

```python
from schema_registry.client import SchemaRegistryClient, schema

client = SchemaRegistryClient(url="http://127.0.0.1:8081")

deployment_schema = {
    "definitions" : {
        "JsonDeployment" : {
            "type" : "object",
            "required" : ["image", "replicas", "port"],
            "properties" : {
                "image" :       {"type" : "string"},
                "replicas" :    {"type" : "integer"},
                "port" :        {"type" : "integer"}
            }
        }
    },
    "$ref" : "#/definitions/JsonDeployment"
}

json_schema = schema.JsonSchema(deployment_schema)

schema_id = client.register("test-deployment", json_schema)
```

or async

```python
from schema_registry.client import AsyncSchemaRegistryClient, schema

async_client = AsyncSchemaRegistryClient(url="http://127.0.0.1:8081")

deployment_schema = {
    "definitions" : {
        "JsonDeployment" : {
            "type" : "object",
            "required" : ["image", "replicas", "port"],
            "properties" : {
                "image" :       {"type" : "string"},
                "replicas" :    {"type" : "integer"},
                "port" :        {"type" : "integer"}
            }
        }
    },
    "$ref" : "#/definitions/JsonDeployment"
}

json_schema = schema.JsonSchema(deployment_schema)

schema_id = await async_client.register("test-deployment", json_schema)
```

## Usage with dataclasses-avroschema for avro schemas

You can generate the `avro schema` directely from a python class using [dataclasses-avroschema](https://github.com/marcosschroh/dataclasses-avroschema)
and use it in the API for `register schemas`, `check versions` and `test compatibility`:

```python
import dataclasses

from dataclasses_avroschema import AvroModel, types

from schema_registry.client import SchemaRegistryClient

client = SchemaRegistryClient(url="http://127.0.0.1:8081")


@dataclasses.dataclass
class UserAdvance(AvroModel):
    name: str
    age: int
    pets: typing.List[str] = dataclasses.field(default_factory=lambda: ["dog", "cat"])
    accounts: typing.Dict[str, int] = dataclasses.field(default_factory=lambda: {"key": 1})
    has_car: bool = False
    favorite_colors: types.Enum = types.Enum(["BLUE", "YELLOW", "GREEN"], default="BLUE")
    country: str = "Argentina"
    address: str = None

# register the schema
schema_id = client.register(subject, UserAdvance.avro_schema())

print(schema_id)
# >>> 12

result = client.check_version(subject, UserAdvance.avro_schema())
print(result)
# >>> SchemaVersion(subject='dataclasses-avroschema-subject-2', schema_id=12, schema=1, version={"type":"record" ...')

compatibility = client.test_compatibility(subject, UserAdvance.avro_schema())
print(compatibility)

# >>> True
```

### Usage with pydantic for json schemas
You can generate the json schema directely from a python class using pydantic and use it in the API for register schemas, check versions and test compatibility:

```python
import typing

from enum import Enum

from pydantic import BaseModel

from schema_registry.client import SchemaRegistryClient

client = SchemaRegistryClient(url="http://127.0.0.1:8081")

class ColorEnum(str, Enum):
  BLUE = "BLUE"
  YELLOW = "YELLOW"
  GREEN = "GREEN"


class UserAdvance(BaseModel):
    name: str
    age: int
    pets: typing.List[str] = ["dog", "cat"]
    accounts: typing.Dict[str, int] = {"key": 1}
    has_car: bool = False
    favorite_colors: ColorEnum = ColorEnum.BLUE
    country: str = "Argentina"
    address: str = None

# register the schema
schema_id = client.register(subject, UserAdvance.schema_json(), schema_type="JSON")

print(schema_id)
# >>> 12

result = client.check_version(subject, UserAdvance.schema_json(), schema_type="JSON")
print(result)
# >>> SchemaVersion(subject='pydantic-jsonschema-subject', schema_id=12, schema=1, version=<schema_registry.client.schema.JsonSchema object at 0x7f40354550a0>)

compatibility = client.test_compatibility(subject, UserAdvance.schema_json(), schema_type="JSON")
print(compatibility)

# >>> True
```

## Serializers

You can use `AvroMessageSerializer` to encode/decode messages in `avro`

```python
from schema_registry.client import SchemaRegistryClient, schema
from schema_registry.serializers import AvroMessageSerializer


client = SchemaRegistryClient("http://127.0.0.1:8081")
avro_message_serializer = AvroMessageSerializer(client)

avro_user_schema = schema.AvroSchema({
    "type": "record",
    "namespace": "com.example",
    "name": "AvroUsers",
    "fields": [
        {"name": "first_name", "type": "string"},
        {"name": "last_name", "type": "string"},
        {"name": "age", "type": "int"},

    ],
})

# We want to encode the user_record with avro_user_schema
user_record = {
    "first_name": "my_first_name",
    "last_name": "my_last_name",
    "age": 20,
}

# Encode the record
message_encoded = avro_message_serializer.encode_record_with_schema(
    "user", avro_user_schema, user_record)

print(message_encoded)
# >>> b'\x00\x00\x00\x00\x01\x1amy_first_name\x18my_last_name('
```

or with `json schemas`

```python
from schema_registry.client import SchemaRegistryClient, schema
from schema_registry.serializers import JsonMessageSerializer


client = SchemaRegistryClient("http://127.0.0.1:8081")
json_message_serializer = JsonMessageSerializer(client)

json_schema = schema.JsonSchema({
  "definitions" : {
    "record:python.test.basic.basic" : {
      "description" : "basic schema for tests",
      "type" : "object",
      "required" : [ "number", "name" ],
      "properties" : {
        "number" : {
          "oneOf" : [ {
            "type" : "integer"
          }, {
            "type" : "null"
          } ]
        },
        "name" : {
          "oneOf" : [ {
            "type" : "string"
          } ]
        }
      }
    }
  },
  "$ref" : "#/definitions/record:python.test.basic.basic"
})

# Encode the record
basic_record = {
    "number": 10,
    "name": "a_name",
}

message_encoded = json_message_serializer.encode_record_with_schema(
    "basic", json_schema, basic_record)

print(message_encoded)
# >>> b'\x00\x00\x00\x00\x02{"number": 10, "name": "a_name"}'
```

## When use this library

Usually, we have a situation like this:

![Confluent Architecture](docs/img/confluent_architecture.png)

So, our producers/consumers have to serialize/deserialize messages every time that they send/receive from Kafka topics. In this picture, we can imagine a `Faust` application receiving messages (encoded with an Avro schema) and we want to deserialize them, so we can ask the `schema server` to do that for us. In this scenario, the `MessageSerializer` is perfect.

Also, could be a use case that we would like to have an Application only to administrate `Avro Schemas` (register, update compatibilities, delete old schemas, etc.), so the `SchemaRegistryClient` is perfect.

## Development

Install the project and development utilities in edit mode:

```bash
pip3 install -e ".[tests,docs,faust]"
```

The tests are run against the `Schema Server` using `docker compose`, so you will need
`Docker` and `Docker Compose` installed.

```bash
./scripts/test
```

You can run tests with arbitrary python version by:

```bash
./scripts/test --python-version 3.x
```

All additional args will be passed to pytest, for example:

```bash
./scripts/test ./tests/client/ --maxfail=1 
```

Run code linting:

```bash
./scripts/lint
```

To perform tests using the python shell you can run the project using `docker-compose`.

1. Build: `docker-compose build --build-arg PYTHON_VERSION=$PYTHON_VERSION`
2. Execute `docker-compose up`. Then, the `schema registry server` will run on `http://127.0.0.1:8081`, then you can interact against it using the `SchemaRegistryClient`:
3. Use the python interpreter (get a python shell typing `python` in your command line)
4. Play with the `schema server`

```python
from schema_registry.client import SchemaRegistryClient, schema

client = SchemaRegistryClient(url="http://127.0.0.1:8081")

# do some operations with the client...
deployment_schema = {
    "type": "record",
    "namespace": "com.kubertenes",
    "name": "AvroDeployment",
    "fields": [
        {"name": "image", "type": "string"},
        {"name": "replicas", "type": "int"},
        {"name": "port", "type": "int"},
    ],
}

avro_schema = schema.AvroSchema(deployment_schema)
client.register("test-deployment", avro_schema)
# >>>> Out[5]: 1
```

Then, you can check the schema using your browser going to the url `http://127.0.0.1:8081/schemas/ids/1`

            

Raw data

            {
    "_id": null,
    "home_page": "https://github.com/marcosschroh/python-schema-registry-client",
    "name": "brads-test-repo-124u9432",
    "maintainer": "",
    "docs_url": null,
    "requires_python": "",
    "maintainer_email": "",
    "keywords": "Schema Registry,Python,Avro,Apache,Apache Avro,JSON,JSON Schema",
    "author": "Marcos Schroh",
    "author_email": "schrohm@gmail.com",
    "download_url": "https://files.pythonhosted.org/packages/36/01/21ce6b0cd85e3c57f302a181d2b41ecd0a44ef3b593000e35db986192529/brads-test-repo-124u9432-2.4.2.tar.gz",
    "platform": null,
    "description": "# Python Rest Client Schema Registry\n\n[![Build Status](https://img.shields.io/endpoint.svg?url=https%3A%2F%2Factions-badge.atrox.dev%2Fmarcosschroh%2Fpython-schema-registry-client%2Fbadge%3Fref%3Dmaster&style=flat)](https://actions-badge.atrox.dev/marcosschroh/python-schema-registry-client/goto?ref=master)\n[![GitHub license](https://img.shields.io/github/license/marcosschroh/python-schema-registry-client.svg)](https://github.com/marcosschroh/python-schema-registry-client/blob/master/LICENSE)\n[![codecov](https://codecov.io/gh/marcosschroh/python-schema-registry-client/branch/master/graph/badge.svg)](https://codecov.io/gh/marcosschroh/python-schema-registry-client)\n[![Python Version](https://img.shields.io/badge/python-3.7+-blue.svg)](https://img.shields.io/badge/python-3.7+-blue.svg)\n\nPython Rest Client to interact against [schema-registry](https://docs.confluent.io/current/schema-registry/index.html) confluent server to manage [Avro](https://docs.oracle.com/database/nosql-12.1.3.1/GettingStartedGuide/avroschemas.html) and [JSON](https://json-schema.org/) schemas resources.\n\n## Requirements\n\npython 3.7+\n\n## Installation\n\n```bash\npip install python-schema-registry-client\n```\n\nIf you want the `Faust` functionality:\n\n```bash\npip install python-schema-registry-client[faust]\n```\n\nNote that this will automatically add a dependency on the [faust-streaming](https://github.com/faust-streaming/faust) fork of faust. If you want to use the\nold faust version, simply install it manually and then install `python-schema-registry-client` without the `faust` extra enabled, the functionality will\nbe the same.\n\n## Client API, Serializer, Faust Integration and Schema Server description\n\n**Documentation**: [https://marcosschroh.github.io/python-schema-registry-client.io](https://marcosschroh.github.io/python-schema-registry-client)\n\n## Avro Schema Usage\n\n```python\nfrom schema_registry.client import SchemaRegistryClient, schema\n\nclient = SchemaRegistryClient(url=\"http://127.0.0.1:8081\")\n\ndeployment_schema = {\n    \"type\": \"record\",\n    \"namespace\": \"com.kubertenes\",\n    \"name\": \"AvroDeployment\",\n    \"fields\": [\n        {\"name\": \"image\", \"type\": \"string\"},\n        {\"name\": \"replicas\", \"type\": \"int\"},\n        {\"name\": \"port\", \"type\": \"int\"},\n    ],\n}\n\navro_schema = schema.AvroSchema(deployment_schema)\n\nschema_id = client.register(\"test-deployment\", avro_schema)\n```\n\nor async\n\n```python\nfrom schema_registry.client import AsyncSchemaRegistryClient, schema\n\nasync_client = AsyncSchemaRegistryClient(url=\"http://127.0.0.1:8081\")\n\ndeployment_schema = {\n    \"type\": \"record\",\n    \"namespace\": \"com.kubertenes\",\n    \"name\": \"AvroDeployment\",\n    \"fields\": [\n        {\"name\": \"image\", \"type\": \"string\"},\n        {\"name\": \"replicas\", \"type\": \"int\"},\n        {\"name\": \"port\", \"type\": \"int\"},\n    ],\n}\n\navro_schema = schema.AvroSchema(deployment_schema)\n\nschema_id = await async_client.register(\"test-deployment\", avro_schema)\n```\n\n## JSON Schema Usage\n\n```python\nfrom schema_registry.client import SchemaRegistryClient, schema\n\nclient = SchemaRegistryClient(url=\"http://127.0.0.1:8081\")\n\ndeployment_schema = {\n    \"definitions\" : {\n        \"JsonDeployment\" : {\n            \"type\" : \"object\",\n            \"required\" : [\"image\", \"replicas\", \"port\"],\n            \"properties\" : {\n                \"image\" :       {\"type\" : \"string\"},\n                \"replicas\" :    {\"type\" : \"integer\"},\n                \"port\" :        {\"type\" : \"integer\"}\n            }\n        }\n    },\n    \"$ref\" : \"#/definitions/JsonDeployment\"\n}\n\njson_schema = schema.JsonSchema(deployment_schema)\n\nschema_id = client.register(\"test-deployment\", json_schema)\n```\n\nor async\n\n```python\nfrom schema_registry.client import AsyncSchemaRegistryClient, schema\n\nasync_client = AsyncSchemaRegistryClient(url=\"http://127.0.0.1:8081\")\n\ndeployment_schema = {\n    \"definitions\" : {\n        \"JsonDeployment\" : {\n            \"type\" : \"object\",\n            \"required\" : [\"image\", \"replicas\", \"port\"],\n            \"properties\" : {\n                \"image\" :       {\"type\" : \"string\"},\n                \"replicas\" :    {\"type\" : \"integer\"},\n                \"port\" :        {\"type\" : \"integer\"}\n            }\n        }\n    },\n    \"$ref\" : \"#/definitions/JsonDeployment\"\n}\n\njson_schema = schema.JsonSchema(deployment_schema)\n\nschema_id = await async_client.register(\"test-deployment\", json_schema)\n```\n\n## Usage with dataclasses-avroschema for avro schemas\n\nYou can generate the `avro schema` directely from a python class using [dataclasses-avroschema](https://github.com/marcosschroh/dataclasses-avroschema)\nand use it in the API for `register schemas`, `check versions` and `test compatibility`:\n\n```python\nimport dataclasses\n\nfrom dataclasses_avroschema import AvroModel, types\n\nfrom schema_registry.client import SchemaRegistryClient\n\nclient = SchemaRegistryClient(url=\"http://127.0.0.1:8081\")\n\n\n@dataclasses.dataclass\nclass UserAdvance(AvroModel):\n    name: str\n    age: int\n    pets: typing.List[str] = dataclasses.field(default_factory=lambda: [\"dog\", \"cat\"])\n    accounts: typing.Dict[str, int] = dataclasses.field(default_factory=lambda: {\"key\": 1})\n    has_car: bool = False\n    favorite_colors: types.Enum = types.Enum([\"BLUE\", \"YELLOW\", \"GREEN\"], default=\"BLUE\")\n    country: str = \"Argentina\"\n    address: str = None\n\n# register the schema\nschema_id = client.register(subject, UserAdvance.avro_schema())\n\nprint(schema_id)\n# >>> 12\n\nresult = client.check_version(subject, UserAdvance.avro_schema())\nprint(result)\n# >>> SchemaVersion(subject='dataclasses-avroschema-subject-2', schema_id=12, schema=1, version={\"type\":\"record\" ...')\n\ncompatibility = client.test_compatibility(subject, UserAdvance.avro_schema())\nprint(compatibility)\n\n# >>> True\n```\n\n### Usage with pydantic for json schemas\nYou can generate the json schema directely from a python class using pydantic and use it in the API for register schemas, check versions and test compatibility:\n\n```python\nimport typing\n\nfrom enum import Enum\n\nfrom pydantic import BaseModel\n\nfrom schema_registry.client import SchemaRegistryClient\n\nclient = SchemaRegistryClient(url=\"http://127.0.0.1:8081\")\n\nclass ColorEnum(str, Enum):\n  BLUE = \"BLUE\"\n  YELLOW = \"YELLOW\"\n  GREEN = \"GREEN\"\n\n\nclass UserAdvance(BaseModel):\n    name: str\n    age: int\n    pets: typing.List[str] = [\"dog\", \"cat\"]\n    accounts: typing.Dict[str, int] = {\"key\": 1}\n    has_car: bool = False\n    favorite_colors: ColorEnum = ColorEnum.BLUE\n    country: str = \"Argentina\"\n    address: str = None\n\n# register the schema\nschema_id = client.register(subject, UserAdvance.schema_json(), schema_type=\"JSON\")\n\nprint(schema_id)\n# >>> 12\n\nresult = client.check_version(subject, UserAdvance.schema_json(), schema_type=\"JSON\")\nprint(result)\n# >>> SchemaVersion(subject='pydantic-jsonschema-subject', schema_id=12, schema=1, version=<schema_registry.client.schema.JsonSchema object at 0x7f40354550a0>)\n\ncompatibility = client.test_compatibility(subject, UserAdvance.schema_json(), schema_type=\"JSON\")\nprint(compatibility)\n\n# >>> True\n```\n\n## Serializers\n\nYou can use `AvroMessageSerializer` to encode/decode messages in `avro`\n\n```python\nfrom schema_registry.client import SchemaRegistryClient, schema\nfrom schema_registry.serializers import AvroMessageSerializer\n\n\nclient = SchemaRegistryClient(\"http://127.0.0.1:8081\")\navro_message_serializer = AvroMessageSerializer(client)\n\navro_user_schema = schema.AvroSchema({\n    \"type\": \"record\",\n    \"namespace\": \"com.example\",\n    \"name\": \"AvroUsers\",\n    \"fields\": [\n        {\"name\": \"first_name\", \"type\": \"string\"},\n        {\"name\": \"last_name\", \"type\": \"string\"},\n        {\"name\": \"age\", \"type\": \"int\"},\n\n    ],\n})\n\n# We want to encode the user_record with avro_user_schema\nuser_record = {\n    \"first_name\": \"my_first_name\",\n    \"last_name\": \"my_last_name\",\n    \"age\": 20,\n}\n\n# Encode the record\nmessage_encoded = avro_message_serializer.encode_record_with_schema(\n    \"user\", avro_user_schema, user_record)\n\nprint(message_encoded)\n# >>> b'\\x00\\x00\\x00\\x00\\x01\\x1amy_first_name\\x18my_last_name('\n```\n\nor with `json schemas`\n\n```python\nfrom schema_registry.client import SchemaRegistryClient, schema\nfrom schema_registry.serializers import JsonMessageSerializer\n\n\nclient = SchemaRegistryClient(\"http://127.0.0.1:8081\")\njson_message_serializer = JsonMessageSerializer(client)\n\njson_schema = schema.JsonSchema({\n  \"definitions\" : {\n    \"record:python.test.basic.basic\" : {\n      \"description\" : \"basic schema for tests\",\n      \"type\" : \"object\",\n      \"required\" : [ \"number\", \"name\" ],\n      \"properties\" : {\n        \"number\" : {\n          \"oneOf\" : [ {\n            \"type\" : \"integer\"\n          }, {\n            \"type\" : \"null\"\n          } ]\n        },\n        \"name\" : {\n          \"oneOf\" : [ {\n            \"type\" : \"string\"\n          } ]\n        }\n      }\n    }\n  },\n  \"$ref\" : \"#/definitions/record:python.test.basic.basic\"\n})\n\n# Encode the record\nbasic_record = {\n    \"number\": 10,\n    \"name\": \"a_name\",\n}\n\nmessage_encoded = json_message_serializer.encode_record_with_schema(\n    \"basic\", json_schema, basic_record)\n\nprint(message_encoded)\n# >>> b'\\x00\\x00\\x00\\x00\\x02{\"number\": 10, \"name\": \"a_name\"}'\n```\n\n## When use this library\n\nUsually, we have a situation like this:\n\n![Confluent Architecture](docs/img/confluent_architecture.png)\n\nSo, our producers/consumers have to serialize/deserialize messages every time that they send/receive from Kafka topics. In this picture, we can imagine a `Faust` application receiving messages (encoded with an Avro schema) and we want to deserialize them, so we can ask the `schema server` to do that for us. In this scenario, the `MessageSerializer` is perfect.\n\nAlso, could be a use case that we would like to have an Application only to administrate `Avro Schemas` (register, update compatibilities, delete old schemas, etc.), so the `SchemaRegistryClient` is perfect.\n\n## Development\n\nInstall the project and development utilities in edit mode:\n\n```bash\npip3 install -e \".[tests,docs,faust]\"\n```\n\nThe tests are run against the `Schema Server` using `docker compose`, so you will need\n`Docker` and `Docker Compose` installed.\n\n```bash\n./scripts/test\n```\n\nYou can run tests with arbitrary python version by:\n\n```bash\n./scripts/test --python-version 3.x\n```\n\nAll additional args will be passed to pytest, for example:\n\n```bash\n./scripts/test ./tests/client/ --maxfail=1 \n```\n\nRun code linting:\n\n```bash\n./scripts/lint\n```\n\nTo perform tests using the python shell you can run the project using `docker-compose`.\n\n1. Build: `docker-compose build --build-arg PYTHON_VERSION=$PYTHON_VERSION`\n2. Execute `docker-compose up`. Then, the `schema registry server` will run on `http://127.0.0.1:8081`, then you can interact against it using the `SchemaRegistryClient`:\n3. Use the python interpreter (get a python shell typing `python` in your command line)\n4. Play with the `schema server`\n\n```python\nfrom schema_registry.client import SchemaRegistryClient, schema\n\nclient = SchemaRegistryClient(url=\"http://127.0.0.1:8081\")\n\n# do some operations with the client...\ndeployment_schema = {\n    \"type\": \"record\",\n    \"namespace\": \"com.kubertenes\",\n    \"name\": \"AvroDeployment\",\n    \"fields\": [\n        {\"name\": \"image\", \"type\": \"string\"},\n        {\"name\": \"replicas\", \"type\": \"int\"},\n        {\"name\": \"port\", \"type\": \"int\"},\n    ],\n}\n\navro_schema = schema.AvroSchema(deployment_schema)\nclient.register(\"test-deployment\", avro_schema)\n# >>>> Out[5]: 1\n```\n\nThen, you can check the schema using your browser going to the url `http://127.0.0.1:8081/schemas/ids/1`\n",
    "bugtrack_url": null,
    "license": "MIT",
    "summary": "Python Rest Client to interact against Schema Registry Confluent Server to manage Avro Schemas",
    "version": "2.4.2",
    "split_keywords": [
        "schema registry",
        "python",
        "avro",
        "apache",
        "apache avro",
        "json",
        "json schema"
    ],
    "urls": [
        {
            "comment_text": "",
            "digests": {
                "blake2b_256": "360121ce6b0cd85e3c57f302a181d2b41ecd0a44ef3b593000e35db986192529",
                "md5": "fecc6f87204b24ae5cfd2943a8fcfa43",
                "sha256": "f48f2672ebcff95c69a852caeab827885c08fca2cfaa4422d8f2e018f026e5fc"
            },
            "downloads": -1,
            "filename": "brads-test-repo-124u9432-2.4.2.tar.gz",
            "has_sig": false,
            "md5_digest": "fecc6f87204b24ae5cfd2943a8fcfa43",
            "packagetype": "sdist",
            "python_version": "source",
            "requires_python": null,
            "size": 31230,
            "upload_time": "2023-04-19T00:58:56",
            "upload_time_iso_8601": "2023-04-19T00:58:56.311513Z",
            "url": "https://files.pythonhosted.org/packages/36/01/21ce6b0cd85e3c57f302a181d2b41ecd0a44ef3b593000e35db986192529/brads-test-repo-124u9432-2.4.2.tar.gz",
            "yanked": false,
            "yanked_reason": null
        }
    ],
    "upload_time": "2023-04-19 00:58:56",
    "github": true,
    "gitlab": false,
    "bitbucket": false,
    "github_user": "marcosschroh",
    "github_project": "python-schema-registry-client",
    "travis_ci": false,
    "coveralls": false,
    "github_actions": true,
    "lcname": "brads-test-repo-124u9432"
}
        
Elapsed time: 0.06912s