Adds celery==5.2.7

CHANGELOG.md

@@ -23,12 +23,14 @@ The types of changes are:
 * Subject Request Details page [#563](https://github.com/ethyca/fidesops/pull/563)
 * Restart Graph from Failure [#578](https://github.com/ethyca/fidesops/pull/578)
 * Redis SSL Support [#611](https://github.com/ethyca/fidesops/pull/611)
+* Celery as a dependency for use in the execution layer [#610](https://github.com/ethyca/fidesops/pull/610)
 * Cache and Surface Resume/Restart Instructions [#591](https://github.com/ethyca/fidesops/pull/591)
 
 ### Changed
 
 * Refactor auth and enable static file serving [#577](https://github.com/ethyca/fidesops/pull/577)
 
+
 ## [1.5.2](https://github.com/ethyca/fidesops/compare/1.5.1...1.5.2)
 
 ### Added

data/config/fidesops.toml

@@ -26,6 +26,8 @@ DRP_JWT_SECRET = "testdrpsecret"
 LOG_LEVEL = "DEBUG"
 
 [execution]
+CELERY_BROKER_URL = "redis://:testpassword@redis:6379"
+CELERY_RESULT_BACKEND = "redis://:testpassword@redis:6379"
 TASK_RETRY_COUNT = 0
 TASK_RETRY_DELAY = 1
 TASK_RETRY_BACKOFF = 1

docker-compose.yml

@@ -10,7 +10,7 @@ services:
     expose:
       - 8080
     healthcheck:
-      test: ["CMD", "curl", "-f", "http://0.0.0.0:8080/health"]
+      test: [ "CMD", "curl", "-f", "http://0.0.0.0:8080/health" ]
       interval: 30s
       timeout: 10s
       retries: 3
@@ -58,6 +58,25 @@ services:
     ports:
       - "0.0.0.0:6379:6379"
 
+  celery:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    command: celery -A fidesops.tasks beat -l info --logfile=/var/log/celery.log
+    depends_on:
+      fidesops:
+        condition: service_healthy
+      db:
+        condition: service_started
+      redis:
+        condition: service_started
+    restart: always
+    volumes:
+      - type: bind
+        source: ./
+        target: /fidesops
+        read_only: False
+
   docs:
     build:
       context: docs/fidesops/

fidesops.toml

@@ -29,11 +29,14 @@ DRP_JWT_SECRET = "secret"
 LOG_LEVEL = "INFO"
 
 [execution]
+CELERY_BROKER_URL = "redis://:testpassword@redis:6379"
+CELERY_RESULT_BACKEND = "redis://:testpassword@redis:6379"
+MASKING_STRICT = true
+REQUIRE_MANUAL_REQUEST_APPROVAL = false
 TASK_RETRY_COUNT = 0
 TASK_RETRY_DELAY = 1
 TASK_RETRY_BACKOFF = 1
-REQUIRE_MANUAL_REQUEST_APPROVAL = false
-MASKING_STRICT = true
 
 [root_user]
 ANALYTICS_OPT_OUT = false
+ANALYTICS_ID = "89187e1f107ebe56c35e4f864af83a90"

requirements.txt

@@ -1,6 +1,7 @@
 alembic==1.6.5
 APScheduler==3.8.0
 boto3~=1.18.14
+celery[pytest]==5.2.7
 click==8.1.3
 cryptography~=3.4.8
 dask==2021.10.0

src/fidesops/core/config.py

@@ -90,6 +90,8 @@ class ExecutionSettings(FidesSettings):
     TASK_RETRY_BACKOFF: int
     REQUIRE_MANUAL_REQUEST_APPROVAL: bool = False
     MASKING_STRICT: bool = True
+    CELERY_BROKER_URL: str
+    CELERY_RESULT_BACKEND: str
 
     class Config:
         env_prefix = "FIDESOPS__EXECUTION__"

src/fidesops/tasks/__init__.py

@@ -0,0 +1,9 @@
+from celery import Celery
+
+app = Celery("tasks")
+app.config_from_object("fidesops.core.config", namespace="EXECUTION")
+app.autodiscover_tasks(["fidesops.tasks", "fidesops.tasks.scheduled"])
+
+
+if __name__ == "__main__":
+    app.worker_main()

tests/conftest.py

@@ -155,3 +155,9 @@ def _build_jwt(webhook: PolicyPreWebhook) -> Dict[str, str]:
 @pytest.fixture(scope="session")
 def integration_config() -> MutableMapping[str, Any]:
     yield load_toml("fidesops-integration.toml")
+
+
+@pytest.fixture(autouse=True, scope="session")
+def celery_enable_logging():
+    """Turns on celery output logs."""
+    return True

tests/fixtures/application_fixtures.py

@@ -744,42 +744,59 @@ def privacy_requests(db: Session, policy: Policy) -> Generator:
 def _create_privacy_request_for_policy(
     db: Session,
     policy: Policy,
+    status: PrivacyRequestStatus = PrivacyRequestStatus.in_processing,
 ) -> PrivacyRequest:
+    data = {
+        "external_id": f"ext-{str(uuid4())}",
+        "requested_at": datetime(
+            2018,
+            12,
+            31,
+            hour=2,
+            minute=30,
+            second=23,
+            microsecond=916482,
+            tzinfo=timezone.utc,
+        ),
+        "status": status,
+        "origin": f"https://example.com/",
+        "policy_id": policy.id,
+        "client_id": policy.client_id,
+    }
+    if status != PrivacyRequestStatus.pending:
+        data["started_processing_at"] = datetime(
+            2019,
+            1,
+            1,
+            hour=1,
+            minute=45,
+            second=55,
+            microsecond=393185,
+            tzinfo=timezone.utc,
+        )
     return PrivacyRequest.create(
         db=db,
-        data={
-            "external_id": f"ext-{str(uuid4())}",
-            "started_processing_at": datetime(
-                2019,
-                1,
-                1,
-                hour=1,
-                minute=45,
-                second=55,
-                microsecond=393185,
-                tzinfo=timezone.utc,
-            ),
-            "requested_at": datetime(
-                2018,
-                12,
-                31,
-                hour=2,
-                minute=30,
-                second=23,
-                microsecond=916482,
-                tzinfo=timezone.utc,
-            ),
-            "status": PrivacyRequestStatus.in_processing,
-            "origin": f"https://example.com/",
-            "policy_id": policy.id,
-            "client_id": policy.client_id,
-        },
+        data=data,
     )
 
 
 @pytest.fixture(scope="function")
 def privacy_request(db: Session, policy: Policy) -> PrivacyRequest:
-    privacy_request = _create_privacy_request_for_policy(db, policy)
+    privacy_request = _create_privacy_request_for_policy(
+        db,
+        policy,
+    )
+    yield privacy_request
+    privacy_request.delete(db)
+
+
+@pytest.fixture(scope="function")
+def privacy_request_status_pending(db: Session, policy: Policy) -> PrivacyRequest:
+    privacy_request = _create_privacy_request_for_policy(
+        db,
+        policy,
+        PrivacyRequestStatus.pending,
+    )
     yield privacy_request
     privacy_request.delete(db)
 

tests/tasks/test_celery.py

@@ -0,0 +1,10 @@
+def test_create_task(celery_app, celery_worker):
+    @celery_app.task
+    def multiply(x, y):
+        return x * y
+
+    # Force `celery_app` to register our new task
+    # See: https://github.com/celery/celery/issues/3642#issuecomment-369057682
+    celery_worker.reload()
+    assert multiply.run(4, 4) == 16
+    assert multiply.delay(4, 4).get(timeout=10) == 16