pydantic_sanic_decorator.py

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.

    Usage:
        from pydantic import BaseModel, validator

        class CustomerIn(BaseModel):
            first_name: str
            last_name: str

        # /customer?first_name=X&last_name=Y
        @validate_unwrap(query_model=CustomerIn)
        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

            @validator("company")
            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(company=company.company, first_name=customer.first_name, last_name=customer.last_name)


        Same idea for body parameters, where body would be a json object.

    Args:
        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):
        @wraps(func)
        async def wrapper(request, **path_params):

            models = []

            if path_model:
                # Validate path parameters
                try:
                    path = path_model(**path_params)
                    models.append(path)
                except ValidationError as e:
                    raise error.BadRequest(e)

            if query_model:
                # Validate query parameters
                try:
                    query = extract_request_arguments(request)
                    query = query_model(**query)
                    models.append(query)
                except ValidationError as e:
                    raise error.BadRequest(e)

            if body_model:
                # Decode json payload
                try:
                    body = json.loads(request.body.decode("utf-8"))
                except json.decoder.JSONDecodeError:
                    raise error.InvalidJSON()

                # Validate payload
                try:
                    body = body_model(**body)
                    models.append(body)
                except ValidationError as e:
                    raise error.BadRequest(e)

            return await func(request, *models)

        return wrapper

    return decorate