wehub-resource-sync
97e91a83f3
Ruff / Ruff (push) Has been cancelled
Test / Core Tests (push) Has been cancelled
Test / Offline Coverage Tests (Python 3.10) (push) Has been cancelled
Test / Offline Coverage Tests (Python 3.11) (push) Has been cancelled
Test / Offline Coverage Tests (Python 3.12) (push) Has been cancelled
Test / Offline Coverage Tests (Python 3.13) (push) Has been cancelled
Test / Offline Coverage Tests (Python 3.9) (push) Has been cancelled
Test / Full Coverage (Python 3.11) (push) Has been cancelled
Test / Core Provider Tests (OpenAI) (push) Has been cancelled
Test / Core Provider Tests (Anthropic) (push) Has been cancelled
Test / Core Provider Tests (Google) (push) Has been cancelled
Test / Core Provider Tests (Other) (push) Has been cancelled
Test / Anthropic Tests (push) Has been cancelled
Test / Gemini Tests (push) Has been cancelled
Test / Google GenAI Tests (push) Has been cancelled
Test / Vertex AI Tests (push) Has been cancelled
Test / OpenAI Tests (push) Has been cancelled
Test / Writer Tests (push) Has been cancelled
Test / Auto Client Tests (push) Has been cancelled
ty / type-check (push) Has been cancelled
76 lines
2.3 KiB
Python
76 lines
2.3 KiB
Python
"""Async validation decorators owned by the v2 runtime."""
|
|
|
|
from inspect import signature
|
|
from typing import Any, Callable, TypeVar
|
|
|
|
from pydantic import ValidationInfo
|
|
|
|
ASYNC_VALIDATOR_KEY = "__async_validator__"
|
|
ASYNC_MODEL_VALIDATOR_KEY = "__async_model_validator__"
|
|
T = TypeVar("T", bound=Callable[..., Any])
|
|
|
|
|
|
class AsyncValidationContext:
|
|
"""Carry context through async validation hooks."""
|
|
|
|
context: dict[str, Any]
|
|
|
|
def __init__(self, context: dict[str, Any]):
|
|
self.context = context
|
|
|
|
|
|
def async_field_validator(field: str, *fields: str) -> Callable[[T], T]:
|
|
"""Mark a callable as an async field validator."""
|
|
field_names = field, *fields
|
|
|
|
def decorator(func: T) -> T:
|
|
params = signature(func).parameters
|
|
requires_validation_context = False
|
|
if len(params) == 3:
|
|
if "info" not in params:
|
|
raise ValueError(
|
|
"Async validator can only have a value parameter and an optional info parameter"
|
|
)
|
|
if params["info"].annotation != ValidationInfo:
|
|
raise ValueError(
|
|
"Async validator info parameter must be of type ValidationInfo"
|
|
)
|
|
requires_validation_context = True
|
|
|
|
setattr(
|
|
func, ASYNC_VALIDATOR_KEY, (field_names, func, requires_validation_context)
|
|
)
|
|
return func
|
|
|
|
return decorator
|
|
|
|
|
|
def async_model_validator() -> Callable[[T], T]:
|
|
"""Mark a callable as an async model validator."""
|
|
|
|
def decorator(func: T) -> T:
|
|
params = signature(func).parameters
|
|
requires_validation_context = False
|
|
if len(params) > 2:
|
|
raise ValueError("Invalid Parameter Count!")
|
|
|
|
if len(params) == 2:
|
|
if "info" not in params:
|
|
raise ValueError(
|
|
"Async validator can only have a value parameter and an optional info parameter"
|
|
)
|
|
if params["info"].annotation != ValidationInfo:
|
|
raise ValueError(
|
|
"Async validator info parameter must be of type ValidationInfo"
|
|
)
|
|
requires_validation_context = True
|
|
|
|
setattr(
|
|
func,
|
|
ASYNC_MODEL_VALIDATOR_KEY,
|
|
(func, requires_validation_context),
|
|
)
|
|
return func
|
|
|
|
return decorator
|