Broker class and necessary configuration

webhook_to_fedora_messaging/config.py

@@ -4,13 +4,15 @@
 
 from functools import cache
 from pathlib import Path
-
+import os
 from pydantic import BaseModel, ConfigDict, DirectoryPath
 from pydantic_settings import BaseSettings
 
 
 DEFAULT_CONFIG_FILE = _config_file = "/etc/webhook-to-fedora-messaging/webhook-to-fedora-messaging.cfg"
 TOP_DIR = Path(__file__).parent
+TEMPLATE = os.path.join(TOP_DIR, "fedmsg/conf/template_conf.toml")
+CONF_PATH = os.path.join(TOP_DIR, "fedmsg/conf/conf.toml") 
 
 
 class SQLAlchemyModel(BaseModel):

webhook_to_fedora_messaging/fedmsg/broker.py

@@ -0,0 +1,76 @@
+from fedora_messaging import api, config, message
+import os
+import toml
+from webhook_to_fedora_messaging.config import TEMPLATE, CONF_PATH
+
+
+
+
+
+class FedMsgBroker:
+
+    def __init__(self, connection_uri:str, key_file: str, cert_file: str, ca_cert: str) -> None:
+        self.__config = self.__generate_config(connection_uri, key_file, cert_file, ca_cert)
+        config.conf.setup_logging()
+        os.environ['FEDORA_MESSAGING_CONF"'] = CONF_PATH
+        self.__save_config()
+
+
+
+    def __generate_config(self, connection_uri:str, key_file: str, cert_file: str, ca_cert: str) -> None:
+        """
+            Generating a config file according to needs.
+
+            Args:
+                connection_uri: rabbitmq host to connect
+                key_file: path to key file
+                cert_file: path to cert file
+                ca_cert: path to ca cert file
+
+        """
+        conf = self.__get_template_config()
+        conf['amqp_url'] = connection_uri if connection_uri != "" else conf['amqp_url']
+        conf['keyfile'] = key_file
+        conf['certfile'] = cert_file
+        conf['ca_cert'] = ca_cert
+        return conf
+
+    def __save_config(self):
+        with open(CONF_PATH, "w") as file:
+            toml.dump(self.__config, file)
+
+
+
+
+    def set_config(self, value: dict):
+        """
+            Takes a dictionary as a value and sets the options in the dictionary keys to the corresponding values in the dictionary.
+
+
+        """
+
+        for key, setting in value.items():
+            self.__config[key] = setting
+        self.__save_config()
+
+
+    # Returns the default config.            
+    def __get_template_config(self):
+        return self.__get_config(TEMPLATE)
+
+
+    def __get_config(self, path: str):
+        with open(path, "r") as file:
+            return toml.load(file)
+
+
+    def publish(self, message: message.Message):
+        api.publish(message)
+
+
+    def consume(self):
+        api.consume(lambda message: print(message))
+
+
+
+

webhook_to_fedora_messaging/fedmsg/conf/conf.toml

@@ -0,0 +1,72 @@
+amqp_url = "localhost"
+callback = ""
+passive_declares = false
+publish_exchange = "amq.topic"
+topic_prefix = ""
+keyfile = "/etc/fedora-messaging/fedora-key.pem"
+certfile = "/etc/fedora-messaging/fedora-cert.pem"
+ca_cert = "/etc/fedora-messaging/cacert.pem"
+[[bindings]]
+queue = "my_queue"
+exchange = "amq.topic"
+routing_keys = [ "#",]
+
+[tls]
+ca_cert = ""
+keyfile = ""
+certfile = ""
+
+[client_properties]
+app = "webhook2fedmsg"
+
+[consumer_config]
+example_key = "for my consumer"
+
+[qos]
+prefetch_size = 0
+prefetch_count = 25
+
+[log_config]
+version = 1
+disable_existing_loggers = true
+
+[exchanges."amq.topic"]
+type = "topic"
+durable = true
+auto_delete = false
+
+[queues.my_queue]
+durable = true
+auto_delete = false
+exclusive = false
+
+[log_config.root]
+level = "ERROR"
+handlers = [ "console",]
+
+[exchanges."amq.topic".arguments]
+
+[queues.my_queue.arguments]
+
+[log_config.formatters.simple]
+format = "[%(levelname)s %(name)s] %(message)s"
+
+[log_config.handlers.console]
+class = "logging.StreamHandler"
+formatter = "simple"
+stream = "ext://sys.stdout"
+
+[log_config.loggers.fedora_messaging]
+level = "INFO"
+propagate = false
+handlers = [ "console",]
+
+[log_config.loggers.twisted]
+level = "INFO"
+propagate = false
+handlers = [ "console",]
+
+[log_config.loggers.pika]
+level = "WARNING"
+propagate = false
+handlers = [ "console",]

webhook_to_fedora_messaging/fedmsg/conf/template_conf.toml

@@ -0,0 +1,76 @@
+# A sample configuration for fedora-messaging. This file is in the TOML format.
+amqp_url = "amqps://fedora:@rabbitmq.fedoraproject.org/%2Fpublic_pubsub"
+callback = ""
+passive_declares = false
+publish_exchange = "amq.topic"
+topic_prefix = ""
+
+[tls]
+ca_cert = ""
+keyfile = ""
+certfile = ""
+
+[client_properties]
+app = "webhook2fedmsg"
+
+# If the exchange or queue name  has a "." in it, use quotes as seen here.
+[exchanges."amq.topic"]
+type = "topic"
+durable = true
+auto_delete = false
+arguments = {}
+
+[queues.my_queue]
+durable = true
+auto_delete = false
+exclusive = false
+arguments = {}
+
+# Note the double brackets below. To add another binding, add another
+# [[bindings]] section. To use multiple routing keys, just expand the list here.
+[[bindings]]
+queue = "my_queue"
+exchange = "amq.topic"
+routing_keys = ["#"]
+
+[consumer_config]
+example_key = "for my consumer"
+
+[qos]
+prefetch_size = 0
+prefetch_count = 25
+
+[log_config]
+version = 1
+disable_existing_loggers = true
+
+[log_config.formatters.simple]
+format = "[%(levelname)s %(name)s] %(message)s"
+
+[log_config.handlers.console]
+class = "logging.StreamHandler"
+formatter = "simple"
+stream = "ext://sys.stdout"
+
+[log_config.loggers.fedora_messaging]
+level = "INFO"
+propagate = false
+handlers = ["console"]
+
+# Twisted is the asynchronous framework that manages the TCP/TLS connection, as well
+# as the consumer event loop. When debugging you may want to lower this log level.
+[log_config.loggers.twisted]
+level = "INFO"
+propagate = false
+handlers = ["console"]
+
+# Pika is the underlying AMQP client library. When debugging you may want to
+# lower this log level.
+[log_config.loggers.pika]
+level = "WARNING"
+propagate = false
+handlers = ["console"]
+
+[log_config.root]
+level = "ERROR"
+handlers = ["console"]