# python-kafka-utils

Python Kafka Utils is a library for publishing messages to kafka and parsing the
messages in event handler lambdas.

![Basic Workflow](./.images/workflow.png)

> Publish new versions using the [dedicated pipeline](https://pipeline.theorchard.io/job/python-kafka-utils-pipeline/)

## Quick Start

Refer to the [examples directory](./examples/) for common usage patterns.

### Installation

```sh
pip install -i https://pypi.theorchard.io/pypi/ kafka_utils

# To include Avro or JSON Schema support, include the optional support packages, e.g.
pip install -i https://pypi.theorchard.io/pypi/ kafka_utils[avro,json]
```

> TROUBLESHOOTING
> If you use pyenv and encounter an error on MacOS like "module not found '_lzma'"
> when using the avro modules, try reinstalling your preferred python version. Using 3.11 as an example:
>
> ```sh
> PYTHON_CONFIGURE_OPTS="--enable-framework" pyenv install 3.11
> ```
>
> For M1 mac users, you may need to install [librdkafka](https://github.com/edenhill/librdkafka#installation) via `brew` and prepend CFLAGS to the pip install command:
>
> ```sh
> CFLAGS="-I/opt/homebrew/include -L/opt/homebrew/lib" env/bin/pip install -r requirements.txt
> ```

### Producing

Example producing a string encoded message to a topic:

```python
from confluent_kafka.serialization import StringSerializer
from kafka_utils.producer.event import EventProducer

string_serializer = StringSerializer()
producer = EventProducer(
  BOOTSTRAP_SERVERS,
  string_serializer,
  string_serializer,
  PROTOCOL)
producer.produce(
    TOPIC,
    'example-product-1',
    {'some': 'message'},
    delivery_callback_function
)
```

Serializers encode message keys and values before producing messages. They can
optionally be passed into the EventProducer constructor to be used by default
when producing. You can also pass them into the `produce` method to override the
serializers that may have been passed into the constructor. Alternately, you can
omit them from both and handle serialization outside of the EventProducer class.

### Consumer

Example consuming a string encoded message in a lambda handler method:

```python
from kafka_utils.consumer.deserializer.string import StringDeserializer
from kafka_utils.consumer.source.mapping import EventSourceMessage

def lambda_handler(event, context):
    deserializer = StringDeserializer()
    message = EventSourceMessage(event)
    for batch_key, msk_message in message:
        event_value = deserializer.deserialize(msk_value)
        { lambda logic here }

```

## Schema Registry

The [schema registry](https://docs.confluent.io/platform/current/schema-registry/index.html) is a remote repository for message structure definitions.

The `Avro` and `JSON Schema` serializers will automatically connect to the schema registry to handle serializing and deserializing messages while validating that data is properly formatted.
Also see [Orchard Kafka Connect README](https://github.com/theorchard/kafka-connect/blob/master/schema_registry/README.md). The readme also contains instructions on running a schema registry, and broker locally for development. The deployed schemas can be inspected in AKHQ from the`Schema Registry` link in the sidebar. [QA AKHQ](https://qa-akhq.theorchard.io) [Dev AKHQ](https://dev-akhq.dev.theorchard.io)

### Versioning

Schemas may change over time, and the individual versions are tracked in the schema registry. The event publisher requires not just the schema topic, but also the version of the schema to publish. This allows for schema updates without breaking any deployed code.

## Utility Classes

### [EventSourceMessage(_event_)](kafka_utils/consumer/source/mapping.py)

- **event** : _dict_ : event object passed to lambda handler

The `EventSourceMessage` class models the data passed from an Event Source Mapping
to a lambda as the event data for an invocation. ![Event Source Mapping](./.images/event_source_mapping.png)

The format looks like [this](./tests/unit/mock_event_data/simple_json.json). There can be multiple kafka messages in each event source message, as configured by the `batch_size` of the mapping. ([AWS Docs](https://docs.aws.amazon.com/lambda/latest/dg/invocation-eventsourcemapping.html)) It is worth noting that the MSK Event Source creates a 'consumer group' for the kafka topic that tracks which messages have been processed, in order. In the event of an error, the event source mapping will continually retry the failed message until it succeeds or the kafka message expires and is removed from the topic. MSK message keys and values are twice encoded, first using `base64` and then using one of the serializers `Avro`, `String`, `Simple JSON`, or `JSON Schema`.

Provides a generator to iterate through each kafka message in the event source mapping message.

```python
message = EventSourceMessage(event)
for batch_key, msk_message in message:
    if msk_message.value:
        # value contains the parsed body of the event, if it exists
        print(replication_event.key)
```

### [DebeziumMessage(_message_[,_allowed-event-ops_])](./kafka_utils/consumer/message/debezium.py)

- **message**: _dict_ : deserialized debezium source message dict
- **[allowed_event_ops]**: _list|None_ : list of ops to allow. If passed, messages with `op` values not in the list will raise a `DebeziumMessageException`

Provides common debezium mysql source message validation and traversal.

```python
db_msg = DebeziumMessage(message, ['u'])
status = db_msg.get_field_values('status')
assert status.get('before') != 'done'
```

### [ProductEventMessage(_message_[,_schemaId_])](kafka_utils/consumer/message/product_review.py)

- **message**: _dict_ : deserialized message dict
- **[schemaId]**: _int|None_ : id of the schema the message was produced with

```python
product_review_message = ProductEventMessage(message)
assert product_review_message.schema_id == 2
```

## Local Development

Running a local schema registry and event broker can be done using `docker-compose`:

```bash
docker-compose up -d
```

You can use the local AKHQ or Schema Registry API to publish your own schemas locally and produce messages to a local topic.

- AKHQ: [http://localhost:8080/](http://localhost:8080/)
- Schema Registry: [http://localhost:8081/](http://localhost:8081/) - Schema registry API calls can be made against this.

This is what your local development values may look like:

```bash
KAFKA_BOOTSTRAP_SERVERS="localhost:9092"
KAFKA_TOPIC="my_topic"
SCHEMA_ID=1
SCHEMA_REGISTRY_URL="http://localhost:8081"
SCHEMA_REGISTRY_SECURITY_PROTOCOL="PLAINTEXT"
```

## Pytest Integration

The confluent-kafka producer is automatically mocked when running pytest tests in
code bases that use this library. The [kafka_mock](kafka_utils/testing/unit/fixture_plugin.py)
automatically replaces the Producer class when tests are collected. It is not necessary
to include the `kafka_mock` fixture in the test method signature, but it can be
included if you wish to inspect the state of the producer in tests.

The MockProducer class offers some helper methods:

- _assert_messages(message list)_: Accepts an ordered list of messages to check
  against the list of messages that the producer handled during the tests. The
  messages should be dicts of keys you want to compare to the actual messages.
  The MockMessage class has many keys which you may not want to verify in your
  tests, so this method only compares the keys that you pass in.
- _assert_call_count(method_name, call_count)_: Asserts that `Producer.[method_name]`
  was called `call_count` number of times. Useful for verifying if methods like
  `flush`, `poll`, or `produce` were called the expected number of times.
- _set_return_error(return_error_bool)_: The mock producer will invoke delivery
  callbacks if they are provided. Use _set_return_error(True)_ to mock an error
  state in the callback, otherwise the callback will be triggered with a mocked
  successful state.
- _reset()_: Fully resets the state of the producer.

Example:

```python
def test_without_explicit_fixture():
    method_that_triggers_producing_a_message()
    # test logic here ...


def test_with_explicit_fixture(kafka_mock):
    method_that_triggers_producing_a_message()
    kafka_mock.assert_messages([{
        'topic': 'event.foo.bar',
        'value': b'{"foo": "bar", "baz": "qux"}',
        'key': b'foo-1'
    }])
    kafka_mock.assert_call_count('flush', 1)
    assert len(kafka_mock.queue) == 0
```