nameko-kafka 0.1.0

Kafka extension for Nameko microservice framework

Nameko-Kafka

Kafka extension for Nameko microservice framework.

Introduction

This is a Nameko microservice framework extension to support Kafka entrypoint and dependency. The motivation behind creating this project is this issue 569. Thus nameko-kafka tries to provide a simple implementation extension based on the approach explained by calumpeterwebb. On topo of that a dependency provider is also included for publishing Kafka messages from within Nameko services.

Installation

The package is supports Python >= 3.5

$ pip install nameko-kafka

Usage

The extension can be used for both, service dependency and entrypoint. Example usage for both the cases are shown below:

Dependency

This is basically a python-kafka producer in the form of Nameko dependency. Nameko uses dependency injection to initiate the producer. You just need to declare it in your service class:

from nameko.rpc import rpc
from nameko_kafka import KafkaProducer


class MyService:
    """
        My microservice
    """
    name = "my-service"
    # Kafak dependency
    producer = KafkaProducer(bootstrap_servers='localhost:1234')
    
    @rpc
    def method(self):
        # Publish message using dependency
        self.producer.send("kafka-topic", value=b"my-message", key=b"my-key")

Here KafkaProducer accepts all options valid for python-kafka's KafkaProducer.

Entrypoint

You can use the nameko_kafka.consume decorator in your services which process Kafka messages:

from nameko_kafka import consume


class MyService:
    """
        My microservice 
    """
    name = "my-service"

    @consume("kafka-topic", group_id="my-group", bootstrap_servers='localhost:1234')
    def method(self, message):
        # Your message handler
        handle_message(message) 

The consume decorator accepts all the options valid for python-kafka's KafkaProducer.

Configurations

The dependency configurations can be set in nameko config.yaml file, or by environment variables.

Config File

# Config for entrypoint
KAFKA_CONSUMER:
  bootstrap_servers: 'localhost:1234'
  retry_backoff_ms: 100
  ...

# Config for dependency
KAFKA_PRODUCER:
  bootstrap_servers: 'localhost:1234'
  retries: 3
  ...

Environment Variables

# Config for entrypoint
KAFKA_CONSUMER='{"bootstrap_servers": "localhost:1234", "retry_backoff_ms": 100}'

# Config for dependency
KAFKA_PRODUCER='{"bootstrap_servers": "localhost:1234", "retries": 3}'

Milestones

[x] Kafka Entrypoint [x] Kafka Dependency [ ] Advanced feature select commit strategies: ALMOST_ONCE_DELIVERY, AT_LEAST_ONCE_DELIVERY, EXACTLY_ONCE_DELIVERY [ ] Commit storage for EXACT_ONCE_DELIVERY strategy

Developers

For development a kafka broker is required. You can spawn one using the docker-compose.yml file in the tests folder:

$ cd tests
$ docker-compose up -d 

To install all package dependencies:

$ pip install -r .[dev]
or
$ make deps

Other useful commands:

$ pytest --cov=nameko_kafka tests/			# to get coverage report
or
$ make coverage

$ pylint nameko_kafka       # to check code quality with PyLint
or
$ make lint

Contributions

Pull requests always welcomed. Thanks!