miss85246 / sanic-dantic Goto Github PK

def validate(request, header=None, path=None, query=None, form=None, body=None, return_json=False, send_first_error_only=True):
    """
    When there are the same parameter name in the model, the parameter in ParsedArgsObj will be overwritten,
    The priorities is: body = form > query > path \n
    :param header: Pydantic model
    :param request: Pydantic model
    :param query: Pydantic model
    :param body: Pydantic model, cannot exist at the same time as form
    :param path: Pydantic model
    :param form: Pydantic model, cannot exist at the same time as body
    :param return_json: Whether to return JSON instead of text
    :param send_first_error_only: Whether to send only the first error in the list or all errors
    :return parsed_args: ParsedArgsObj
    """
    try:
        error_message = "sanic-dantic: body model must be used together with one of ['POST','PUT','PATCH','DELETE']"
        assert request.method in ["POST", "PUT", "PATCH", "DELETE"] if body else True, error_message
        models, parsed_args = [header, path, query] + [form or body], ParsedArgsObj()
        body_storage = {k: v[0] if len(v) == 1 else v for k, v in request.form.items()} if form else request.json
        params = {k: v[0] if len(v) == 1 else v for k, v in request.args.items()}
        storages = [request.headers, request.match_info, params, body_storage]
        models = [item for item in zip(models, storages) if item[0]]
        [parsed_args.update(model(**storage).dict()) for model, storage in models]
    except ValidationError as e:
        errors = e.errors()
        if send_first_error_only:
            errors = [e.errors()[0]]

        request.ctx.sanic_dantic_validation_errors = errors

        message = '\n'.join([f'{msg.get("loc")[0]} {msg.get("msg")}' for msg in errors])

        if not return_json:
            request.ctx.sanic_dantic_validation_errors = message
            request.ctx.sanic_dantic_return_type = 'text'
        else:
            request.ctx.sanic_dantic_return_type = 'json'
        raise InvalidUsage(message)

async def catch_errors(request, exception):
  if hasattr(request.ctx, 'sanic_dantic_validation_errors'):
    logging.error(request.ctx.sanic_dantic_validation_errors)
    return getattr(sanic.response, request.ctx.sanic_dantic_return_type)(request.ctx.sanic_dantic_validation_errors, status=exception.status_code)

app.error_handler.add(InvalidUsage, catch_errors)

@parse_params(path=List[str])
@parse_params(path=List[Person])