ValidMind Async Email Pattern

gistfile0.md

Async Email Pattern - ValidMind

Pattern for sending emails that follows best practices and is async by default.

Flow

Route/Service                    Celery Task                      Email Notification
────────────                     ───────────                      ──────────────────
1. Call task with .delay()  →   2. Fetch entities by CUID    →   3. Build template data
   Pass only CUIDs (strings)       Check org toggle                  Call send_email()

Step-by-Step

1. Define Celery Task (tasks/notifications/email.py)

@shared_task(bind=True)
def on_your_event(self, user_cuid, org_cuid, entity_cuid):
    """Celery task for your email notification."""
    with celery_db_session() as session:
        user = session.query(User).filter_by(cuid=user_cuid).first()
        org = session.query(Organization).filter_by(cuid=org_cuid).first()
        entity = session.query(YourEntity).filter_by(cuid=entity_cuid).first()

        if not user or not org or not entity:
            logger.warning(f"Could not find entities for email notification")
            return

        # Optional: check org-level toggle
        if not _is_email_event_enabled_for_org(session, org.id, "your_event_key"):
            logger.info("Email event disabled by org settings")
            return

        EmailNotification.on_your_event(user, org, entity)

2. Add Email Logic (notifications/email_notifications.py)

@classmethod
def on_your_event(cls, user, org, entity):
    """Send email when your event happens."""
    data = {
        "subject": f"Something happened in {org.name}",
        "org_name": org.name,
        "actor_name": user.full_name,
        "url": f"{get_site_url_for_user(user.email, org.id)}/path/to/{entity.cuid}",
        # ... other template vars
    }

    cls.send_email(
        data=data,
        to_emails=[entity.recipient_email],
        template_id=EmailTemplate.YOUR_EVENT_TEMPLATE_ID,
    )

3. Add Template ID (emails/email_sender.py)

class EmailTemplate(Enum):
    # ... existing
    YOUR_EVENT_TEMPLATE_ID = 15  # next number

# In PostMarkEmailSender.templates:
EmailTemplate.YOUR_EVENT_TEMPLATE_ID: "12345678",  # Postmark template ID

4. Export Task (tasks/notifications/__init__.py)

from .email import (
    # ... existing
    on_your_event,
)

5. Call from Route/Service

from tasks.notifications import email as email_async

# In your route handler:
email_async.on_your_event.delay(
    user.cuid,
    org.cuid, 
    entity.cuid
)

Key Rules

Rule	Why
Pass only CUIDs (strings)	Celery serializes args to JSON
Use celery_db_session()	Proper session lifecycle in worker
Fetch entities inside task	Avoids stale/detached SQLAlchemy objects
Check org toggle (optional)	Respects org email preferences
Use EmailTemplate enum	Consistent template management
Call .delay()	Async execution via Celery

Existing Email Tasks

Task	Toggle Key
on_inventory_model_created	inventory_model_created
on_organization_user_added	user_added_to_organization
on_model_user_deleted	model_stakeholder_deleted
on_model_user_added	model_stakeholder_added
on_user_invited	(always enabled)
on_model_finding_created	artifact_created
on_model_comment_added	inventory_model_documentation_comments
on_workflow_transition_user_action	workflow_transitions
on_workflow_transition_status_set	workflow_transitions
on_inventory_model_monitoring_breach	inventory_model_monitoring_breach
on_broadcast_notification	workflow_broadcast_notification