Support for Limited FIDES__CELERY__* Env Vars (#4980)

Add limited support for setting some celery values via Fides Env vars: FIDES__CELERY__TASK_ALWAYS_EAGER, FIDES__CELERY__EVENT_QUEUE_PREFIX, and FIDES__CELERY__TASK_DEFAULT_QUEUE.

CHANGELOG.md

@@ -23,6 +23,7 @@ The types of changes are:
 - Added option in FidesJS SDK to only disable notice-served API [#4965](https://github.com/ethyca/fides/pull/4965)
 - External ID support for consent management [#4927](https://github.com/ethyca/fides/pull/4927)
 - Added access and erasure support for the Greenhouse Harvest integration [#4945](https://github.com/ethyca/fides/pull/4945)
+- Support for Limited FIDES__CELERY__* Env Vars [#4980](https://github.com/ethyca/fides/pull/4980)
 - Implement sending emails via property-specific messaging templates [#4950](https://github.com/ethyca/fides/pull/4950)
 
 ### Changed

src/fides/config/__init__.py

@@ -16,6 +16,7 @@
 from fides.common.utils import echo_red
 
 from .admin_ui_settings import AdminUISettings
+from .celery_settings import CelerySettings
 from .cli_settings import CLISettings
 from .consent_settings import ConsentSettings
 from .credentials_settings import merge_credentials_environment
@@ -69,9 +70,7 @@ class FidesConfig(FidesSettings):
     admin_ui: AdminUISettings
     consent: ConsentSettings
     cli: CLISettings
-    celery: Dict = Field(
-        description="This section can be used to pass config vars to Celery directly.",
-    )
+    celery: CelerySettings
     credentials: Dict = Field(
         description="This is a special section that is used to store arbitrary key/value pairs to be used as credentials."
     )
@@ -151,6 +150,7 @@ def build_config(config_dict: Dict[str, Any]) -> FidesConfig:
 
     settings_map: Dict[str, Any] = {
         "admin_ui": AdminUISettings,
+        "celery": CelerySettings,
         "consent": ConsentSettings,
         "cli": CLISettings,
         "database": DatabaseSettings,
@@ -172,9 +172,6 @@ def build_config(config_dict: Dict[str, Any]) -> FidesConfig:
         credentials_dict=config_environment_dict
     )
 
-    # The Celery subsection is uniquely simple
-    settings_map["celery"] = config_dict.get("celery", {})
-
     fides_config = FidesConfig(**settings_map)
     return fides_config
 

src/fides/config/celery_settings.py

@@ -0,0 +1,26 @@
+from pydantic import Field
+
+from .fides_settings import FidesSettings
+
+ENV_PREFIX = "FIDES__CELERY__"
+
+
+class CelerySettings(FidesSettings):
+    """Configuration settings for Celery.  Only a small subset can be configured through Fides env vars"""
+
+    event_queue_prefix: str = Field(
+        default="fides_worker",
+        description="The prefix to use for event receiver queue names",
+    )
+    task_default_queue: str = Field(
+        default="fides",
+        description="The name of the default queue if a message has no route or no custom queue has been specified",
+    )
+    task_always_eager: bool = Field(
+        default=True,
+        description="If true, tasks are executed locally instead of being sent to the queue.  "
+        "If False, tasks are sent to the queue.",
+    )
+
+    class Config:
+        env_prefix = ENV_PREFIX

tests/ops/tasks/test_celery.py

@@ -5,7 +5,7 @@
 from sqlalchemy.pool import QueuePool
 
 from fides.api.tasks import DatabaseTask, _create_celery
-from fides.config import CONFIG, get_config
+from fides.config import CONFIG, CelerySettings, get_config
 
 
 @pytest.fixture
@@ -48,18 +48,24 @@ def get_virtualised_worker_config():
 
 def test_celery_default_config() -> None:
     config = get_config()
+    assert isinstance(config.celery, CelerySettings)
+    assert config.celery.task_always_eager
+    assert config.celery.event_queue_prefix == "fides_worker"
+    assert config.celery.task_default_queue == "fides"
+
     celery_app = _create_celery(config)
     assert celery_app.conf["broker_url"] == CONFIG.redis.connection_url
     assert celery_app.conf["result_backend"] == CONFIG.redis.connection_url
     assert celery_app.conf["event_queue_prefix"] == "fides_worker"
     assert celery_app.conf["task_default_queue"] == "fides"
+    assert celery_app.conf["task_always_eager"] is True
 
 
 def test_celery_config_override() -> None:
     config = get_config()
 
-    config.celery["event_queue_prefix"] = "overridden_fides_worker"
-    config.celery["task_default_queue"] = "overridden_fides"
+    config.celery.event_queue_prefix = "overridden_fides_worker"
+    config.celery.task_default_queue = "overridden_fides"
 
     celery_app = _create_celery(config=config)
     assert celery_app.conf["event_queue_prefix"] == "overridden_fides_worker"