add email sending functionality with Django Ninja API

api/admin.py

@@ -0,0 +1,49 @@
+import smtplib
+
+from django.contrib import admin
+from django import forms
+from django.core.exceptions import ValidationError
+
+from api.models import MailProvider
+
+class MailProviderForm(forms.ModelForm):
+    host_password = forms.CharField(widget=forms.PasswordInput(render_value=True))
+
+    class Meta:
+        model = MailProvider
+        fields = "__all__"
+
+    def clean(self):
+        cleaned_data = super().clean()
+        host = cleaned_data.get("host")
+        port = cleaned_data.get("port")
+        user = cleaned_data.get("host_user")
+        password = cleaned_data.get("host_password")
+        use_tls = cleaned_data.get("use_tls")
+
+        if host and port and user and password:
+            try:
+                connection = smtplib.SMTP(host, port, timeout=10)
+                if use_tls:
+                    connection.starttls()
+                connection.login(user, password)
+                connection.quit()
+            except Exception as e:
+                raise ValidationError(f'Could not connect to the mail server: {e}')
+        return cleaned_data
+
+@admin.register(MailProvider)
+class MailProviderAdmin(admin.ModelAdmin):
+    form = MailProviderForm
+    list_display =  ("from_email", "host", "port", "use_tls")
+    search_fields = ("from_email", "host_user", "host")
+    list_filter = ("use_tls",)
+
+    fieldsets = (
+        (None, {
+            "fields": ("from_email", "host", "port", "use_tls")
+        }),
+        ("Authentication", {
+            "fields": ("host_user", "host_password")
+        }),
+    )

api/apps.py

@@ -0,0 +1,6 @@
+from django.apps import AppConfig
+
+
+class ApiConfig(AppConfig):
+    default_auto_field = 'django.db.models.BigAutoField'
+    name = 'api'

api/endpoints.py

@@ -0,0 +1,63 @@
+from typing import Optional, List
+
+from ninja import Router, Form, UploadedFile, File
+from ninja.errors import HttpError
+from ninja_jwt.authentication import JWTAuth
+
+from api.schema import NewMailMessageIn, NewBulkMailMessageIn
+from api.services import EmailService, MailProviderNotFound
+
+router = Router()
+email_service = EmailService()
+
+
+@router.post("/send-email", auth=JWTAuth())
+def send_email(request, data: NewMailMessageIn):
+    try:
+        email_service.send_email(data)
+        return {"status": "Email sent successfully"}
+    except MailProviderNotFound:
+        raise HttpError(404, "Mail provider not found")
+
+
+@router.post("/send-email-form", auth=JWTAuth())
+def send_email_form(request,
+                    from_email: str = Form(...),
+                    to: str = Form(...),
+                    subject: str = Form(...),
+                    body: str = Form(...),
+                    html_body: Optional[str] = Form(None),
+                    cc: Optional[str] = Form(None),
+                    bcc: Optional[str] = Form(None),
+                    attachments: List[UploadedFile] = File(default=[]),
+                    ):
+    try:
+        _attachments = []
+        for uploaded_file in attachments:
+            content = uploaded_file.read()
+            _attachments.append((uploaded_file.name, content, uploaded_file.content_type))
+
+        dto = NewMailMessageIn(
+            from_email=from_email,
+            to=to,
+            subject=subject,
+            body=body,
+            html_body=html_body,
+            cc=cc,
+            bcc=bcc,
+            attachments=_attachments
+        )
+
+        email_service.send_email(dto)
+        return {"status": "Email sent successfully"}
+    except MailProviderNotFound:
+        raise HttpError(404, "Mail provider not found")
+
+
+@router.post("/send-bulk-email", auth=JWTAuth())
+def send_bulk_email(request, data: NewBulkMailMessageIn):
+    try:
+        email_service.send_bulk_email(data)
+        return {"status": "Bulk email sent successfully"}
+    except MailProviderNotFound:
+        raise HttpError(404, "Mail provider not found")

api/migrations/0001_initial.py

@@ -0,0 +1,26 @@
+# Generated by Django 5.2.1 on 2025-05-17 11:55
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='MailProvider',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('host', models.URLField()),
+                ('port', models.IntegerField(default=587)),
+                ('host_user', models.CharField(max_length=255)),
+                ('host_password', models.CharField(max_length=255)),
+                ('use_tls', models.BooleanField(default=True)),
+                ('from_email', models.EmailField(max_length=254)),
+            ],
+        ),
+    ]

api/models.py

@@ -0,0 +1,12 @@
+from django.db import models
+
+class MailProvider(models.Model):
+    host = models.URLField()
+    port = models.IntegerField(default=587)
+    host_user = models.CharField(max_length=255)
+    host_password = models.CharField(max_length=255)
+    use_tls = models.BooleanField(default=True)
+    from_email = models.EmailField()
+
+    def __str__(self):
+        return f"{self.host} ({self.from_email})"

api/schema.py

@@ -0,0 +1,24 @@
+from typing import List, Optional
+
+from ninja import Schema, Form, UploadedFile, File
+
+
+class UserSchema(Schema):
+    username: str
+    is_authenticated: bool
+
+
+class NewMailMessageIn(Schema):
+    from_email: str
+    subject: str
+    body: str # plain text fallback
+    to: str
+    html_body: str | None = None
+    cc: str | None = None
+    bcc: str | None = None
+    attachments: List[tuple] = []
+
+
+class NewBulkMailMessageIn(Schema):
+    from_email: str
+    messages: List[NewMailMessageIn]

api/services.py

@@ -0,0 +1,64 @@
+from typing import Any
+
+from django.core.mail import get_connection, EmailMessage, EmailMultiAlternatives
+
+from api.models import MailProvider
+from api.schema import NewMailMessageIn, NewBulkMailMessageIn
+
+
+class MailProviderNotFound(Exception):
+    pass
+
+class EmailService:
+    def _get_mail_provider(self, from_email: str) -> MailProvider:
+        mail_provider = MailProvider.objects.filter(
+            from_email=from_email,
+        )
+
+        if not mail_provider.exists():
+            raise MailProviderNotFound()
+
+        return mail_provider.first()
+
+    def _get_connection(self, mail_provider: MailProvider):
+        return get_connection(
+            host=mail_provider.host,
+            port=mail_provider.port,
+            username=mail_provider.host_user,
+            password=mail_provider.host_password,
+            use_tls=mail_provider.use_tls,
+        )
+
+    def _build_email(self, dto: NewMailMessageIn, connection: Any) -> EmailMultiAlternatives:
+        email = EmailMultiAlternatives(
+            subject=dto.subject,
+            body=dto.body,
+            from_email=dto.from_email,
+            to=[dto.to],
+            cc=[dto.cc] if dto.cc else None,
+            bcc=[dto.bcc] if dto.bcc else None,
+            connection=connection,
+        )
+
+        if dto.html_body:
+            email.attach_alternative(dto.html_body, "text/html")
+
+        for attachment in dto.attachments:
+            email.attach(*attachment)
+
+        return email
+
+    def send_email(self, dto: NewMailMessageIn):
+        mail_provider = self._get_mail_provider(dto.from_email)
+        connection = self._get_connection(mail_provider)
+        email = self._build_email(dto, connection)
+        email.send()
+
+
+    def send_bulk_email(self, dto: NewBulkMailMessageIn):
+        mail_provider = self._get_mail_provider(dto.from_email)
+        connection = self._get_connection(mail_provider)
+        messages = [self._build_email(dto, connection) for dto in dto.messages]
+        connection.send_messages(messages)
+
+

api/tests.py

@@ -0,0 +1,3 @@
+from django.test import TestCase
+
+# Create your tests here.

api/urls.py

@@ -0,0 +1,14 @@
+from django.urls import path
+from ninja_extra import NinjaExtraAPI
+from ninja_jwt.controller import NinjaJWTDefaultController
+
+from .endpoints import router
+
+api = NinjaExtraAPI()
+api.register_controllers(NinjaJWTDefaultController)
+api.add_router("/", router)
+
+
+urlpatterns = [
+    path('', api.urls),
+]

api/views.py

@@ -0,0 +1,3 @@
+from django.shortcuts import render
+
+# Create your views here.