Dynamic index lookups

For many applications it is useful to be able to dynamically load schema entities. For instance, if building an application that inspects the traffic between a consumer or producer and a broker, we would like the ability to parse arbitrary requests and responses sent between the two.

The Apache Kafka® Protocol has lots of entities, and while loading them all at import time is definitely sometimes an option, as this is a fairly slow process, it is not suitable for all applications.

The kio library provides the following APIs for applications where it’s instead deemed preferable to dynamically load schema entities.

Schema entity import functions

exception kio.index.KioIndexError[source]

Bases: RuntimeError

exception kio.index.UnknownAPIKey[source]

Bases: KioIndexError

exception kio.index.UnknownEntity[source]

Bases: KioIndexError

kio.index.load_entity_module(name: str, version: int, entity_type: Literal[EntityType.header, EntityType.data, EntityType.response, EntityType.request]) → ModuleType[source]

Load a schema module given the schema name, version, and type.

Raises:

UnknownEntity – When the given arguments do not map to a known entity.

>>> load_entity_module("metadata", 12, EntityType.request)
<module 'kio.schema.metadata.v12.request' from '.../kio/schema/metadata/v12/request.py'>
Parameters:

  • name (str)

  • version (int)

  • entity_type (Literal[<EntityType.header: 3>, <EntityType.data: 4>] | Literal[<EntityType.response: 2>, <EntityType.request: 1>])

Return type:

ModuleType

kio.index.load_payload_module(api_key: int, version: int, entity_type: Literal[EntityType.response, EntityType.request]) → ModuleType[source]

Load a schema module of an API message payload, given the API key, version, and type.

Raises:

  • UnknownAPIKey – When the given API key does not map to an API name.

  • UnknownEntity – When the given arguments do not map to a known entity.

>>> load_payload_module(3, 12, EntityType.request)
<module 'kio.schema.metadata.v12.request' from '.../kio/schema/metadata/v12/request.py'>
Parameters:

  • api_key (int)

  • version (int)

  • entity_type (Literal[<EntityType.response: 2>, <EntityType.request: 1>])

Return type:

ModuleType

kio.index.load_entity_schema(name: str, version: int, entity_type: Literal[EntityType.header, EntityType.data, EntityType.response, EntityType.request]) → type[Entity][source]

Load an entity schema module given the schema name, version, and type.

Raises:

UnknownEntity – When the given arguments do not map to a known entity.

>>> load_entity_schema("metadata", 12, EntityType.response)
<class 'kio.schema.metadata.v12.response.MetadataResponse'>
Parameters:

  • name (str)

  • version (int)

  • entity_type (Literal[<EntityType.header: 3>, <EntityType.data: 4>] | Literal[<EntityType.response: 2>, <EntityType.request: 1>])

Return type:

type[Entity]

kio.index.load_response_schema(api_key: int, version: int) → type[HeaderV0ResponsePayload | HeaderV1ResponsePayload][source]

Load a response payload schema entity given its API key, and version.

Raises:

  • UnknownAPIKey – When the given API key does not map to an API name.

  • UnknownEntity – When the given arguments do not map to a known entity.

>>> load_response_schema(3, 12)
<class 'kio.schema.metadata.v12.response.MetadataResponse'>
Parameters:

  • api_key (int)

  • version (int)

Return type:

type[HeaderV0ResponsePayload | HeaderV1ResponsePayload]

kio.index.load_request_schema(api_key: int, version: int) → type[HeaderV0RequestPayload | HeaderV1RequestPayload | HeaderV2RequestPayload][source]

Load a request payload schema entity given its API key, and version.

Raises:

  • UnknownAPIKey – When the given API key does not map to an API name.

  • UnknownEntity – When the given arguments do not map to a known entity.

>>> load_request_schema(3, 12)
<class 'kio.schema.metadata.v12.request.MetadataRequest'>
Parameters:

  • api_key (int)

  • version (int)

Return type:

type[HeaderV0RequestPayload | HeaderV1RequestPayload | HeaderV2RequestPayload]

kio.index.load_response_from_request(request_type: type[HeaderV0RequestPayload | HeaderV1RequestPayload | HeaderV2RequestPayload] | HeaderV0RequestPayload | HeaderV1RequestPayload | HeaderV2RequestPayload) → type[HeaderV0ResponsePayload | HeaderV1ResponsePayload][source]

Load a response payload schema entity given its corresponding request type.

Raises:

  • UnknownAPIKey – When the given API key does not map to an API name.

  • UnknownEntity – When the given arguments do not map to a known entity.

>>> from kio.schema.metadata.v12 import MetadataRequest
>>> load_response_from_request(MetadataRequest)
<class 'kio.schema.metadata.v12.response.MetadataResponse'>
Parameters:

request_type (type[HeaderV0RequestPayload | HeaderV1RequestPayload | HeaderV2RequestPayload] | HeaderV0RequestPayload | HeaderV1RequestPayload | HeaderV2RequestPayload)

Return type:

type[HeaderV0ResponsePayload | HeaderV1ResponsePayload]

kio.index.load_request_from_response(response_type: type[HeaderV0ResponsePayload | HeaderV1ResponsePayload] | HeaderV0ResponsePayload | HeaderV1ResponsePayload) → type[HeaderV0RequestPayload | HeaderV1RequestPayload | HeaderV2RequestPayload][source]

Load a request payload schema entity given its corresponding response type.

Raises:

  • UnknownAPIKey – When the given API key does not map to an API name.

  • UnknownEntity – When the given arguments do not map to a known entity.

>>> from kio.schema.metadata.v12 import MetadataResponse
>>> load_request_from_response(MetadataResponse)
<class 'kio.schema.metadata.v12.request.MetadataRequest'>
Parameters:

response_type (type[HeaderV0ResponsePayload | HeaderV1ResponsePayload] | HeaderV0ResponsePayload | HeaderV1ResponsePayload)

Return type:

type[HeaderV0RequestPayload | HeaderV1RequestPayload | HeaderV2RequestPayload]