refacto: use kombu mixin

Author: ogiorgis

git_hg_sync/config.py

@@ -1,21 +1,15 @@
 import configparser
 import json
-from pathlib import Path
 
-HERE = Path(__file__).parent
 
-
-def get_pulse_config():
+def get_pulse_config(config_file_path):
+    assert config_file_path.exists(), f"config file {config_file_path} doesn't exists"
     config = configparser.ConfigParser()
-    config.read(HERE.parent / "config.ini")
+    config.read(config_file_path)
     return config
 
 
-def get_repos_config():
-    with open(HERE.parent / "repos.json") as f:
+def get_repos_config(repo_file_path):
+    with open(repo_file_path) as f:
         repos = json.load(f)
     return repos
-
-
-if __name__ == "__main__":
-    get_pulse_config()

git_hg_sync/git_cinnabar.py

@@ -1,33 +1,26 @@
-from kombu import Connection, Exchange, Queue
-
-
-def get_connection(pulse_conf):
-    return Connection(
-        hostname=pulse_conf["host"],
-        port=pulse_conf["port"],
-        userid=pulse_conf["userid"],
-        password=pulse_conf["password"],
-        ssl=True,
-    )
-
-
-def get_consumer(pulse_conf, connection, callbacks):
-    exchange = pulse_conf["exchange"]
-    exchange = Exchange(exchange, type="topic")
-    exchange(connection).declare(
-        passive=True
-    )  # raise an error if exchange doesn't exist
-
-    queue = Queue(
-        name=pulse_conf["queue"],
-        exchange=exchange,
-        routing_key=pulse_conf["routing_key"],
-        durable=True,
-        exclusive=False,
-        auto_delete=False,
-    )
-
-    consumer = connection.Consumer(queue, auto_declare=False, callbacks=callbacks)
-    consumer.queues[0].queue_declare()
-    consumer.queues[0].queue_bind()
-    return consumer
+import logging
+
+from kombu.mixins import ConsumerMixin
+
+from git_hg_sync import sync_repos
+
+logger = logging.getLogger()
+
+
+class Worker(ConsumerMixin):
+
+    def __init__(self, connection, queue):
+        self.connection = connection
+        self.task_queue = queue
+
+    def get_consumers(self, Consumer, channel):
+        consumer = Consumer(
+            self.task_queue, auto_declare=False, callbacks=[self.on_task]
+        )
+        return [consumer]
+
+    def on_task(self, body, message):
+        logger.info(f"Received message: {body}")
+        message.ack()
+        raw_entity = body["payload"]
+        sync_repos.process(raw_entity)

git_hg_sync/pulse_consumer.py

@@ -0,0 +1,43 @@
+import logging
+from pathlib import Path
+
+from kombu import Connection, Exchange, Queue
+
+from git_hg_sync import config
+from git_hg_sync.pulse_consumer import Worker
+
+HERE = Path(__file__).parent
+
+
+def get_connection(config):
+    return Connection(
+        hostname=config["host"],
+        port=config["port"],
+        userid=config["userid"],
+        password=config["password"],
+        ssl=True,
+    )
+
+
+def get_queue(config):
+    exchange = Exchange(config["exchange"], type="topic")
+    return Queue(
+        name=config["queue"],
+        exchange=exchange,
+        routing_key=config["routing_key"],
+        exclusive=False,
+    )
+
+
+def main():
+    logging.basicConfig(level=logging.INFO)
+    pulse_config = config.get_pulse_config(HERE.parent / "config.ini")["pulse"]
+    connection = get_connection(pulse_config)
+    queue = get_queue(pulse_config)
+    with connection as conn:
+        worker = Worker(conn, queue)
+        worker.run()
+
+
+if __name__ == "__main__":
+    main()

git_hg_sync/service.py

@@ -0,0 +1,65 @@
+import logging
+from dataclasses import dataclass
+from pathlib import Path
+
+from git import Repo
+
+from git_hg_sync import config
+
+logger = logging.getLogger()
+HERE = Path(__file__).parent
+
+
+@dataclass
+class Push:
+    type: str
+    repo_url: str
+    heads: list[str]
+    commits: list[str]
+    time: int
+    pushid: int
+    user: str
+    push_json_url: str
+
+
+@dataclass
+class Tag:
+    type: str
+    repo_url: str
+    tag: str
+    commit: str
+    time: int
+    pushid: int
+    user: str
+    push_json_url: str
+
+
+def parse_entity(raw_entity):
+    if raw_entity["type"] == "push":
+        entity = Push(**raw_entity)
+    elif raw_entity["type"] == "tag":
+        entity = Tag(**raw_entity)
+    else:
+        raise AttributeError(f"unsupported type {raw_entity['type']}")
+    return entity
+
+
+def process(raw_entity):
+    entity = parse_entity(raw_entity)
+    repo_config = config.get_repos_config(HERE.parent / "repos.json").get(
+        entity.repo_url
+    )
+    if not repo_config:
+        logger.warning(f"repo {entity.repo_url} is not supported yet")
+    else:
+        repo = Repo(repo_config["clone"])
+        # fetch new commits
+        remote = repo.remote(repo_config["remote"])
+        remote.fetch()
+        # add commits to the good branch
+        for commit_sha in entity.commits:
+            logger.info(f"handle commit {commit_sha}")
+            repo.git.cherry_pick(commit_sha)
+        # push on good repo/branch
+        remote = repo.remote(repo_config["target"])
+        remote.push()

git_hg_sync/sync.py

@@ -0,0 +1,8 @@
+[pulse]
+userid = test
+host = localhost
+port = 5671
+exchange = exchange/test/test
+routing_key = #
+queue = queue/test/test
+password = pass_test

git_hg_sync/sync_repos.py

@@ -1,11 +1,13 @@
 import logging
 from datetime import datetime
+from pathlib import Path
 
 import kombu
 
 from git_hg_sync import config
 
 logger = logging.getLogger()
+HERE = Path(__file__).parent
 
 
 def send_pulse_message(pulse_config, payload):
@@ -76,5 +78,5 @@ def send_pulse_message(pulse_config, payload):
         "user": "user",
         "push_json_url": "push_json_url",
     }
-    pulse_conf = config.get_pulse_config()["pulse"]
+    pulse_conf = config.get_pulse_config(HERE.parent / "config.ini")["pulse"]
     send_pulse_message(pulse_conf, payload)