feat: create a kafka consumer

.github/workflows/trigger_partner_docs_update.yml

@@ -12,16 +12,16 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Trigger partner docs update
-        if: github.repository == 'pactflow/example-consumer-python-sns'
+        if: github.repository == 'pactflow/example-consumer-python-kafka'
         uses: peter-evans/repository-dispatch@v1
         with:
           token: ${{ secrets.GHTOKENFORTRIGGERINGPACTDOCSUPDATE }}
           repository: pactflow/partners.pactflow.io
-          event-type: pactflow-example-consumer-python-sns-updated
+          event-type: pactflow-example-consumer-python-kafka-updated
       - name: Trigger docs update
-        if: github.repository == 'pactflow/example-consumer-python-sns'
+        if: github.repository == 'pactflow/example-consumer-python-kafka'
         uses: peter-evans/repository-dispatch@v1
         with:
           token: ${{ secrets.GHTOKENFORTRIGGERINGPACTDOCSUPDATE }}
           repository: pactflow/docs.pactflow.io
-          event-type: pactflow-example-consumer-python-sns-updated
+          event-type: pactflow-example-consumer-python-kafka-updated

Makefile

@@ -1,6 +1,6 @@
 # Default to the read only token - the read/write token will be present on Travis CI.
 # It's set as a secure environment variable in the .travis.yml file
-PACTICIPANT := "pactflow-example-consumer-python-sns"
+PACTICIPANT := "pactflow-example-consumer-python-kafka"
 GITHUB_WEBHOOK_UUID := "c76b601e-d66a-4eb1-88a4-6ebc50c0df8b"
 PACT_CLI="docker run --rm -v ${PWD}:${PWD} -e PACT_BROKER_BASE_URL -e PACT_BROKER_TOKEN pactfoundation/pact-cli:latest"
 
@@ -84,7 +84,7 @@ create_github_token_secret:
 create_or_update_github_webhook:
 	@"${PACT_CLI}" \
 	  broker create-or-update-webhook \
-	  'https://api.github.com/repos/pactflow/example-consumer-python-sns/statuses/$${pactbroker.consumerVersionNumber}' \
+	  'https://api.github.com/repos/pactflow/example-consumer-python-kafka/statuses/$${pactbroker.consumerVersionNumber}' \
 	  --header 'Content-Type: application/json' 'Accept: application/vnd.github.v3+json' 'Authorization: token $${user.githubCommitStatusToken}' \
 	  --request POST \
 	  --data @${PWD}/pactflow/github-commit-status-webhook.json \
@@ -106,7 +106,7 @@ test_github_webhook:
 ## ======================
 ## Python additions
 ## ======================
-PROJECT := example-consumer-python-sns
+PROJECT := example-consumer-python-kafka
 PYTHON_MAJOR_VERSION := 3.11
 
 sgr0 := $(shell tput sgr0)
@@ -116,9 +116,6 @@ green := $(shell tput setaf 2)
 deps:
 	poetry install
 
-integration:
-	sam local invoke ProductEventHandler --event  ./tests/resources/events/update.json
-
 venv:
 	@if [ -d "./.venv" ]; then echo "$(red).venv already exists, not continuing!$(sgr0)"; exit 1; fi
 	@type pyenv >/dev/null 2>&1 || (echo "$(red)pyenv not found$(sgr0)"; exit 1)
@@ -139,11 +136,5 @@ venv:
 	@echo "\n$(green)Use it! (populate .python-version)$(sgr0)"
 	pyenv local ${PROJECT}
 
-deploy_sam:
-	scripts/deploy.sh
-
-publish_sam:
-	scripts/publish.sh
-
-logs:
-	sam logs -n ProductEventHandler --stack-name pactflow-example-consumer-python-sns -t
+run:
+	python3 run.py

README.md

@@ -1,17 +1,16 @@
-# Example Python AWS SNS Consumer
+# Example Python AWS Kafka Consumer
 
-[![Build Status](https://github.com/pactflow/example-consumer-python-sns/actions/workflows/build.yml/badge.svg)](https://github.com/pactflow/example-consumer-python-sns/actions)
+[![Build Status](https://github.com/pactflow/example-consumer-python-kafka/actions/workflows/build.yml/badge.svg)](https://github.com/pactflow/example-consumer-python-kafka/actions)
 
-[![Can I deploy Status](https://test.pactflow.io/pacticipants/pactflow-example-consumer-python-sns/branches/main/latest-version/can-i-deploy/to-environment/production/badge.svg)](https://test.pactflow.io/pacticipants/pactflow-example-consumer-python-sns/branches/main/latest-version/can-i-deploy/to-environment/production/badge)
+[![Can I deploy Status](https://test.pactflow.io/pacticipants/pactflow-example-consumer-python-kafka/branches/main/latest-version/can-i-deploy/to-environment/production/badge.svg)](https://test.pactflow.io/pacticipants/pactflow-example-consumer-python-kafka/branches/main/latest-version/can-i-deploy/to-environment/production/badge)
 
-[![Pact Status](https://test.pactflow.io/pacts/provider/pactflow-example-provider-python-sns/consumer/pactflow-example-consumer-python-sns/latest/badge.svg?label=consumer)](https://test.pactflow.io/pacts/provider/pactflow-example-provider-python-sns/consumer/pactflow-example-consumer-python-sns/latest) (latest pact)
+[![Pact Status](https://test.pactflow.io/pacts/provider/pactflow-example-provider-python-kafka/consumer/pactflow-example-consumer-python-kafka/latest/badge.svg?label=consumer)](https://test.pactflow.io/pacts/provider/pactflow-example-provider-python-kafka/consumer/pactflow-example-consumer-python-kafka/latest) (latest pact)
 
-[![Pact Status](https://test.pactflow.io/matrix/provider/pactflow-example-provider-python-sns/latest/main/consumer/pactflow-example-consumer-python-sns/latest/main/badge.svg?label=consumer)](https://test.pactflow.io/pacts/provider/pactflow-example-provider-python-sns/consumer/pactflow-example-consumer-python-sns/latest/prod) (prod/prod pact)
+[![Pact Status](https://test.pactflow.io/matrix/provider/pactflow-example-provider-python-kafka/latest/main/consumer/pactflow-example-consumer-python-kafka/latest/main/badge.svg?label=consumer)](https://test.pactflow.io/pacts/provider/pactflow-example-provider-python-kafka/consumer/pactflow-example-consumer-python-kafka/latest/prod) (prod/prod pact)
 
-This is an example of a Python AWS SNS consumer that uses Pact, [Pactflow](https://pactflow.io) and GitHub Actions to ensure that it is compatible with the expectations its consumers have of it.
+This is an example of a Python Kafka consumer that uses Pact, [Pactflow](https://pactflow.io) and GitHub Actions to ensure that it is compatible with the expectations its consumers have of it.
 
-All examples in the series `example-consumer-<language>-sns` provide the same functionality to be easily comparable across languages.
-As such, please refer to [https://docs.pactflow.io/docs/examples/aws/sns/consumer/](AWS SNS Consumer Examples) to avoid unnecessary duplication of details here.
+All examples in the series `example-consumer-<language>-kafka` provide the same functionality to be easily comparable across languages.
 
 Language specific sections which differ from the canonical example only can be found below.
 
@@ -25,7 +24,7 @@ You're probably familiar with layered architectures such as Ports and Adaptors (
 
 This code base is setup with this modularity in mind:
 
-- [Lambda Handler](src/_lambda/product.py)
+- [Kafka Handler](src/_kafka/product.py)
 - [Event Service](src/product/product_service.py)
 - Business Logic
   - [Product](src/product/product.py)
@@ -43,4 +42,4 @@ See also:
 ### Testing
 
 - Run the unit tests: `make test`
-- Run a (local) lambda integration test: `make integration`
+- Run the kafka consumer: `make run`

pyproject.toml

@@ -1,19 +1,20 @@
 [tool.poetry]
-name = "example-consumer-python-sns"
+name = "example-consumer-python-kafka"
 version = "0.0.0"
-description = "Example Python AWS SNS Consumer"
+description = "Example Python Kafka Consumer"
 license = "Proprietary"
 authors = ["Mike Geeves <mike.geeves@bearly-compiling.com>"]
 readme = 'README.md'
-repository = "https://github.com/pactflow/example-consumer-python-sns"
-homepage = "https://github.com/pactflow/example-consumer-python-sns"
+repository = "https://github.com/pactflow/example-consumer-python-kafka"
+homepage = "https://github.com/pactflow/example-consumer-python-kafka"
 
 [build-system]
 requires = ["poetry-core>=1.0.0"]
 build-backend = "poetry.core.masonry.api"
 
 [tool.poetry.dependencies]
 python = "^3.11"
+confluent-kafka = "^2.6.0"
 
 [tool.poetry.dev-dependencies]
 pytest = "8.3.3"

run.py

@@ -0,0 +1,4 @@
+from src._kafka.product import consume_messages
+
+if __name__ == "__main__":
+    consume_messages()

src/_kafka/product.py

@@ -0,0 +1,57 @@
+#!/usr/bin/env python
+import logging
+import asyncio
+import json
+from confluent_kafka import Consumer
+from typing import Dict
+from src.product.product_service import receive_product_update
+logger = logging.getLogger(__name__)
+
+def message_handler(event):
+    loop = asyncio.get_event_loop()
+    return loop.run_until_complete(handler(event))
+
+async def handler(event: Dict):
+
+    logger.info(f'{event=}')
+
+    # Read the SNS message and pass the contents to the actual message handler
+    message: Dict = json.loads(event.value().decode('utf-8'))
+    logger.info(message)
+    results = await receive_product_update(message)
+    logger.info(results)
+
+    # Return the current size of the repository
+    return len(results.keys())
+
+
+def consume_messages():
+    config = {
+        'bootstrap.servers': 'localhost:9092',
+        'group.id':          'pactflow-example-consumer-python-kafka',
+    }
+
+    # Create Consumer instance
+    consumer = Consumer(config)
+
+    # Subscribe to topic
+    topic = "products"
+    consumer.subscribe([topic])
+
+    # Poll for new messages from Kafka and print them.
+    try:
+        while True:
+            msg = consumer.poll(1.0)
+            if msg is None:
+                print("Waiting...")
+            elif msg.error():
+                print("ERROR: %s".format(msg.error()))
+            else:
+                print("Consumed event from topic {topic}: value = {value:12}".format(
+                    topic=msg.topic(), value=msg.value().decode('utf-8')))
+                message_handler(msg)
+    except KeyboardInterrupt:
+        pass
+    finally:
+        # Leave group and commit final offsets
+        consumer.close()