refactor: improve exception handling

- clear exception context
- use TypeError instead of asserts

nyacme/config.py

@@ -1,8 +1,8 @@
 import logging
 import os
 import sys
-from typing import Any, Optional
 from pathlib import Path
+from typing import Any
 
 import tomllib
 
@@ -40,15 +40,30 @@ def config_parse_dict(raw_conf: dict[str, Any], key: str) -> dict[str, str]:
         sys.exit(1)
 
     for k, v in raw_conf[key].items():
-        assert isinstance(k, str), f'"{k}" is not a string'
-        assert isinstance(v, str), f'"{k}" value "{v}" is not a string'
+        if not isinstance(k, str):
+            raise TypeError(f'"{k}" is not a string')
+        if not isinstance(v, str):
+            raise TypeError(f'"{k}" value "{v}" is not a string')
 
     result: dict[str, str] = raw_conf[key]
 
     return result
 
+def config_parse_list(raw_conf: dict[str, Any], key: str) -> list[str]:
+    result: list[str] = []
 
-def read_config(path: Optional[str]) -> Config:
+    if not isinstance(raw_conf[key], list):
+        raise TypeError(f'"{key}" must be a list, not {type(raw_conf[key]).__name__}')
+
+    for item in raw_conf[key]:
+        if not isinstance(item, str):
+            raise TypeError(f'"{key}" list item must be a string')
+        result.append(item)
+
+    return result
+
+
+def read_config(path: str | None) -> Config:
     if not path:
         # should be here only when running from hook
         path = os.getenv('NYACME_CONFIG', '/etc/nyacme.toml')
@@ -67,10 +82,7 @@ def read_config(path: Optional[str]) -> Config:
 
     post_acquire = []
     if 'post_acquire' in raw_conf:
-        assert isinstance(raw_conf['post_acquire'], list), 'post_acquire is not a list'
-        for cmd in raw_conf['post_acquire']:
-            assert isinstance(cmd, str), 'post_acquire item has to be a string'
-            post_acquire.append(cmd)
+        post_acquire = config_parse_list(raw_conf, 'post_acquire')
 
     c.post_acquire = post_acquire
 
@@ -83,9 +95,6 @@ def read_config(path: Optional[str]) -> Config:
         log.error('missing "certificates"')
         sys.exit(1)
 
-    c.certificates = []
-    for cert in raw_conf['certificates']:
-        assert isinstance(cert, str), 'certificate should be a string'
-        c.certificates.append(cert)
+    c.certificates = config_parse_list(raw_conf, 'certificates')
 
     return c

nyacme/handlers/he.py

@@ -29,7 +29,7 @@ class HEHandler(Handler):
         except urllib.error.HTTPError as ex:
             self.log.error('cannot set the record %s: %s', record_name, ex)
             res = ex.fp.read().decode('utf-8')
-            raise Exception(res)
+            raise Exception(res) from ex
 
     def create(self, record_name: str, record_value: str) -> None:
         self.log.info('creating %s with value %s', record_name, record_value)

nyacme/handlers/hetzner.py

@@ -1,6 +1,6 @@
 import json
 import urllib.request
-from typing import Any, Optional
+from typing import Any
 
 from ..config import Config
 from .base import Handler
@@ -22,7 +22,7 @@ class HetznerHandler(Handler):
                 self.nameservers = zone['ns']
                 break
 
-    def fetch(self, url: str, data: Optional[Any] = None, **kwargs: str) -> Any:
+    def fetch(self, url: str, data: Any | None = None, **kwargs: str) -> Any:
         req = urllib.request.Request('https://dns.hetzner.com/api/v1' + url)
         req.add_header('Auth-API-Token', self.secret)
         method = 'GET'
@@ -39,9 +39,9 @@ class HetznerHandler(Handler):
             self.log.error('cannot %s %s: %s', req.method, url, ex)
             res = ex.fp.read().decode('utf-8')
             try:
-                raise Exception(json.loads(res)['error'])
+                raise Exception(json.loads(res)['error']) from None
             except json.JSONDecodeError:
-                raise Exception(res)
+                raise Exception(res) from None
 
     def create(self, record_name: str, record_value: str) -> None:
         self.remove(record_name)

nyacme/handlers/porkbun.py

@@ -17,9 +17,11 @@ class PorkbunHandler(Handler):
 
         self.nameservers = self.fetch(f'/domain/getNs/{self.zone}')['ns']
 
-    def fetch(self, url: str, data: dict[str, Any] = {}) -> Any:
+    def fetch(self, url: str, data: dict[str, Any] | None = None) -> Any:
         req = urllib.request.Request('https://api.porkbun.com/api/json/v3' + url)
 
+        if data is None:
+            data = {}
         data['apikey'] = self.apikey
         data['secretapikey'] = self.secretapikey
         req.data = json.dumps(data).encode('utf-8')
@@ -27,17 +29,17 @@ class PorkbunHandler(Handler):
         req.method = 'POST'
         try:
             with urllib.request.urlopen(req) as f:
-                data = json.load(f)
-                if data['status'] != 'SUCCESS':
-                    raise Exception(data['message'])
-                return data
+                res = json.load(f)
+                if res['status'] != 'SUCCESS':
+                    raise Exception(res['message'])
+                return res
         except urllib.error.HTTPError as ex:
             self.log.error('cannot %s %s: %s', req.method, url, ex)
             res = ex.fp.read().decode('utf-8')
             try:
                 raise Exception(json.loads(res)['message'])
             except json.JSONDecodeError:
-                raise Exception(res)
+                raise Exception(res) from None
 
     def create(self, record_name: str, record_value: str) -> None:
         self.remove(record_name)