lockfale-connectors 1.5.0

Connectors for Kafka, MQTT, and PostgreSQL. These connectors are specifically for use in the Cyberpartner applications.

Cyberpartner Connectors

Connectors Package

This repository contains a collection of connectors for Kafka, MQTT, and PostgreSQL that can be used across any microservice in relation to Cyberpartner applications.

There is specific enrichment and schema validation for Cyberpartner routing with Kafka.

The MQTT and PGSQL connectors are fairly generic and can be used in any project, but the kafka wrapper is not.

Connectors Included

• Kafka: Kafka producer and consumer connectors
• MQTT: MQTT client, publisher, and subscriber connectors
• PostgreSQL: PostgreSQL database connector

AWS CodeArtifact

Project Usage

# Configure poetry to use AWS CodeArtifact
export CODEARTIFACT_AUTH_TOKEN=$(aws codeartifact get-authorization-token \
  --domain YOUR_DOMAIN \
  --domain-owner YOUR_ACCOUNT_ID \
  --query authorizationToken \
  --output text)

export CODEARTIFACT_REPO_URL=$(aws codeartifact get-repository-endpoint \
  --domain YOUR_DOMAIN \
  --domain-owner YOUR_ACCOUNT_ID \
  --repository YOUR_REPO \
  --format pypi \
  --query repositoryEndpoint \
  --output text)

# Configure Poetry
poetry config repositories.codeartifact $CODEARTIFACT_REPO_URL
poetry config http-basic.codeartifact aws $CODEARTIFACT_AUTH_TOKEN

# Add the package to your project
poetry add lockfaleconnectors

PyPi

Publish

poetry build
poetry config pypi-token.pypi <your-token-here>
poetry publish --build

Project Usage

poetry add lockfaleconnectors

Kafka

from lockfale_connectors.lf_kafka.kafka_consumer import KafkaConsumer
from lockfale_connectors.lf_kafka.kafka_producer import KafkaProducer

if __name__ == "__main__":
    consumer = KafkaConsumer("kafka:9092", ["list", "of", "topics"], "group-id")
    producer_client = KafkaProducer(kafka_broker="kafka:9092")
    
    for message in consumer.consumer:
        data = json.loads(message) if isinstance(message, str) else message
        producer_client.send_message(source_topic="ingress-topic", destination_topic='dead-end-topic', message=data)

MQTT - Generic Message Handlers

Set username and password as env vars:

export MQTT_USERNAME=your-username
export MQTT_PASSWORD=your-password

from lockfale_connectors.mqtt.mqtt_subscriber import MQTTSubscriber
from lockfale_connectors.mqtt.mqtt_publisher import MQTTPublisher

if __name__ == "__main__":
    subscriber = MQTTSubscriber("subscriber-client-id")
    subscriber.connect()
    subscriber.start_listener()
    
    publisher = MQTTPublisher("publisher-client-id")
    publisher.connect(keep_alive=20)
    publisher.client.loop_start()
    publisher.publish("topic", "message")

MQTT - Specific Message Handlers

from lockfale_connectors.mqtt.mqtt_subscriber import MQTTSubscriber
from lockfale_connectors.mqtt.mqtt_publisher import MQTTPublisher

def custom_on_connect(self, client, userdata, flags, reason_code, properties):
    """Handles MQTT connection"""
    if reason_code.is_failure:
        logger.error(f"Failed to connect: {reason_code}")
        return

    subscription_list = [
        ("your/custom/topic", 1),
        ("your/custom/topic/wildcard/#", 1),
        ("your/custom/topic/dual/+/wildcard/#", 1),
    ]
    logger.info(f"Subscribing to:")
    logger.info(subscription_list)
    client.subscribe(subscription_list)

def custom_on_message(self, client, userdata, message: MQTTMessage):
    """Handles incoming MQTT messages"""
    if message.retain:
        logger.info("Skipping retained message")
        return

    data = json.loads(message.payload.decode("utf-8"))
    logger.info(f"{message.topic} | Received message: {data}")


if __name__ == "__main__":
    subscriber = MQTTSubscriber(
        f"subscriber-translate-mqtt-kafka-{args.service}",
        _on_connect=custom_on_connect,
        _on_message=custom_on_message,
    )
    
    subscriber.connect()
    subscriber.start_listener()
    
    publisher = MQTTPublisher("publisher-client-id")
    publisher.connect(keep_alive=20)
    publisher.client.loop_start()
    publisher.publish("topic", "message")

PostgreSQL

Required Env Vars:

export PG_DB_HOST=your-host
export PG_DB_CKC_POOL_PORT=your-port
export PG_DB_CKC_POOL=your-database
export PG_DB_CONNECTION_LIMIT=your-connection-limit
export PG_DB_USER=your-user
export PG_DB_PASSWORD=your-password

from lockfale_connectors.postgres.pgsql import PostgreSQLConnector

if __name__ == "__main__":
    pgsql = PostgreSQLConnector()
    
    # select df
    pgsql.select_dataframe("SELECT * FROM your_table")
    
    # select dict
    pgsql.select_dict("SELECT * FROM your_table")

    # insert
    INSERT_QRY = """
            INSERT INTO schema.your_table (name, is_active) 
            VALUES (%(name)s, 1) RETURNING id
        """
    params = {
        "name": "Alex",
    }
    record_id = pgsql.execute(query=INSERT_QRY, params=params)

TODO

• Use confluent-kafka
• Update Psycopg to v3.3
• Change env vars to be more generic