Fix EncodeError: kombu.exceptions.EncodeError: Object of type datetime is not JSON serializable in Celery

This error occurs when a Celery task argument or return value contains a type that cannot be serialized with the configured serializer (default: JSON). Python datetime objects, SQLAlchemy models, and other complex types are not JSON-serializable. Fix it by converting arguments to JSON-safe types like strings, dicts, or IDs before passing them to tasks.

Reading the Stack Trace

Here's what each line means:

• File "/app/app/routes.py", line 18, in schedule_reminder: The route handler passes a datetime object directly to the Celery task, which triggers serialization.
• File "/app/venv/lib/python3.12/site-packages/kombu/serialization.py", line 220, in dumps: Kombu attempts to serialize the task arguments using the configured serializer (JSON by default).
• kombu.exceptions.EncodeError: Object of type datetime is not JSON serializable: Python's json module cannot encode datetime objects. They must be converted to ISO format strings first.

Common Causes

from datetime import datetime

@app.route('/remind', methods=['POST'])
def schedule_reminder():
    reminder_time = datetime.utcnow()
    send_reminder.delay(user.id, reminder_time)  # datetime not serializable

send_welcome_email.delay(user)  # User object, not user.id

@celery.task
def get_report():
    return {'generated_at': datetime.utcnow(), 'data': [...]}  # datetime in return

The Fix

Convert datetime objects to ISO 8601 strings with .isoformat() before passing them to Celery tasks, then parse them back with datetime.fromisoformat() inside the task. The same pattern applies to any non-serializable type: pass primitive values or IDs, not complex objects.

@app.route('/remind', methods=['POST'])
def schedule_reminder():
    reminder_time = datetime.utcnow()
    send_reminder.delay(user.id, reminder_time)

@app.route('/remind', methods=['POST'])
def schedule_reminder():
    reminder_time = datetime.utcnow().isoformat()
    send_reminder.delay(user.id, reminder_time)

@celery.task
def send_reminder(user_id, reminder_time_iso):
    reminder_time = datetime.fromisoformat(reminder_time_iso)
    # ... process reminder

Testing the Fix

import pytest
from datetime import datetime
from app.tasks import send_reminder
from app import create_app

@pytest.fixture
def app():
    app = create_app()
    app.config['CELERY_ALWAYS_EAGER'] = True
    app.config['TESTING'] = True
    return app

def test_task_accepts_iso_string(app):
    with app.app_context():
        result = send_reminder(1, datetime.utcnow().isoformat())
        assert result is not None

def test_task_rejects_raw_datetime(app):
    # Verify that passing a raw datetime raises during serialization
    with pytest.raises(Exception):
        send_reminder.delay(1, datetime.utcnow())

def test_iso_roundtrip():
    original = datetime(2026, 4, 10, 14, 30, 0)
    iso = original.isoformat()
    parsed = datetime.fromisoformat(iso)
    assert original == parsed

Run your tests:

pytest tests/ -v

Pushing Through CI/CD

git checkout -b fix/celery-serialization-datetime,git add app/routes.py app/tasks.py tests/test_tasks.py,git commit -m "fix: convert datetime to ISO string before passing to Celery task",git push origin fix/celery-serialization-datetime

Your CI config should look something like this:

name: CI
on:
  pull_request:
    branches: [main]
jobs:
  test:
    runs-on: ubuntu-latest
    services:
      redis:
        image: redis:7-alpine
        ports:
          - 6379:6379
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'
          cache: 'pip'
      - run: pip install -r requirements.txt
      - run: pytest tests/ -v --tb=short
        env:
          CELERY_BROKER_URL: redis://localhost:6379/0

The Full Manual Process: 18 Steps

Here's every step you just went through to fix this one bug:

1. Notice the error alert or see it in your monitoring tool
2. Open the error dashboard and read the stack trace
3. Identify the file and line number from the stack trace
4. Open your IDE and navigate to the file
5. Read the surrounding code to understand context
6. Reproduce the error locally
7. Identify the root cause
8. Write the fix
9. Run the test suite locally
10. Fix any failing tests
11. Write new tests covering the edge case
12. Run the full test suite again
13. Create a new git branch
14. Commit and push your changes
15. Open a pull request
16. Wait for code review
17. Merge and deploy to production
18. Monitor production to confirm the error is resolved

Total time: 30-60 minutes. For one bug.

pip install bugstack

import bugstack

bugstack.init(api_key=os.environ["BUGSTACK_API_KEY"])

Deploying the Fix (Manual Path)

1. Run pytest locally to confirm tasks accept serialized arguments.
2. Open a pull request with the serialization fix.
3. Wait for CI checks to pass on the PR.
4. Have a teammate review and approve the PR.
5. Merge to main and verify tasks execute correctly in staging.

Frequently Asked Questions