-
Notifications
You must be signed in to change notification settings - Fork 159
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
feat(BA-572): Add pydantic-only API hander decorator #3511
base: main
Are you sure you want to change the base?
feat(BA-572): Add pydantic-only API hander decorator #3511
Conversation
Co-authored-by: octodog <[email protected]>
…tps://github.com/lablup/backend.ai into feat/add-pydantic-handling-decorator-for-req-res
class BackendError(web.HTTPError): | ||
""" | ||
An RFC-7807 error class as a drop-in replacement of the original | ||
aiohttp.web.HTTPError subclasses. | ||
""" |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Ah, BackendError wasn’t in common package…
For now, I’ll apply it this way, and I’ll create a separate issue to organize exceptions and refactor later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I'll keep it in mind until the next refactoring
origin_name = get_origin(param_type).__name__ | ||
pydantic_model = get_args(param_type)[0] | ||
param_instance = param_type(pydantic_model) | ||
|
||
match origin_name: | ||
case "BodyParam": |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Is it difficult to use type information instead of origin_name
?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@HyeockJinKim
I think it can be achieved by if origin_type is BodyParam
. However, since it is not recommended to use is
to compare between classes, how about make enum, register it as a class variable and compare them?
class _ParamType(Enum):
BODY = auto()
QUERY = auto()
HEADER = auto()
PATH = auto()
MIDDLEWARE = auto()
class BodyParam(Generic[T]):
_param_type: _ParamType = _ParamType.BODY
async def _extract_param_value(
request: web.Request, parsed_signature: _ParsedSignature
) -> Optional[Any]:
try:
input_param_type = parsed_signature.input_param_type
origin_type = get_origin(input_param_type)
if not hasattr(origin_type, "_param_type"):
raise UnknownRequestParamType
match origin_type._param_type:
case _ParamType.BODY:
return param_instance.from_body(body)
case _ParamType.QUERY:
return param_instance.from_query(request.query)
case _ParamType.HEADER:
return param_instance.from_header(request.headers)
case _ParamType.PATH:
return param_instance.from_path(request.match_info)
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think using a match case would make it easier to distinguish types. Why is it necessary to introduce an enum?
for example:
def match_case_example(param):
match param:
case BodyParam():
param.from_body()
case QueryParam():
param.from_query()
case PathParam():
param.from_path()
case HeaderParam():
param.from_header()
async def pydantic_handler(request: web.Request, handler) -> web.Response: | ||
signature = inspect.signature(handler) | ||
handler_params = _HandlerParameters() | ||
for name, param in signature.parameters.items(): | ||
# Raise error when parameter has no type hint or not wrapped by 'Annotated' | ||
if param.annotation is inspect.Parameter.empty: | ||
raise InvalidAPIParameters( | ||
f"Type hint or Annotated must be added in API handler signature: {param.name}" | ||
) | ||
|
||
parsed_signature = _ParsedSignature(name=name, param_type=param.annotation) | ||
value = await extract_param_value(request=request, parsed_signature=parsed_signature) | ||
|
||
if not value: | ||
raise InvalidAPIParameters( | ||
f"Type hint or Annotated must be added in API handler signature: {param.name}" | ||
) | ||
|
||
handler_params.add(name, value) | ||
|
||
response = await handler(**handler_params.get_all()) | ||
|
||
if not isinstance(response, BaseResponse): | ||
raise InvalidAPIParameters( | ||
f"Only Response wrapped by BaseResponse Class can be handle: {type(response)}" | ||
) | ||
|
||
return web.json_response(response.data.model_dump(mode="json"), status=response.status_code) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Shouldn’t it be a private function?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes. I will change function name
class UserPathModel(BaseModel): | ||
user_id: str | ||
|
||
|
||
class UserPathResponse(BaseModel): | ||
user_id: str |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Please make it clear in the naming that the mocked request and response are for testing purposes.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I'll add prefix 'Test' in request and response models
@dataclass | ||
class BaseResponse: | ||
data: BaseModel | ||
status_code: int = 200 |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think it would be better not to set a default value and instead explicitly specify the status code for each handler.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I also think defining specific status code for each handler would be more explicit and prevent unintended behavior. I will fix it
class _HandlerParameters: | ||
def __init__(self) -> None: | ||
self.params: dict[str, Any] = {} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Please declare all the fields used in the class.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
If the value should not be accessed outside the class, please set it as private.
""" | ||
This decorator processes HTTP request parameters using Pydantic models. | ||
|
||
1. Request Body: | ||
@pydantic_api_handler | ||
async def handler(body: BodyParam[UserModel]): # UserModel is a Pydantic model | ||
user = body.parsed # 'parsed' property gets pydantic model you defined | ||
return BaseResponse(data=YourResponseModel(user=user.id)) | ||
|
||
2. Query Parameters: | ||
@pydantic_api_handler | ||
async def handler(query: QueryParam[QueryPathModel]): | ||
parsed_query = query.parsed | ||
return BaseResponse(data=YourResponseModel(search=parsed_query.query)) | ||
|
||
3. Headers: | ||
@pydantic_api_handler | ||
async def handler(headers: HeaderParam[HeaderModel]): | ||
parsed_header = headers.parsed | ||
return BaseResponse(data=YourResponseModel(data=parsed_header.token)) | ||
|
||
4. Path Parameters: | ||
@pydantic_api_handler | ||
async def handler(path: PathModel = PathParam(PathModel)): | ||
parsed_path = path.parsed | ||
return BaseResponse(data=YourResponseModel(path=parsed_path)) | ||
|
||
5. Middleware Parameters: | ||
# Need to extend MiddlewareParam and implement 'from_request' | ||
class AuthMiddlewareParam(MiddlewareParam): | ||
user_id: str | ||
user_email: str | ||
@classmethod | ||
def from_request(cls, request: web.Request) -> Self: | ||
# Extract and validate data from request | ||
user_id = request["user"]["uuid"] | ||
user_email = request["user"]["email"] |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This comment should be added as a comment in the function that will be used as a decorator.
@property | ||
def parsed(self) -> T: | ||
if not self._parsed: | ||
raise ValueError("Data not yet parsed") |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
How about defining a custom exception internally instead of using a ValueError?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Okay, I will replace them to custom exceptions
resolves #3512 (BA-572)
Description
Implement Pydantic api handler decorator to handle all request components (body, query params, path variables) through Pydantic models instead of web.Request
Implementation
To use
MiddlewareParam
Multiple Parameters
Test
Checklist: (if applicable)
ai.backend.test
docs
directory📚 Documentation preview 📚: https://sorna--3511.org.readthedocs.build/en/3511/
📚 Documentation preview 📚: https://sorna-ko--3511.org.readthedocs.build/ko/3511/