Skip to content

Instantly share code, notes, and snippets.

Created July 9, 2021 08:55
Show Gist options
  • Save RamiAwar/9b59d7f111c30fa25be2169fc4e88b50 to your computer and use it in GitHub Desktop.
Save RamiAwar/9b59d7f111c30fa25be2169fc4e88b50 to your computer and use it in GitHub Desktop.
def validate_unwrap(path_model=None, query_model=None, body_model=None):
"""Validates a Sanic route handler using path, query and body pydantic models.
This decorator would automatically validate the request path/query/body models and
plug in instances of the pydantic models as additional arguments to the route handler.
from pydantic import BaseModel, validator
class CustomerIn(BaseModel):
first_name: str
last_name: str
# /customer?first_name=X&last_name=Y
def get_customer_by_name(request, customer): <-- customer here would be a CustomerIn instance
this is automatically unwrapped by the decorator added as an extra argument here
return db.customer.get(first_name=customer.first_name, last_name=customer.last_name)
class Company(BaseModel):
company: str
def company_must_be_valid(cls, v):
if not db.get(v):
raise ValueError("Invalid company")
return v
# /<company>/customer?first_name=X&last_name=Y
@validate_unwrap(path_model=Company, query_model=CustomerIn)
def get_company_customer(request, company: Company, customer: CustomerIn): <-- here we use path model to also get company
validated before reaching the route handler
!! Note that company here is not a string, but a
Company instance. This is different from our old code.
return db.customer.get(, first_name=customer.first_name, last_name=customer.last_name)
Same idea for body parameters, where body would be a json object.
path_model (pydantic.BaseModel, optional): Path model for validating path parameters. Defaults to None.
query_model (pydantic.BaseModel, optional): Query model for validating query parameters. Defaults to None.
body_model (pydantic.BaseModel, optional): Body model for validating body parameters. Defaults to None.
def decorate(func):
async def wrapper(request, **path_params):
models = []
if path_model:
# Validate path parameters
path = path_model(**path_params)
except ValidationError as e:
raise error.BadRequest(e)
if query_model:
# Validate query parameters
query = extract_request_arguments(request)
query = query_model(**query)
except ValidationError as e:
raise error.BadRequest(e)
if body_model:
# Decode json payload
body = json.loads(request.body.decode("utf-8"))
except json.decoder.JSONDecodeError:
raise error.InvalidJSON()
# Validate payload
body = body_model(**body)
except ValidationError as e:
raise error.BadRequest(e)
return await func(request, *models)
return wrapper
return decorate
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment