Merge pull request #22

Version 1.1.2

acacore/__version__.py

@@ -1 +1 @@
-__version__ = "1.1.1"
+__version__ = "1.1.2"

acacore/reference_files/__init__.py

@@ -1,2 +1,2 @@
-from .ref_files import get_actions
-from .ref_files import get_custom_signatures
+from .get import get_actions
+from .get import get_custom_signatures

acacore/reference_files/ref_files.py → acacore/reference_files/get.py

@@ -1,7 +1,7 @@
 from functools import lru_cache
-from http.client import HTTPException
 from http.client import HTTPResponse
-from urllib import request
+from urllib.error import HTTPError
+from urllib.request import urlopen
 
 from pydantic import TypeAdapter
 from yaml import load
@@ -10,26 +10,25 @@
 from acacore.models.reference_files import Action
 from acacore.models.reference_files import CustomSignature
 
-actions_url: str = "https://raw.githubusercontent.com/aarhusstadsarkiv/reference-files/main/fileformats.yml"
-custom_signatures_url: str = (
-    "https://raw.githubusercontent.com/aarhusstadsarkiv/reference-files/main/custom_signatures.json"
-)
+download_url: str = "https://github.com/aarhusstadsarkiv/reference-files/releases/latest/download/"
+actions_file: str = "fileformats.yml"
+custom_signatures_file: str = "custom_signatures.json"
 
 
 @lru_cache
-def _get_actions() -> dict[str, Action]:
-    response: HTTPResponse = request.urlopen(actions_url)
+def _get_actions(url: str) -> dict[str, Action]:
+    response: HTTPResponse = urlopen(url)
     if response.getcode() != 200:
-        raise HTTPException(response.getcode())
+        raise HTTPError(url, response.getcode(), "", response.headers, response)
 
     return TypeAdapter(dict[str, Action]).validate_python(load(response.read(), Loader))
 
 
 @lru_cache
-def _get_custom_signatures() -> list[CustomSignature]:
-    response: HTTPResponse = request.urlopen(custom_signatures_url)
+def _get_custom_signatures(url: str) -> list[CustomSignature]:
+    response: HTTPResponse = urlopen(url)
     if response.getcode() != 200:
-        raise HTTPException(response.getcode())
+        raise HTTPError(url, response.getcode(), "", response.headers, response)
 
     return TypeAdapter(list[CustomSignature]).validate_json(response.read())
 
@@ -46,10 +45,17 @@ def get_actions(use_cache: bool = True) -> dict[str, Action]:
     Returns:
         dict[str, Action]: A dictionary with PUID keys and Action values.
 
+    Raises:
+        urllib.error.HTTPError: If there is an issue with the request.
+
     See Also:
         https://github.com/aarhusstadsarkiv/reference-files/blob/main/actions.yml
     """
-    return _get_actions() if use_cache else _get_actions.__wrapped__()
+    return (
+        _get_actions(f"{download_url.rstrip('/')}/{actions_file.lstrip('/')}")
+        if use_cache
+        else _get_actions.__wrapped__()
+    )
 
 
 def get_custom_signatures(use_cache: bool = True) -> list[CustomSignature]:
@@ -62,9 +68,16 @@ def get_custom_signatures(use_cache: bool = True) -> list[CustomSignature]:
         use_cache (bool): Use cached data if True, otherwise fetch data regardless of cache status
 
     Returns:
-        list[CustomSignature]: A list of CustomSignature objects
+        list[CustomSignature]: A list of CustomSignature objects.
+
+    Raises:
+        urllib.error.HTTPError: If there is an issue with the request.
 
     See Also:
         https://github.com/aarhusstadsarkiv/reference-files/blob/main/custom_signatures.json
     """
-    return _get_custom_signatures() if use_cache else _get_custom_signatures.__wrapped__()
+    return (
+        _get_custom_signatures(f"{download_url.rstrip('/')}/{custom_signatures_file.lstrip('/')}")
+        if use_cache
+        else _get_custom_signatures.__wrapped__()
+    )

acacore/utils/helpers.py

@@ -1,27 +1,33 @@
 from types import TracebackType
 from typing import Optional
+from typing import Sequence
 from typing import Type
 
 
 class ExceptionManager:
     """
     A context manager class that catches specified exceptions and stores the exception and traceback for later use.
 
+    Exceptions whose class is explicitly declared in the 'catch' argument are always caught,
+    even if they subclass from classes passed int the 'allow' argument.
+
     Args:
         *catch (Type[BaseException]): Exception types that should be caught and not allowed to rise.
 
     Attributes:
         exception (Optional[BaseException]): The exception that was raised within the context, if any.
         traceback (Optional[TracebackType]): The traceback associated with the exception, if any.
         catch (tuple[Type[BaseException], ...]): Tuple of exceptions that should be caught instead of letting them rise.
+        allow (tuple[Type[BaseException], ...]): Tuple of exceptions that should be allowed to rise.
     """
 
-    __slots__ = ("exception", "traceback", "catch")
+    __slots__ = ("exception", "traceback", "catch", "allow")
 
-    def __init__(self, *catch: Type[BaseException]) -> None:
+    def __init__(self, *catch: Type[BaseException], allow: Optional[Sequence[Type[BaseException]]] = None) -> None:
         self.exception: Optional[BaseException] = None
         self.traceback: Optional[TracebackType] = None
         self.catch: tuple[Type[BaseException], ...] = catch
+        self.allow: tuple[Type[BaseException], ...] = tuple(allow or [])
 
     def __enter__(self) -> "ExceptionManager":
         return self
@@ -34,4 +40,10 @@ def __exit__(
     ) -> bool:
         self.exception = exc_val
         self.traceback = exc_tb
-        return any(issubclass(exc_type, e) for e in self.catch) if exc_type else False
+
+        if not exc_type:
+            return False
+
+        return any(issubclass(exc_type, e) for e in self.catch) and (
+            exc_type in self.catch or not any(issubclass(exc_type, e) for e in self.allow)
+        )

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "acacore"
-version = "1.1.1"
+version = "1.1.2"
 description = ""
 authors = ["Matteo Campinoti <[email protected]>"]
 license = "GPL-3.0"

tests/test_reference_files.py

@@ -1,10 +1,21 @@
-from acacore.reference_files import get_actions
-from acacore.reference_files import get_custom_signatures
+from urllib.error import HTTPError
+
+import pytest
+
+import acacore.reference_files as reference_files
 
 
 def test_actions():
-    assert get_actions()
+    assert reference_files.get_actions()
+    with pytest.raises(HTTPError) as error:
+        reference_files.get.actions_file = f"wrong/path/{reference_files.get.actions_file}"
+        reference_files.get_actions()
+    assert error.value.code == 404
 
 
 def test_custom_signatures():
-    assert get_custom_signatures()
+    assert reference_files.get_custom_signatures()
+    with pytest.raises(HTTPError) as error:
+        reference_files.get.custom_signatures_file = f"wrong/path/{reference_files.get.actions_file}"
+        reference_files.get_custom_signatures()
+    assert error.value.code == 404

tests/test_utils.py

@@ -49,13 +49,30 @@ def test_helpers():
     assert context.exception.code == 3
     assert context.traceback is not None
 
-    with pytest.raises(KeyboardInterrupt) as raises, ExceptionManager(Exception) as context:
+    with (
+        pytest.raises(KeyboardInterrupt) as raises,
+        ExceptionManager(Exception) as context,
+    ):
         raise KeyboardInterrupt
 
     assert isinstance(raises.value, KeyboardInterrupt)
     assert isinstance(context.exception, KeyboardInterrupt)
     assert context.traceback is not None
 
+    with (
+        pytest.raises(OSError) as raises,  # noqa: PT011
+        ExceptionManager(BaseException, allow=[OSError]) as context,
+    ):
+        raise FileNotFoundError
+
+    assert isinstance(raises.value, FileNotFoundError)
+    assert isinstance(context.exception, FileNotFoundError)
+
+    with ExceptionManager(BaseException, FileNotFoundError, allow=[OSError]) as context:
+        raise FileNotFoundError
+
+    assert isinstance(context.exception, FileNotFoundError)
+
     with ExceptionManager() as context:
         pass