Workflows

🧭 General idea

A workflow is a sequence of steps (WorkflowStep) with well‑defined inputs/outputs.
Defined in apps/workflows/collections/*.py.

Steps use WorkflowStepDataReference:

• Workflow.input("field") — take a value from workflow input
• step.output("field") — take a value from step output
• .map(...) — transform a list
• .merge(...) — merge data by key

📚 Workflow list

• active_subscribers
• source_param_subscribers
• channel_reports
• campaign_reports_delivery
• duplicate_pools
• update_campaign
• process_subscribers
• firebase_reports
• firebase_subscribers_deliveries
• reactivating_unsubscribed_users
• campaign_planner
• campaign_runner
• unsubscription_spike_monitoring
• subscription_drop_monitoring
• SlackNotificationWorkflow
• start_new_workflow (internal)

➕ How to add a new workflow

1. Create a collection
   apps/workflows/collections/<new_workflow>.py
   Define BaseWorkflow, name, input_model, steps via add_step(...).

2. Define functions
   Add classes to apps/workflows/functions/
   Each function must have name, service, specs.

3. Register the workflow
   Add it to WorkflowsStock._data in apps/workflows/collections/stock.py.

4. Allow start via API
   Add it to the workflows list in apps/api/v1/workflows/serializers.py (meta‑class WorkflowSerializerFieldsMeta).

5. LOCAL functions
   If service = LOCAL, add the class to LocalTasksConsumer._get_function_class() (apps/workflows/consumer.py).

6. Celery / scheduling (if needed)

   • create apps/workflows/tasks/collections/<task>.py
   • add a task in tasks/*.py
   • add schedule to CELERY_BEAT_SCHEDULE

7. Tests
   Add tests in tests/apps/workflow/ and/or tests/api/workflow/.