[+] update pydantic, validate_params

1. extend to support sync/async methods;
This commit is contained in:
Siarhei Siniak 2025-10-11 17:07:45 +03:00
parent 31cd7c35c9
commit cc4a703bcb
4 changed files with 43 additions and 9 deletions

@ -1,2 +1,5 @@
[**/*.py] [**/*.py]
vim_modeline = set noet ts=2 sts=2 sw=2 ai ci vim_modeline = set noet ts=2 sts=2 sw=2 ai ci
[**/meson.build]
vim_modeline = set noet ts=2 sts=2 sw=2 ai ci

@ -5,7 +5,7 @@ project(
).stdout().strip('\n'), ).stdout().strip('\n'),
# 'online.fxreader.uv', # 'online.fxreader.uv',
# ['c', 'cpp'], # ['c', 'cpp'],
version: '0.1.5.28', version: '0.1.5.29',
# default_options: [ # default_options: [
# 'cpp_std=c++23', # 'cpp_std=c++23',
# # 'prefer_static=true', # # 'prefer_static=true',

@ -1,4 +1,7 @@
import pydantic import pydantic
import functools
# import asgiref.sync
import inspect import inspect
import collections import collections
@ -9,13 +12,23 @@ from typing import (
Optional, Optional,
Mapping, Mapping,
cast, cast,
Awaitable,
overload,
) )
P = TypeVar('P') P = TypeVar('P')
R = TypeVar('R') R = TypeVar('R')
def validate_params(view: Callable[..., R]) -> Callable[..., R]: @overload
def validate_params(view: Callable[..., Awaitable[R]]) -> Callable[..., Awaitable[R]]: ...
@overload
def validate_params(view: Callable[..., R]) -> Callable[..., R]: ...
def validate_params(view: Callable[..., Awaitable[R]] | Callable[..., R]) -> Any:
class Parameter: class Parameter:
kind: Any kind: Any
annotation: Any annotation: Any
@ -52,7 +65,10 @@ def validate_params(view: Callable[..., R]) -> Callable[..., R]:
), ),
) )
def wrapper(*args: Any, **kwargs: Any) -> R: sync_view: Optional[Callable[..., R]] = None
async_view: Optional[Callable[..., Awaitable[R]]] = None
def validate_params(*args: Any, **kwargs: Any) -> None:
# data = model.model_validate( # data = model.model_validate(
kwargs_to_check: dict[str, Any] = {k: v for k, v in kwargs.items()} kwargs_to_check: dict[str, Any] = {k: v for k, v in kwargs.items()}
@ -71,10 +87,22 @@ def validate_params(view: Callable[..., R]) -> Callable[..., R]:
) )
# ).dict() # ).dict()
return view( if inspect.iscoroutinefunction(view):
# **data, async_view = cast(Callable[..., Awaitable[R]], view)
*args, raise NotImplementedError
**kwargs,
) @functools.wraps(async_view)
async def wrapper(*args: Any, **kwargs: Any) -> R:
validate_params(*args, **kwargs)
return await async_view(*args, **kwargs)
else:
sync_view = cast(Callable[..., R], view)
@functools.wraps(sync_view)
def wrapper(*args: Any, **kwargs: Any) -> R:
validate_params(*args, **kwargs)
return sync_view(*args, **kwargs)
return wrapper return wrapper

BIN
releases/whl/online_fxreader_pr34-0.1.5.29-py3-none-any.whl (Stored with Git LFS) Normal file

Binary file not shown.