feature: define a WorkerConfig

neo4j-app/neo4j_app/icij_worker/__init__.py

@@ -1,4 +1,4 @@
 from .app import ICIJApp
 from .task import Task, TaskError, TaskEvent, TaskResult, TaskStatus
-from .worker import Worker, Neo4jAsyncWorker
+from .worker import Worker, Neo4jAsyncWorker, WorkerConfig
 from .event_publisher import EventPublisher, Neo4jEventPublisher

neo4j-app/neo4j_app/icij_worker/worker/__init__.py

@@ -1,3 +1,4 @@
 from .worker import Worker
 from .neo4j import Neo4jAsyncWorker
+from .config import WorkerConfig
 from .process import ProcessWorkerMixin

neo4j-app/neo4j_app/icij_worker/worker/config.py

@@ -0,0 +1,15 @@
+from abc import ABC
+
+from pydantic import Field
+
+from neo4j_app.icij_worker.utils.registrable import RegistrableConfig
+
+
+class WorkerConfig(RegistrableConfig, ABC):
+    registry_key: str = Field(const=True, default="type")
+    log_level: str = "INFO"
+    type: str
+
+    class Config:
+        env_prefix = "ICIJ_WORKER_"
+        case_sensitive = False

neo4j-app/neo4j_app/tests/conftest.py

@@ -5,6 +5,7 @@
 import os
 import random
 import traceback
+from copy import copy
 from datetime import datetime
 from pathlib import Path
 from time import monotonic, sleep
@@ -653,3 +654,13 @@ async def sleep_for(
 @pytest.fixture(scope="session")
 def test_async_app(test_config: AppConfig) -> ICIJApp:
     return test_config.to_async_app()
+
+
+@pytest.fixture()
+def reset_env():
+    old_env = copy(dict(os.environ))
+    try:
+        yield
+    finally:
+        os.environ.clear()
+        os.environ.update(old_env)

neo4j-app/neo4j_app/tests/icij_worker/worker/test_config.py

@@ -0,0 +1,31 @@
+# pylint: disable=redefined-outer-name
+import os
+from typing import Optional
+
+import pytest
+
+from neo4j_app.icij_worker import WorkerConfig
+
+
+@pytest.fixture()
+def env_log_level(reset_env, request):  # pylint: disable=unused-argument
+    log_level = request.param
+    if log_level is not None:
+        os.environ["ICIJ_WORKER_LOG_LEVEL"] = log_level
+
+
+@pytest.mark.parametrize(
+    "env_log_level,expected_level",
+    [(None, "INFO"), ("DEBUG", "DEBUG"), ("INFO", "INFO")],
+    indirect=["env_log_level"],
+)
+def test_config_from_env(
+    env_log_level: Optional[str], expected_level: str  # pylint: disable=unused-argument
+):
+    # When
+    class WorkerImplConfig(WorkerConfig):
+        type: str = "worker_impl"
+
+    config = WorkerImplConfig()
+    # Then
+    assert config.log_level == expected_level