Clean Rest API Error Messages with Python

A great practice in software development is clear error message communication.

Something like "An error occurred. Please try again later.", simply will not do because it's simply easier to be explicit.

Our goal here would be to ensure that a consistent interface is communicated to our client applications regardless of request.

Tools and frameworks like Ruby-on-Rails make this really easy with it's ActiveRecord implementation (Active record is literally a design pattern btw).

You'll find that in most "fully featured" frameworks with ORMs like Rails, Django, Sails, etc, there is some sort of implementation that enables a clean interface for transmitting error messages, but often we don't want to carry the bloat of these frameworks along.

Enter: types.

Types are great. They provide building blocks for framing tautologies around how our system functions and how objects interact.

My favorite typing framework has to be pydantic. It allows us to declare explicit types and validations for those types and we can use it anywhere. So-long having to rely on bulky frameworks for data validation support.

To show how pydantic will help us, let's say for example that we're trying to submit data from a rest api to a postgresql database.

Without using pydantic, if we're using an unopinionated api layer, (say fastapi), we would have to send sql explicity to our database service. Which is tough because we have to rely directly on our database response in order to communicate that in our response to the client. Tough. Especially with these being the database api errors that sqlalchemy returns.

Let's take a reminders application for example. Make sure you're using python 3.6+ for this.

We'll setup our pydantic schemas, data models, and crud functions here.

# database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "postgresql://USERNAME:PASSWORD@localhost/reminders"

engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

# models.py
from database import Base
from sqlalchemy import Column, DateTime, Integer, String


class Reminder(Base):
    __tablename__ = "reminders"
    id = Column(Integer, primary_key=True, index=True, nullable=False)
    due = Column(DateTime, nullable=False)
    title = Column(String, nullable=False)
    description = Column(String, nullable=False)

# schemas.py
from datetime import datetime

from pydantic import BaseModel


class ReminderBase(BaseModel):
    due: datetime
    title: str
    description: str


class ReminderCreate(ReminderBase):
    pass


class Reminder(ReminderBase):
    id: int

    class Config:
        orm_mode = True

# crud.py
from typing import List

import models
import schemas
from sqlalchemy.orm import Session


def get_reminders(db: Session,
                  skip: int = 0,
                  limit: int = 100) -> List[models.Reminder]:
    results = db.query(models.Reminder).offset(skip).limit(limit).all()
    db.commit()
    return results


def create_reminder(
        db: Session,
        reminder_create: schemas.ReminderCreate) -> models.Reminder:
    db_reminder = models.Reminder(**reminder_create.dict())
    db.add(db_reminder)
    db.commit()
    return db_reminder

# main.py
from typing import List

import crud
import models
import schemas
from database import SessionLocal, engine
from fastapi import Depends, FastAPI
from sqlalchemy.orm import Session

models.Base.metadata.create_all(bind=engine)

api = FastAPI()


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@api.post('/reminders', response_model=schemas.Reminder, tags=['reminder'])
def create_reminder(reminder_create: schemas.ReminderCreate,
                    db: Session = Depends(get_db)):
    return crud.create_reminder(db, reminder_create=reminder_create)


@api.get('/reminders',
         response_model=List[schemas.Reminder],
         tags=['reminder'])
def read_reminders(skip: int = 0,
                   limit: int = 100,
                   db: Session = Depends(get_db)):
    return crud.get_reminders(db, skip=skip, limit=limit)

# requirements.txt
fastapi
psycopg2
sqlalchemy
uvicorn

To note, please do not use this application in production as it's intended for instructional purposes only. For something more production ready, checkout this.

Make sure to create the database first. I'm using postgres from the cli so how you create your db will vary.

createdb reminders

Now setup your env and kickoff your local server ...

python3 -m venv env
source env/bin/activate
pip3 install -r requirements.txt
uvicorn main:api --reload

If all went well you should see ...

$ uvicorn main:api --reload
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [29526] using statreload
INFO:     Started server process [29528]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

Now if we make an empty post request to our service we should receive a nice 422 error and a great error message.

$ curl -X POST http://localhost:8000/reminders -d '{}' | jq
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100   249  100   247  100     2  35285    285 --:--:-- --:--:-- --:--:-- 35571
{
  "detail": [
    {
      "loc": [
        "body",
        "due"
      ],
      "msg": "field required",
      "type": "value_error.missing"
    },
    {
      "loc": [
        "body",
        "title"
      ],
      "msg": "field required",
      "type": "value_error.missing"
    },
    {
      "loc": [
        "body",
        "description"
      ],
      "msg": "field required",
      "type": "value_error.missing"
    }
  ]
}

You should see something like INFO: 127.0.0.1:59487 - "POST /reminders HTTP/1.1" 422 Unprocessable Entity in your server log stdout.

Now let's add in our fields.

$ curl -X POST http://localhost:8000/reminders -d \
     '{
         "title": "Take a nap",
         "description": "Before the party.",
         "due": "2020-08-01T12:00:00"
     }' | jq
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100   180  100    91  100    89   3956   3869 --:--:-- --:--:-- --:--:--  7826
{
  "due": "2020-08-01T12:00:00",
  "title": "Take a nap",
  "description": "Before the party.",
  "id": 1
}

Nice!

But let's see what happens when we put in some bad data.

$ curl -X POST http://localhost:8000/reminders -d \
    '{
        "title": "Take a nap",
        "description": "Before the party.",
        "due": "august 8th 2020 at noon"
    }' | jq
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100   190  100    97  100    93  16166  15500 --:--:-- --:--:-- --:--:-- 31666
{
  "detail": [
    {
      "loc": [
        "body",
        "due"
      ],
      "msg": "invalid datetime format",
      "type": "value_error.datetime"
    }
  ]
}

From the message we can see that our date-time format was invalid.

Hopefully this illustrates why utilizing libraries like pydantic are necessary in making error resolution easier.

If you were following along and something didn't go quite right, let me know; a twitter dm works just fine!