Complex Request Validation in FastAPI with Pydantic

Complex Request Validation in FastAPI with Pydantic
Python Development

Why should you care? 🤔

Pydantic + FastAPI gets along very well, and provide easy to code, type-annotation based basic validations for atomic types and complex types (created from atomic types).

What happens when these basic validations aren’t sufficient for you and you would like to do much more complex? 😲

Let’s dive in.. 🤿

Here’s how we’ll go about this, we are gonna construct a simple API endpoint for printing the date given the day number of the present year. However, if the day number is weekend day (Saturday or Sunday), return a request-validation error response.

I know the requirement sounds absurd 😛. However, I hope this requirement can help you understand how pydantic works.

There are two ways to go about this:

  • Method 1: Perform the complex validation along with all your other main logic.
  • Method 2: Perform the validation outside the place containing your main logic, in other words, delegating the complex validation to Pydantic .

Method 1: Performing validation along with main logic

import uvicorn
from fastapi import FastAPI, Path
from fastapi.responses import JSONResponse
import calendar
import datetime


def is_weekend(day_number: int) -> bool:
    ''' 
    date.isoweekday() - Return the day of the week as an integer, where Monday is 1 and Sunday is 7.
    For example, date(2002, 12, 4).isoweekday() == 3, a Wednesday
    '''
    year   = datetime.datetime.today().year 
    dt     = datetime.date(year, 1, 1) + datetime.timedelta(day_number - 1)
    number = dt.isoweekday()
    return number == 6 or number == 7


def daynumber_to_date(day_number: int) -> datetime.date:
    '''
    return datetime.date from start of actual year to days arg. 
    example: if actual year is 2021, daynumber_to_date(365) -> datetime.date(2021, 12, 31)
    '''
    year = datetime.datetime.today().year
    dt   = datetime.date(year, 1, 1) + datetime.timedelta(day_number - 1)
    return dt


def get_day_name(number: int) -> str:
    week_days = { 
        1: "Monday",
        2: "Tuesday",
        3: "Wednesday",
        4: "Thursday",
        5: "Friday",
        6: "Saturday",
        7: "Sunday"
    }
    return week_days[number]
   

app = FastAPI()


@app.get("/v1/api/date/{day_number}")
async def get_date(day_number: int = Path(..., gt=0, lt=367, title="Day number of the present year")):
    response = {}
    if (calendar.isleap(datetime.datetime.now().year) is False) and day_number > 365:
        return JSONResponse(
            status_code=422,
            content={"message": "day_number: Invalid day number",
                     "data": None},
        )
    elif (is_weekend(day_number) is True):
        return JSONResponse(
            status_code=422,
            content={"message": "day_number: Current day number is weekend day",
                     "data": None},
        )
    else:
        dt = daynumber_to_date(day_number)
        dn = dt.isoweekday()
        response['data'] = dt
        response['name'] = get_day_name(dn)
        response['message'] = "Successfully Computed!"
        return response


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

OUTPUT:

# 1. January 2021 - Friday (days 1)
# 3. January 2021 - Sunday (days 3)
# 17. September 2021 - Friday (days 260)
# 18. September 2021 - Saturday (days 261)

# run our fastapi api endpoint
~] python3 main.py

# testing our application:
~] curl -s http://127.0.0.1:8000/v1/api/date/1 | json_pp
{
   "data" : "2021-01-01",
   "message" : "Successfully Computed!",
   "name" : "Friday"
}

~] curl -s http://127.0.0.1:8000/v1/api/date/3 | json_pp
{
   "data" : null,
   "message" : "day_number: Current day number is weekend day"
}

~] curl -s http://127.0.0.1:8000/v1/api/date/260 | json_pp
{
   "name" : "Friday",
   "data" : "2021-09-17",
   "message" : "Successfully Computed!"
}

~] curl -s http://127.0.0.1:8000/v1/api/date/261 | json_pp
{
   "data" : null,
   "message" : "day_number: Current day number is weekend day"
}

~] curl -s http://127.0.0.1:8000/v1/api/date/586 | json_pp
{
   "detail" : [
      {
         "msg" : "ensure this value is less than 367",
         "loc" : [
            "path",
            "day_number"
         ],
         "type" : "value_error.number.not_lt",
         "ctx" : {
            "limit_value" : 367
         }
      }
   ]
}

~] curl -s http://127.0.0.1:8000/v1/api/date/abc | json_pp
{
   "detail" : [
      {
         "type" : "type_error.integer",
         "loc" : [
            "path",
            "day_number"
         ],
         "msg" : "value is not a valid integer"
      }
   ]
}

Take-away points from the above code:

  • Just the basic range validation , 0 to 366, is performed by Pydantic
  • Complex weekend day validation and leap year validation is performed by us at the place where the main logic of printing the date resides.
  • We return custom error responses for the two validations performed by us. For the validation error response by Pydantic, you will observe the response is quite different. However, this can be alleviated, read on!

Method 2: Performing entire validation with Pydantic

import uvicorn
from fastapi import FastAPI, Path, Depends
from fastapi.exceptions import RequestValidationError, ValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel, validator
import calendar
import datetime
import json



def is_weekend(day_number: int) -> bool:
    ''' 
    date.isoweekday() - Return the day of the week as an integer, where Monday is 1 and Sunday is 7.
    For example, date(2002, 12, 4).isoweekday() == 3, a Wednesday
    '''
    year   = datetime.datetime.today().year 
    dt     = datetime.date(year, 1, 1) + datetime.timedelta(day_number - 1)
    number = dt.isoweekday()
    return number == 6 or number == 7


def daynumber_to_date(day_number: int) -> datetime.date:
    '''
    return datetime.date from start of actual year to days arg. 
    example: if actual year is 2021, daynumber_to_date(365) -> datetime.date(2021, 12, 31)
    '''
    year = datetime.datetime.today().year
    dt   = datetime.date(year, 1, 1) + datetime.timedelta(day_number - 1)
    return dt


def get_day_name(number: int) -> str:
    week_days = { 
        1: "Monday",
        2: "Tuesday",
        3: "Wednesday",
        4: "Thursday",
        5: "Friday",
        6: "Saturday",
        7: "Sunday"
    }
    return week_days[number]
   

app = FastAPI()

@app.exception_handler(RequestValidationError)
@app.exception_handler(ValidationError)
async def validation_exception_handler(request, exc):
    print(f"OMG! The client sent invalid data!: {exc}")
    exc_json = json.loads(exc.json())
    response = {"message": [], "data": None}

    for error in exc_json:
        response['message'].append(error['loc'][-1]+f": {error['msg']}")

    return JSONResponse(response, status_code=422)


class DayNumber(BaseModel):
    day_number:  int = Path(..., gt=0, lt=367, title="Day number of the present year")

    @validator('day_number')
    def daynumber_validation(cls, day_number):
        if (calendar.isleap(datetime.now().year) is False) and day_number > 365:
            raise ValueError("Invalid Day Number")
        if is_weekend(day_number) is True:
            raise ValueError("The day-number must not be a weekend day")
        return day_number


@app.get("/v1/api/date/{day_number}")
async def get_date(day_number: DayNumber = Depends(DayNumber)):
    response = {}
    day_number = day_number.day_number
    print(day_number,"DAY")

    dt = daynumber_to_date(day_number)
    dn = dt.isoweekday()

    response['data'] = dt
    response['name'] = get_day_name(dn)
    response['message'] = "Successfully Computed!"
    return response


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

OUTPUT:

# 1. January 2021 - Friday (days 1)
# 3. January 2021 - Sunday (days 3)
# 17. September 2021 - Friday (days 260)
# 18. September 2021 - Saturday (days 261)

# run our fastapi api endpoint
~] python3 main.py

# testing our application:

~] curl -s http://127.0.0.1:8000/v1/api/date/1 | json_pp
{
   "name" : "Friday",
   "data" : "2021-01-01",
   "message" : "Successfully Computed!"
}

~] curl -s http://127.0.0.1:8000/v1/api/date/3 | json_pp
{
   "data" : null,
   "message" : "day_number: Current day number is weekend day"
}

curl -s http://127.0.0.1:8000/v1/api/date/260 | json_pp
{
   "name" : "Friday",
   "message" : "Successfully Computed!",
   "data" : "2021-09-17"
}

~] curl -s http://127.0.0.1:8000/v1/api/date/261 | json_pp
{
   "message" : "day_number: Current day number is weekend day",
   "data" : null
}

~] curl -s http://127.0.0.1:8000/v1/api/date/586 | json_pp
{
   "detail" : [
      {
         "msg" : "ensure this value is less than 367",
         "ctx" : {
            "limit_value" : 367
         },
         "type" : "value_error.number.not_lt",
         "loc" : [
            "path",
            "day_number"
         ]
      }
   ]
}

~] curl -s http://127.0.0.1:8000/v1/api/date/abc | json_pp
{
   "detail" : [
      {
         "type" : "type_error.integer",
         "loc" : [
            "path",
            "day_number"
         ],
         "msg" : "value is not a valid integer"
      }
   ]
}

Take-away points from the above code:

  • A Pydantic class DayNumber that contains our request parameters.
  • A validation function daynumber_validation for every parameter is marked with validator decorator inside the class. That’s our Hero
  • A ValueError exception is raised in case the validation fails. Which is then transformed to ValidationError exception by our dear Pydantic
  • An exception handler validation_exception_handler that intercepts RequestValidationError and ValidationError . It makes sure you’ll have a consistent error response format, unlike the first method.

SUBSCRIBE FOR NEW ARTICLES

@
comments powered by Disqus