"""Built-in automation action handlers. Six platform-level actions are registered here and are available to every tenant out of the box. Module-specific actions are registered by each app's own ``automations.py`` (auto-discovered via ``AppConfig.ready``). Registration order does not matter. All handlers use **local imports** so this module can be imported early (before all apps are fully ready) without causing circular-import errors. Params schema conventions -------------------------- Every ``params_schema`` is a JSON Schema ``"object"`` so the frontend no-code builder can render a dynamic form and validate before saving. Handler contract ----------------- Handlers receive an ``ActionContext`` and must: - Be idempotent (the executor may retry on partial failure). - Raise ``ValueError`` for unrecoverable bad-params (executor marks action as ``"error"``). - Let transient errors bubble as regular exceptions so Celery can retry. """ from __future__ import annotations import json import urllib.error import urllib.request from typing import TYPE_CHECKING import structlog from simorgh.apps.automation.registry import ActionContext, ActionSpec, register_action if TYPE_CHECKING: pass _log = structlog.get_logger("simorgh.automation.builtin") # --------------------------------------------------------------------------- # 1. notifications.send_notification # --------------------------------------------------------------------------- def _handle_send_notification(ctx: ActionContext) -> None: """Send a notification to one or more recipients via a template slug. Params ------ template : str Notification template ``kind`` (registered in the notifications app). recipients : list[int] List of User PKs to notify. context : dict, optional Extra variables merged into the template render context. organization_node_id : int, optional Org-node scope for the notification row. Defaults to None → the executor's calling context supplies it when available. """ from simorgh.apps.notifications.services import dispatch as notify_dispatch params = ctx.params template_kind: str = params.get("template", "") recipients = params.get("recipients", []) if not template_kind: raise ValueError("notifications.send_notification: 'template' param is required") if not recipients: raise ValueError("notifications.send_notification: 'recipients' list is empty") org_node_id: int | None = params.get("organization_node_id") notify_dispatch( template_kind, recipients=recipients, context=params.get("context", {}), tenant_id=ctx.tenant_id, organization_node_id=org_node_id, ) _log.info("automation.action.send_notification", rule_id=ctx.rule_id, recipients=recipients) _SPEC_SEND_NOTIFICATION = ActionSpec( key="notifications.send_notification", label="Send Notification", module="notifications", params_schema={ "type": "object", "properties": { "template": { "type": "string", "title": "Template kind", "description": "Notification kind slug registered in the notifications app.", }, "recipients": { "type": "array", "items": {"type": "integer"}, "title": "Recipient user IDs", }, "context": { "type": "object", "title": "Template context", "description": "Extra variables passed to the template renderer.", "additionalProperties": True, }, "organization_node_id": { "type": ["integer", "null"], "title": "Organization node ID", }, }, "required": ["template", "recipients"], "additionalProperties": False, }, handler=_handle_send_notification, ) # --------------------------------------------------------------------------- # 2. iam.assign_role # --------------------------------------------------------------------------- def _handle_assign_role(ctx: ActionContext) -> None: """Assign a tenant role to a user (idempotent — get_or_create). Params ------ role_code : str ``Role.code`` value within the rule's tenant. user_id : int PK of the user to assign the role to. """ from simorgh.apps.iam.models import Role, UserRole params = ctx.params role_code: str = params.get("role_code", "") user_id: int | None = params.get("user_id") if not role_code: raise ValueError("iam.assign_role: 'role_code' param is required") if not user_id: raise ValueError("iam.assign_role: 'user_id' param is required") role = Role.objects.filter( code=role_code, tenant_id=ctx.tenant_id, ).first() if role is None: # Try global (non-tenant) roles as a fallback. role = Role.objects.filter(code=role_code, tenant__isnull=True).first() if role is None: raise ValueError( f"iam.assign_role: role with code {role_code!r} not found " f"for tenant {ctx.tenant_id}" ) _assignment, created = UserRole.objects.get_or_create( user_id=user_id, role=role, ) _log.info( "automation.action.assign_role", rule_id=ctx.rule_id, role_code=role_code, user_id=user_id, created=created, ) _SPEC_ASSIGN_ROLE = ActionSpec( key="iam.assign_role", label="Assign Role", module="iam", params_schema={ "type": "object", "properties": { "role_code": { "type": "string", "title": "Role code", "description": "Code of the Role within the tenant.", }, "user_id": { "type": "integer", "title": "User ID", "description": "PK of the user to receive the role assignment.", }, }, "required": ["role_code", "user_id"], "additionalProperties": False, }, handler=_handle_assign_role, ) # --------------------------------------------------------------------------- # 3. workflow.fire_transition # --------------------------------------------------------------------------- def _handle_fire_transition(ctx: ActionContext) -> None: """Fire a workflow transition on an existing WorkflowInstance. Params ------ instance_id : int PK of the ``WorkflowInstance`` to advance. transition : str Name of the transition to fire (must exist in the definition). """ from django.contrib.auth import get_user_model from simorgh.apps.workflow.engine import fire_transition from simorgh.apps.workflow.models import WorkflowInstance params = ctx.params instance_id: int | None = params.get("instance_id") transition: str = params.get("transition", "") if not instance_id: raise ValueError("workflow.fire_transition: 'instance_id' param is required") if not transition: raise ValueError("workflow.fire_transition: 'transition' param is required") try: instance = WorkflowInstance.objects.get(pk=instance_id, tenant_id=ctx.tenant_id) except WorkflowInstance.DoesNotExist: raise ValueError( f"workflow.fire_transition: WorkflowInstance {instance_id} not found" ) from None actor = None if ctx.actor_id: User = get_user_model() actor = User.objects.filter(pk=ctx.actor_id).first() fire_transition(instance, transition, actor=actor) _log.info( "automation.action.fire_transition", rule_id=ctx.rule_id, instance_id=instance_id, transition=transition, ) _SPEC_FIRE_TRANSITION = ActionSpec( key="workflow.fire_transition", label="Fire Workflow Transition", module="workflow", params_schema={ "type": "object", "properties": { "instance_id": { "type": "integer", "title": "Workflow instance ID", }, "transition": { "type": "string", "title": "Transition name", "description": "Name of the transition as defined in the workflow definition.", }, }, "required": ["instance_id", "transition"], "additionalProperties": False, }, handler=_handle_fire_transition, ) # --------------------------------------------------------------------------- # 4. http.send_webhook # --------------------------------------------------------------------------- def _handle_send_webhook(ctx: ActionContext) -> None: """Dispatch an outbound HTTP request to an arbitrary URL. Uses the Python standard-library ``urllib`` so no extra dependencies are needed. A ``requests``-based implementation can replace this once ``requests`` is added to requirements. Params ------ url : str Target URL (must be https in production — no validation here so admins can use http in development). method : str, optional HTTP method. Default ``"POST"``. payload : dict, optional JSON-serialisable body. Sent with ``Content-Type: application/json``. headers : dict, optional Extra HTTP headers to merge into the request. timeout : int, optional Seconds before giving up. Default 15. """ params = ctx.params url: str = params.get("url", "") if not url: raise ValueError("http.send_webhook: 'url' param is required") method: str = params.get("method", "POST").upper() payload: dict = params.get("payload", {}) extra_headers: dict = params.get("headers", {}) timeout: int = int(params.get("timeout", 15)) body: bytes = json.dumps(payload).encode() req = urllib.request.Request(url, data=body, method=method) req.add_header("Content-Type", "application/json") for header_name, header_value in extra_headers.items(): req.add_header(str(header_name), str(header_value)) try: with urllib.request.urlopen(req, timeout=timeout) as resp: # noqa: S310 status_code = resp.status except urllib.error.HTTPError as exc: raise RuntimeError( f"http.send_webhook: HTTP {exc.code} from {url}" ) from exc except urllib.error.URLError as exc: raise RuntimeError( f"http.send_webhook: network error reaching {url}: {exc.reason}" ) from exc _log.info( "automation.action.send_webhook", rule_id=ctx.rule_id, url=url, method=method, status_code=status_code, ) _SPEC_SEND_WEBHOOK = ActionSpec( key="http.send_webhook", label="Send HTTP Webhook", module="http", params_schema={ "type": "object", "properties": { "url": { "type": "string", "format": "uri", "title": "URL", }, "method": { "type": "string", "enum": ["POST", "PUT", "PATCH", "GET"], "default": "POST", "title": "HTTP method", }, "payload": { "type": "object", "title": "Request body", "additionalProperties": True, }, "headers": { "type": "object", "title": "Extra headers", "additionalProperties": {"type": "string"}, }, "timeout": { "type": "integer", "minimum": 1, "maximum": 60, "default": 15, "title": "Timeout (seconds)", }, }, "required": ["url"], "additionalProperties": False, }, handler=_handle_send_webhook, ) # --------------------------------------------------------------------------- # 5. modules.enable_feature # --------------------------------------------------------------------------- def _handle_enable_feature(ctx: ActionContext) -> None: """Enable (install) a platform module for the rule's tenant. Maps to ``modules.services.enable_module``. Idempotent — enabling an already-active module is a no-op. Params ------ feature_code : str Module name / feature code as registered in the module registry. """ from simorgh.apps.modules.services import enable_module from simorgh.apps.tenants.models import Tenant params = ctx.params feature_code: str = params.get("feature_code", "") if not feature_code: raise ValueError("modules.enable_feature: 'feature_code' param is required") try: tenant = Tenant.objects.get(pk=ctx.tenant_id) except Tenant.DoesNotExist: raise ValueError( f"modules.enable_feature: tenant {ctx.tenant_id} not found" ) from None enable_module(tenant, feature_code) _log.info( "automation.action.enable_feature", rule_id=ctx.rule_id, feature_code=feature_code, tenant_id=ctx.tenant_id, ) _SPEC_ENABLE_FEATURE = ActionSpec( key="modules.enable_feature", label="Enable Module Feature", module="modules", params_schema={ "type": "object", "properties": { "feature_code": { "type": "string", "title": "Feature / module code", "description": "Module name as registered in the module registry.", }, }, "required": ["feature_code"], "additionalProperties": False, }, handler=_handle_enable_feature, ) # --------------------------------------------------------------------------- # 6. subscription.change_status # --------------------------------------------------------------------------- def _handle_change_status(ctx: ActionContext) -> None: """Change the subscription/lifecycle status of the rule's tenant. Maps ``new_status`` to ``Tenant.status``. Valid values are the members of ``tenants.models.TenantStatus`` (e.g. ``"active"``, ``"suspended"``, ``"cancelled"``). Params ------ new_status : str The target ``TenantStatus`` value. """ from simorgh.apps.tenants.models import Tenant, TenantStatus params = ctx.params new_status: str = params.get("new_status", "") if not new_status: raise ValueError("subscription.change_status: 'new_status' param is required") valid_statuses = {choice[0] for choice in TenantStatus.choices} if new_status not in valid_statuses: raise ValueError( f"subscription.change_status: invalid status {new_status!r}. " f"Valid values: {sorted(valid_statuses)}" ) updated = Tenant.objects.filter(pk=ctx.tenant_id).update(status=new_status) if not updated: raise ValueError( f"subscription.change_status: tenant {ctx.tenant_id} not found" ) _log.info( "automation.action.change_status", rule_id=ctx.rule_id, new_status=new_status, tenant_id=ctx.tenant_id, ) _SPEC_CHANGE_STATUS = ActionSpec( key="subscription.change_status", label="Change Subscription Status", module="subscription", params_schema={ "type": "object", "properties": { "new_status": { "type": "string", "enum": ["active", "suspended", "cancelled", "trial"], "title": "New status", }, }, "required": ["new_status"], "additionalProperties": False, }, handler=_handle_change_status, ) # --------------------------------------------------------------------------- # Registration — called once from events.py at app startup # --------------------------------------------------------------------------- def register_builtin_actions() -> None: """Register all built-in action specs into the ActionRegistry. Idempotent — safe to call multiple times (the registry ignores identical re-registration of the same spec object). """ for spec in ( _SPEC_SEND_NOTIFICATION, _SPEC_ASSIGN_ROLE, _SPEC_FIRE_TRANSITION, _SPEC_SEND_WEBHOOK, _SPEC_ENABLE_FEATURE, _SPEC_CHANGE_STATUS, ): register_action(spec)