Python Mailing

A lightweight mailing package with SendGrid and Resend providers, plus optional asynchronous execution via Celery and RabbitMQ.

Target Python version: 3.12+

Quick start

1. Create and activate a virtual environment

• macOS/Linux:

  python3 -m venv .venv
  source .venv/bin/activate

• Windows (PowerShell):

  py -3 -m venv .venv
  .venv\Scripts\Activate.ps1

2. Install dependencies

• Development setup (with pre-commit and tooling):

  pip install -r requirements/local.txt

• Production/runtime setup (only what the app needs to run):

  pip install -r requirements/prod.txt

• Minimal/base (shared by both):

  pip install -r requirements/base.txt

  Or use the convenience file at project root:

  pip install -r requirements.txt

Mailing flow (async)

• services.py -> tasks.py -> mailing.py
• services.send_email enqueues a Celery task (mailing.tasks.send_email_task).
• tasks.send_email_task executes mailing.mailing.send_email which routes to the provider methods on Email.

If you need to call synchronously (e.g., from a script), you can directly call mailing.mailing.send_email.

Async examples

• Using the high-level services helper (recommended):

  from mailing.services import send_email
  from mailing.constants import PROVIDER_SENDGRID, PROVIDER_RESEND
  
  # Enqueue an async task; returns a Celery AsyncResult
  task = send_email(
      provider=PROVIDER_SENDGRID,  # or PROVIDER_RESEND
      subject="Hello",
      message="Plain text body",
      recipient_list=["user@example.com"],
      # from_email="no-reply@yourdomain.com",
      # html_content="<p>Hi!</p>",
      # api_key="..."  # optional override per call
  )
  
  # Optionally wait for the result (blocks until the task completes)
  result = task.get(timeout=30)
  print("Provider response:", result)

• Calling the Celery task directly:

  from mailing.tasks import send_email_task
  from mailing.constants import PROVIDER_RESEND
  
  # Enqueue with .delay()
  task = send_email_task.delay(
      provider=PROVIDER_RESEND,
      subject="Hello",
      message="Body",
      recipient_list=["user@example.com"],
  )
  
  # Or customize with apply_async (ETA, countdown, queue, etc.)
  task = send_email_task.apply_async(
      kwargs=dict(
          provider=PROVIDER_RESEND,
          subject="Report",
          message="See attached",
          recipient_list=["team@example.com"],
      ),
      countdown=10,  # run in 10 seconds
      queue="emails",
  )

Notes:

• The task returns whatever the provider call returns (or None on failure for SendGrid path); .get() will surface that value.
• Ensure RabbitMQ and the Celery worker are running (see Docker section) before enqueuing tasks.

Pre-commit hooks

We use pre-commit to enforce formatting and basic checks before each commit.

Install local tooling and set up hooks:

pip install -r requirements/local.txt
pre-commit install

Run hooks on all files anytime:

pre-commit run --all-files

Formatting is handled by Black (line length 100) and import ordering by isort, both configured in pyproject.toml.

Running tests

Pytest is configured. Run the test suite with:

pytest -q

Run with Docker (includes Celery + RabbitMQ)

You can use Docker to build and run the project, including a RabbitMQ broker and a Celery worker.

Prerequisites: Docker Desktop or a compatible Docker Engine installed and running.

Basic usage:

# Build images defined in docker-compose.yml
docker compose build

# Start the stack (RabbitMQ, app, and Celery worker)
docker compose up

• The app service runs pytest -q by default.
• RabbitMQ is exposed on ports 5672 and 15672 (management UI at http://localhost:15672, default creds guest/guest).
• Celery worker uses the app image and runs celery -A mailing.tasks.celery_app worker -l info.

Env vars

Set via your shell or .env loaded by Docker Compose. Important variables:

• SENDGRID_API_KEY
• SENDGRID_FROM_EMAIL
• RESEND_API_KEY
• CELERY_BROKER_URL (default: amqp://guest:guest@rabbitmq:5672//)
• CELERY_RESULT_BACKEND (default: rpc://)

Notes:

• Stop the stack with Ctrl+C (in the same terminal) or by running docker compose down in another terminal.

API reference

mailing.services.send_email

Enqueue an asynchronous task to send an email using the specified provider.

Signature:

from mailing.services import send_email

result = send_email(
    provider: str,
    subject: str,
    message: str,
    recipient_list: list[str],
    from_email: str | None = None,
    html_content: str | None = None,
    api_key: str | None = None,
)

Parameters:

• provider: One of mailing.constants.PROVIDER_SENDGRID or PROVIDER_RESEND.
• subject: Subject line.
• message: Plain-text body.
• recipient_list: List of recipient email addresses.
• from_email: Optional sender address. Defaults to settings.SENDGRID_FROM_EMAIL if not provided.
• html_content: Optional HTML content.
• api_key: Optional per-call API key override for the chosen provider.

Returns:

• Celery AsyncResult. Call .get() to wait for the provider response if needed.

Example:

from mailing.services import send_email
from mailing.constants import PROVIDER_SENDGRID

async_result = send_email(
    provider=PROVIDER_SENDGRID,
    subject="Hi",
    message="Body",
    recipient_list=["user@example.com"],
)
# Optionally block
response = async_result.get(timeout=30)

---

mailing.mailing.send_email

Low-level synchronous helper used internally (and by the Celery task) to send an email via the selected provider.

Prefer using mailing.services.send_email in application code to enqueue work asynchronously.

Signature:

from mailing.mailing import send_email

response = send_email(
    provider: str,
    subject: str,
    message: str,
    recipient_list: list[str],
    from_email: str | None = None,
    html_content: str | None = None,
    api_key: str | None = None,
)

Returns:

• Provider response object on success (SendGrid: Response; Resend: dict-like), or None on failure for the SendGrid path. Raises mailing.exceptions.ProviderNotConfigured for unknown providers.