5.2.4. Build validation
Data validation is an important part, and it must occur before executing the main functions in the application. In this section, we will set up a validator using Pydantic, a library that provides classes to create schemas and validate data.
A special point in this section is that we will not create a specific validator for any particular input. Instead, we will create a wrapper for the validator that can be used in the pipeline.
In the validation directory, create a pydantic folder, and inside pydantic/, create helpers.py. Add some code to helpers.py.
from typing import Callable, Type
from pydantic import BaseModel, ValidationError
# Import errors
from core.error import ClientError
This function is used to create a step executor, and it will be used within a pipeline step.
def create_validation_step_executor(
pipeline, validator_class: Type[BaseModel]
) -> Callable:
"""
Tạo executor cho bước xác minh dữ liệu trong pipeline.
Args:
pipeline: Pipeline mà bạn muốn thêm step vào.
validator_class: Lớp Pydantic dùng để xác minh dữ liệu.
Returns:
Callable thực thi bước xác minh dữ liệu trong pipeline.
"""
async def executor(ctx):
body = await ctx.get_body()
try:
validated = validator_class(**body)
return validated
except ValidationError as e:
pipeline.stop(ctx)
err = ClientError("Validation failed")
err.as_http_error("BadRequest")
for detail in e.errors():
err.add_error_detail(
{
"source": ".".join(str(loc) for loc in detail.get("loc", [])),
"desc": detail.get("msg"),
}
)
return ctx.send_error(err)
return executor
