Skip to content

forge.validation

Pydantic-integrated input validation with consistent error responses.

forge.validation

Classes

ValidationErrorDetail

Bases: BaseModel

Details of a single validation error.

Source code in src/forge/validation/__init__.py 
16
17
18
19
20
21
class ValidationErrorDetail(BaseModel):
    """Details of a single validation error."""

    loc: list[str | int]
    msg: str
    type: str

ValidationErrorResponse

Bases: BaseModel

Standard error response for validation failures.

Source code in src/forge/validation/__init__.py 
24
25
26
27
class ValidationErrorResponse(BaseModel):
    """Standard error response for validation failures."""

    detail: list[ValidationErrorDetail]

ValidationModule

Bases: ForgeModule

Lightweight module providing Pydantic integration and consistent validation helpers.

Source code in src/forge/validation/module.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
class ValidationModule(ForgeModule):
    """
    Lightweight module providing Pydantic integration and consistent validation helpers.
    """

    name = "validation"
    dependencies: ClassVar[list[str]] = []

    async def setup(self, runtime: Runtime) -> None:
        """Set up the validation module."""

    async def teardown(self) -> None:
        """Teardown the validation module."""

    def health_check(self) -> HealthResult:
        """Return the health check status of the validation module."""
        return HealthResult.ok()
Methods:
health_check 
health_check() -> HealthResult

Return the health check status of the validation module.

Source code in src/forge/validation/module.py 
25
26
27
def health_check(self) -> HealthResult:
    """Return the health check status of the validation module."""
    return HealthResult.ok()
setup async 
setup(runtime: ForgeRuntime) -> None

Set up the validation module.

Source code in src/forge/validation/module.py 
19
20
async def setup(self, runtime: Runtime) -> None:
    """Set up the validation module."""
teardown async 
teardown() -> None

Teardown the validation module.

Source code in src/forge/validation/module.py 
22
23
async def teardown(self) -> None:
    """Teardown the validation module."""

Functions:

format_validation_error 

format_validation_error(exc: ValidationError) -> ValidationErrorResponse

Convert Pydantic ValidationError to standard error response format.

Source code in src/forge/validation/__init__.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def format_validation_error(exc: ValidationError) -> ValidationErrorResponse:
    """Convert Pydantic ValidationError to standard error response format."""
    details = []
    for error in exc.errors():
        loc = list(error.get("loc", []))
        # Ensure location starts with "body" for consistency with FastAPI
        if not loc or loc[0] != "body":
            loc.insert(0, "body")

        details.append(
            ValidationErrorDetail(
                loc=loc,
                msg=error.get("msg", ""),
                type=error.get("type", ""),
            )
        )
    return ValidationErrorResponse(detail=details)

validate 

validate(schema: type[BaseModel]) -> Callable[[F], F]

Decorator that validates function input against a Pydantic schema.

On validation error, raises HTTPException(422) with structured error details.

Source code in src/forge/validation/__init__.py 
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def validate(schema: type[BaseModel]) -> Callable[[F], F]:
    """
    Decorator that validates function input against a Pydantic schema.

    On validation error, raises HTTPException(422) with structured error details.
    """

    def decorator(func: F) -> F:
        # Determine the position of the argument to validate
        # By default, we inspect the first parameter (after self/cls if applicable)
        first_param_name: str | None = None
        has_self_cls = False
        try:
            sig = inspect.signature(func)
            params = list(sig.parameters.keys())
            if params and params[0] in ("self", "cls"):
                has_self_cls = True
                if len(params) > 1:
                    first_param_name = params[1]
            elif params:
                first_param_name = params[0]
        except Exception:  # noqa: S110
            pass

        def _validate_args(
            args: tuple[Any, ...],
            kwargs: dict[str, Any],
        ) -> tuple[tuple[Any, ...], dict[str, Any]]:
            new_args = list(args)
            new_kwargs = dict(kwargs)

            # Check positional arguments
            body: Any = None
            body_index: int | None = None

            if args:
                if has_self_cls and len(args) > 1:
                    body = args[1]
                    body_index = 1
                elif not has_self_cls:
                    body = args[0]
                    body_index = 0

            # Check keyword arguments if not found positionally
            if body is None and first_param_name and first_param_name in kwargs:
                body = kwargs[first_param_name]

            if isinstance(body, dict):
                try:
                    validated = schema.model_validate(body)
                    if body_index is not None:
                        new_args[body_index] = validated
                    elif first_param_name:
                        new_kwargs[first_param_name] = validated
                except ValidationError as exc:
                    formatted = format_validation_error(exc)
                    raise HTTPException(
                        status_code=422,
                        detail=formatted.model_dump()["detail"],
                    ) from exc
            elif isinstance(body, schema):
                # Already correct type, pass through
                pass

            return tuple(new_args), new_kwargs

        if inspect.iscoroutinefunction(func):

            @wraps(func)
            async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
                validated_args, validated_kwargs = _validate_args(args, kwargs)
                return await func(*validated_args, **validated_kwargs)

            return cast("F", async_wrapper)

        @wraps(func)
        def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
            validated_args, validated_kwargs = _validate_args(args, kwargs)
            return func(*validated_args, **validated_kwargs)

        return cast("F", sync_wrapper)

    return decorator