feat: use a pydantic model to parse config file

Author: fbessou

config.ini.example

@@ -0,0 +1,8 @@
+[pulse]
+userid = "<username>"
+host = "pulse.mozilla.org"
+port = 5671
+exchange = "exchange/<username>/test"
+routing_key = ""
+queue = "queue/<username>/test"
+password = "<pulse-password>"

config.toml.example

@@ -5,7 +5,7 @@
 from kombu import Connection, Exchange, Queue
 from mozlog import commandline
 
-from git_hg_sync import config
+from git_hg_sync.config import Config, get_repos_config
 from git_hg_sync.pulse_worker import PulseWorker
 from git_hg_sync.repo_synchronizer import RepoSynchronyzer
 
@@ -19,21 +19,21 @@ def get_parser():
 
 def get_connection(config):
     return Connection(
-        hostname=config["host"],
-        port=config["port"],
-        userid=config["userid"],
-        password=config["password"],
+        hostname=config.host,
+        port=config.port,
+        userid=config.userid,
+        password=config.password,
         heartbeat=10,
         ssl=True,
     )
 
 
 def get_queue(config):
-    exchange = Exchange(config["exchange"], type="topic")
+    exchange = Exchange(config.exchange, type="topic")
     return Queue(
-        name=config["queue"],
+        name=config.queue,
         exchange=exchange,
-        routing_key=config["routing_key"],
+        routing_key=config.routing_key,
         exclusive=False,
     )
 
@@ -43,10 +43,11 @@ def main():
     commandline.add_logging_group(parser)
     args = parser.parse_args()
     logger = commandline.setup_logging("service", args, {"raw": sys.stdout})
-    pulse_config = config.get_pulse_config(HERE.parent / "config.ini")["pulse"]
+    config = Config.from_file(HERE.parent / "config.toml")
+    pulse_config = config.pulse
     connection = get_connection(pulse_config)
 
-    repos_config = config.get_repos_config(HERE.parent / "repos.json")
+    repos_config = get_repos_config(HERE.parent / "repos.json")
 
     queue = get_queue(pulse_config)
     repo_synchronyzer = RepoSynchronyzer(repos_config=repos_config)

git_hg_sync/__main__.py

@@ -1,15 +1,31 @@
-import configparser
+import pathlib 
+import tomllib
 import json
 
+import pydantic
 
-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(config_file_path)
-    return config
+class PulseConfig(pydantic.BaseModel):
+    userid: str
+    host: str
+    port: int
+    exchange: str
+    routing_key: str
+    queue: str
+    password: str
 
 
-def get_repos_config(repo_file_path):
+class Config(pydantic.BaseModel):
+    pulse: PulseConfig
+
+    @staticmethod
+    def from_file(file_path: pathlib.Path) -> "Config":
+        assert file_path.exists(), f"config file {file_path} doesn't exists"
+        with open(file_path, "rb") as config_file:
+            config = tomllib.load(config_file)
+        return Config(**config)
+
+
+def get_repos_config(repo_file_path: pathlib.Path):
     assert repo_file_path.exists(), f"config file {repo_file_path} doesn't exists"
     with open(repo_file_path) as f:
         repos = json.load(f)

git_hg_sync/config.py

@@ -7,7 +7,7 @@ name = "git-hg-sync"
 readme = "README.md"
 requires-python = ">=3.10"
 version = "0.1"
-dependencies = ['kombu', 'mozillapulse', 'GitPython', 'mozlog']
+dependencies = ['kombu', 'mozillapulse', 'GitPython', 'mozlog', "pydantic"]
 
 [tool.ruff]
 line-length = 100

pyproject.toml

@@ -3,7 +3,7 @@
 
 import kombu
 
-from git_hg_sync import config
+from git_hg_sync.config import Config 
 
 HERE = Path(__file__).parent
 
@@ -14,13 +14,13 @@ def send_pulse_message(pulse_config, payload):
     The Pulse message will be constructed from the specified payload
     and sent to the requested exchange.
     """
-    userid = pulse_config["userid"]
-    password = pulse_config["password"]
-    routing_key = pulse_config["routing_key"]
-    host = pulse_config["host"]
-    port = pulse_config["port"]
-    exchange = pulse_config["exchange"]
-    queue = pulse_config["queue"]
+    userid = pulse_config.userid
+    password = pulse_config.password
+    routing_key = pulse_config.routing_key
+    host = pulse_config.host
+    port = pulse_config.port
+    exchange = pulse_config.exchange
+    queue = pulse_config.queue
     print(f"connecting to pulse at {host}:{port} as {userid}")
 
     connection = kombu.Connection(
@@ -74,5 +74,5 @@ def send_pulse_message(pulse_config, payload):
         "user": "user",
         "push_json_url": "push_json_url",
     }
-    pulse_conf = config.get_pulse_config(HERE.parent / "config.ini")["pulse"]
-    send_pulse_message(pulse_conf, payload)
+    config = Config.from_file(HERE.parent / "config.toml")
+    send_pulse_message(config.pulse, payload)