Allows for custom email templates

src/dispatch/__init__.py

@@ -73,6 +73,7 @@
     )
     from dispatch.forms.type.models import FormsType  # noqa lgtm[py/unused-import]
     from dispatch.forms.models import Forms  # noqa lgtm[py/unused-import]
+    from dispatch.email_templates.models import EmailTemplates  # noqa lgtm[py/unused-import]
 
 
 except Exception:

src/dispatch/api.py

@@ -41,6 +41,7 @@
 from dispatch.project.views import router as project_router
 from dispatch.forms.views import router as forms_router
 from dispatch.forms.type.views import router as forms_type_router
+from dispatch.email_templates.views import router as email_template_router
 
 
 from dispatch.signal.views import router as signal_router
@@ -233,7 +234,7 @@ def get_organization_path(organization: OrganizationSlug):
 authenticated_organization_api_router.include_router(
     forms_type_router, prefix="/forms_type", tags=["forms_type"]
 )
-
+authenticated_organization_api_router.include_router(email_template_router, prefix="/email_template", tags=["email_template"])
 
 @api_router.get("/healthcheck", include_in_schema=False)
 def healthcheck():

src/dispatch/database/revisions/tenant/versions/2024-03-11_91bd05855ad1.py

@@ -0,0 +1,41 @@
+"""Adds email template table
+
+Revision ID: 91bd05855ad1
+Revises: 27e6558e26a8
+Create Date: 2024-03-11 10:40:08.313520
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision = "91bd05855ad1"
+down_revision = "27e6558e26a8"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "email_templates",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("email_template_type", sa.String(), nullable=True),
+        sa.Column("welcome_text", sa.String(), nullable=True),
+        sa.Column("welcome_body", sa.String(), nullable=True),
+        sa.Column("components", sa.String(), nullable=True),
+        sa.Column("enabled", sa.Boolean(), nullable=True),
+        sa.Column("project_id", sa.Integer(), nullable=True),
+        sa.Column("created_at", sa.DateTime(), nullable=True),
+        sa.Column("updated_at", sa.DateTime(), nullable=True),
+        sa.ForeignKeyConstraint(["project_id"], ["project.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("email_template_type", "project_id"),
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table("email_templates")
+    # ### end Alembic commands ###

src/dispatch/email_templates/enums.py

@@ -0,0 +1,5 @@
+from dispatch.enums import DispatchEnum
+
+
+class EmailTemplateTypes(DispatchEnum):
+    welcome = "Incident Welcome Email"

src/dispatch/email_templates/models.py

@@ -0,0 +1,49 @@
+from datetime import datetime
+from pydantic import Field
+from typing import Optional, List
+
+from sqlalchemy import Column, Integer, String, Boolean, UniqueConstraint
+
+from dispatch.database.core import Base
+from dispatch.models import DispatchBase, TimeStampMixin, PrimaryKey, Pagination, ProjectMixin
+from dispatch.project.models import ProjectRead
+
+
+class EmailTemplates(TimeStampMixin, ProjectMixin, Base):
+    __table_args__ = (UniqueConstraint("email_template_type", "project_id"),)
+    # Columns
+    id = Column(Integer, primary_key=True)
+    email_template_type = Column(String, nullable=True)
+    welcome_text = Column(String, nullable=True)
+    welcome_body = Column(String, nullable=True)
+    components = Column(String, nullable=True)
+    enabled = Column(Boolean, default=True)
+
+
+# Pydantic models
+class EmailTemplatesBase(DispatchBase):
+    email_template_type: Optional[str] = Field(None, nullable=True)
+    welcome_text: Optional[str] = Field(None, nullable=True)
+    welcome_body: Optional[str] = Field(None, nullable=True)
+    components: Optional[str] = Field(None, nullable=True)
+    enabled: Optional[bool]
+
+
+class EmailTemplatesCreate(EmailTemplatesBase):
+    project: Optional[ProjectRead]
+
+
+class EmailTemplatesUpdate(EmailTemplatesBase):
+    id: PrimaryKey = None
+
+
+class EmailTemplatesRead(EmailTemplatesBase):
+    id: PrimaryKey
+    project: Optional[ProjectRead]
+    created_at: Optional[datetime] = None
+    updated_at: Optional[datetime] = None
+
+
+class EmailTemplatesPagination(Pagination):
+    items: List[EmailTemplatesRead]
+    total: int

src/dispatch/email_templates/service.py

@@ -0,0 +1,78 @@
+import logging
+from typing import List, Optional
+
+from sqlalchemy.orm import Session
+
+from .models import EmailTemplates, EmailTemplatesUpdate, EmailTemplatesCreate
+from dispatch.project import service as project_service
+
+log = logging.getLogger(__name__)
+
+
+def get(*, email_template_id: int, db_session: Session) -> Optional[EmailTemplates]:
+    """Gets an email template by its id."""
+    return (
+        db_session.query(EmailTemplates)
+        .filter(EmailTemplates.id == email_template_id)
+        .one_or_none()
+    )
+
+
+def get_by_type(*, email_template_type: str, project_id: int, db_session: Session) -> Optional[EmailTemplates]:
+    """Gets an email template by its type."""
+    return (
+        db_session.query(EmailTemplates)
+        .filter(EmailTemplates.project_id == project_id)
+        .filter(EmailTemplates.email_template_type == email_template_type)
+        .filter(EmailTemplates.enabled == True)  # noqa
+        .first()
+    )
+
+
+def get_all(*, db_session: Session) -> List[Optional[EmailTemplates]]:
+    """Gets all email templates."""
+    return db_session.query(EmailTemplates)
+
+
+def create(*, email_template_in: EmailTemplatesCreate, db_session: Session) -> EmailTemplates:
+    """Creates email template data."""
+    project = project_service.get_by_name_or_raise(
+        db_session=db_session, project_in=email_template_in.project
+    )
+
+    email_template = EmailTemplates(
+        **email_template_in.dict(exclude={"project"}), project=project
+    )
+
+    db_session.add(email_template)
+    db_session.commit()
+    return email_template
+
+
+def update(
+    *,
+    email_template: EmailTemplates,
+    email_template_in: EmailTemplatesUpdate,
+    db_session: Session,
+) -> EmailTemplates:
+    """Updates an email template."""
+    new_template = email_template.dict()
+    update_data = email_template_in.dict(skip_defaults=True)
+
+    for field in new_template:
+        if field in update_data:
+            setattr(email_template, field, update_data[field])
+
+    db_session.commit()
+    return email_template
+
+
+def delete(*, db_session, email_template_id: int):
+    """Deletes an email template."""
+    email_template = (
+        db_session.query(EmailTemplates)
+        .filter(EmailTemplates.id == email_template_id)
+        .one_or_none()
+    )
+    db_session.delete(email_template)
+    db_session.commit()

src/dispatch/email_templates/views.py

@@ -0,0 +1,110 @@
+import logging
+from fastapi import APIRouter, HTTPException, status, Depends
+from pydantic.error_wrappers import ErrorWrapper, ValidationError
+
+from sqlalchemy.exc import IntegrityError
+
+from dispatch.auth.permissions import (
+    SensitiveProjectActionPermission,
+    PermissionsDependency,
+)
+from dispatch.database.core import DbSession
+from dispatch.auth.service import CurrentUser
+from dispatch.database.service import search_filter_sort_paginate, CommonParameters
+from dispatch.models import PrimaryKey
+from dispatch.exceptions import ExistsError
+
+from .models import EmailTemplatesRead, EmailTemplatesUpdate, EmailTemplatesPagination, EmailTemplatesCreate
+from .service import get, create, update, delete
+
+log = logging.getLogger(__name__)
+router = APIRouter()
+
+
+@router.get("", response_model=EmailTemplatesPagination)
+def get_email_templates(commons: CommonParameters):
+    """Get all email templates, or only those matching a given search term."""
+    return search_filter_sort_paginate(model="EmailTemplates", **commons)
+
+
+@router.get("/{email_template_id}", response_model=EmailTemplatesRead)
+def get_email_template(db_session: DbSession, email_template_id: PrimaryKey):
+    """Get an email template by its id."""
+    email_template = get(db_session=db_session, email_template_id=email_template_id)
+    if not email_template:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=[{"msg": "An email template with this id does not exist."}],
+        )
+    return email_template
+
+
+@router.post("", response_model=EmailTemplatesRead)
+def create_email_template(
+    db_session: DbSession,
+    email_template_in: EmailTemplatesCreate,
+    current_user: CurrentUser,
+):
+    """Create a new email template."""
+    try:
+        return create(
+            db_session=db_session, email_template_in=email_template_in
+        )
+    except IntegrityError:
+        raise ValidationError(
+            [
+                ErrorWrapper(
+                    ExistsError(msg="An email template with this type already exists."), loc="name"
+                )
+            ],
+            model=EmailTemplatesRead,
+        ) from None
+
+
+@router.put(
+    "/{email_template_id}",
+    response_model=EmailTemplatesRead,
+    dependencies=[Depends(PermissionsDependency([SensitiveProjectActionPermission]))],
+)
+def update_email_template(
+    db_session: DbSession,
+    email_template_id: PrimaryKey,
+    email_template_in: EmailTemplatesUpdate,
+):
+    """Update a search filter."""
+    email_template = get(db_session=db_session, email_template_id=email_template_id)
+    if not email_template:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=[{"msg": "An email template with this id does not exist."}],
+        )
+    try:
+        email_template = update(
+            db_session=db_session, email_template=email_template, email_template_in=email_template_in
+        )
+    except IntegrityError:
+        raise ValidationError(
+            [
+                ErrorWrapper(
+                    ExistsError(msg="An email template with this type already exists."), loc="name"
+                )
+            ],
+            model=EmailTemplatesUpdate,
+        ) from None
+    return email_template
+
+
+@router.delete(
+    "/{email_template_id}",
+    response_model=None,
+    dependencies=[Depends(PermissionsDependency([SensitiveProjectActionPermission]))],
+)
+def delete_email_template(db_session: DbSession, email_template_id: PrimaryKey):
+    """Delete an email template, returning only an HTTP 200 OK if successful."""
+    email_template = get(db_session=db_session, email_template_id=email_template_id)
+    if not email_template:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=[{"msg": "An email template with this id does not exist."}],
+        )
+    delete(db_session=db_session, email_template_id=email_template_id)