Fix alias in query and form (#184)

* Fix alias in query and form

* Fix list

* Fix mypy

---------

Co-authored-by: luolingchun <luolingchun@outloook.com>

Author: luolingchun

flask_openapi3/request.py

@@ -3,78 +3,121 @@
 # @Time    : 2022/4/1 16:54
 import json
 from json import JSONDecodeError
-from typing import Any, Type, Optional, Dict, get_origin, get_args
+from typing import Any, Type, Optional, Dict
 
 from flask import request, current_app, abort
 from pydantic import ValidationError, BaseModel
+from pydantic.fields import FieldInfo
+from werkzeug.datastructures.structures import MultiDict
 
-from .models import FileStorage
 
+def _get_list_value(model: Type[BaseModel], args: MultiDict, model_field_key: str, model_field_value: FieldInfo):
+    if model_field_value.alias and model.model_config.get("populate_by_name"):
+        key = model_field_value.alias
+        value = args.getlist(model_field_value.alias) or args.getlist(model_field_key)
+    elif model_field_value.alias:
+        key = model_field_value.alias
+        value = args.getlist(model_field_value.alias)
+    else:
+        key = model_field_key
+        value = args.getlist(model_field_key)
+
+    return key, value
+
+
+def _get_value(model: Type[BaseModel], args: MultiDict, model_field_key: str, model_field_value: FieldInfo):
+    if model_field_value.alias and model.model_config.get("populate_by_name"):
+        key = model_field_value.alias
+        value = args.get(model_field_value.alias) or args.get(model_field_key)
+    elif model_field_value.alias:
+        key = model_field_value.alias
+        value = args.get(model_field_value.alias)
+    else:
+        key = model_field_key
+        value = args.get(model_field_key)
 
-def _validate_header(header: Type[BaseModel], func_kwargs):
+    return key, value
+
+
+def _validate_header(header: Type[BaseModel], func_kwargs: dict):
     request_headers = dict(request.headers)
-    for key, value in header.model_fields.items():
-        key_title = key.replace("_", "-").title()
-        # Add original key
-        if key_title in request_headers.keys():
-            if value.alias:
-                request_headers[value.alias] = request_headers[key] = request_headers[key_title]
-            else:
-                request_headers[key] = request_headers[key_title]
-    func_kwargs["header"] = header.model_validate(obj=request_headers)
+    header_dict = {}
+    for model_field_key, model_field_value in header.model_fields.items():
+        key_title = model_field_key.replace("_", "-").title()
+        if model_field_value.alias and header.model_config.get("populate_by_name"):
+            key = model_field_value.alias
+            key_alias_title = model_field_value.alias.replace("_", "-").title()
+            value = request_headers.get(key_alias_title) or request_headers.get(key_title)
+        elif model_field_value.alias:
+            key = model_field_value.alias
+            key_alias_title = model_field_value.alias.replace("_", "-").title()
+            value = request_headers.get(key_alias_title)
+        else:
+            key = model_field_key
+            value = request_headers[key_title]
+        if value is not None:
+            header_dict[key] = value
+    func_kwargs["header"] = header.model_validate(obj=header_dict)
 
 
-def _validate_cookie(cookie: Type[BaseModel], func_kwargs):
+def _validate_cookie(cookie: Type[BaseModel], func_kwargs: dict):
     request_cookies = dict(request.cookies)
     func_kwargs["cookie"] = cookie.model_validate(obj=request_cookies)
 
 
-def _validate_path(path: Type[BaseModel], path_kwargs, func_kwargs):
+def _validate_path(path: Type[BaseModel], path_kwargs: dict, func_kwargs: dict):
     func_kwargs["path"] = path.model_validate(obj=path_kwargs)
 
 
-def _validate_query(query: Type[BaseModel], func_kwargs):
+def _validate_query(query: Type[BaseModel], func_kwargs: dict):
     request_args = request.args
     query_dict = {}
-    for k, v in query.model_fields.items():
-        if get_origin(v.annotation) is list:
-            value = request_args.getlist(v.alias or k) or request_args.getlist(k)
+    model_properties = query.model_json_schema().get("properties", {})
+    for model_field_key, model_field_value in query.model_fields.items():
+        model_field_schema = model_properties.get(model_field_value.alias or model_field_key)
+        if model_field_schema.get("type") == "array":
+            key, value = _get_list_value(query, request_args, model_field_key, model_field_value)
         else:
-            value = request_args.get(v.alias or k) or request_args.get(k)  # type:ignore
+            key, value = _get_value(query, request_args, model_field_key, model_field_value)
         if value is not None and value != []:
-            query_dict[k] = value
+            query_dict[key] = value
     func_kwargs["query"] = query.model_validate(obj=query_dict)
 
 
-def _validate_form(form: Type[BaseModel], func_kwargs):
+def _validate_form(form: Type[BaseModel], func_kwargs: dict):
     request_form = request.form
     request_files = request.files
     form_dict = {}
-    for k, v in form.model_fields.items():
-        if get_origin(v.annotation) is list:
-            if get_args(v.annotation)[0] is FileStorage:
-                value = request_files.getlist(v.alias or k) or request_files.getlist(k)
+    model_properties = form.model_json_schema().get("properties", {})
+    for model_field_key, model_field_value in form.model_fields.items():
+        model_field_schema = model_properties.get(model_field_value.alias or model_field_key)
+        if model_field_schema.get("type") == "array":
+            if model_field_schema.get("items") == {"format": "binary", "type": "string"}:
+                # list[FileStorage]
+                key, value = _get_list_value(form, request_files, model_field_key, model_field_value)
             else:
                 value = []
-                for i in request_form.getlist(v.alias or k) or request_form.getlist(k):
+                key, value_list = _get_list_value(form, request_form, model_field_key, model_field_value)
+                for _value in value_list:
                     try:
-                        value.append(json.loads(i))
+                        value.append(json.loads(_value))
                     except (JSONDecodeError, TypeError):
-                        value.append(i)  # type:ignore
-        elif v.annotation is FileStorage:
-            value = request_files.get(v.alias or k) or request_files.get(k)  # type:ignore
+                        value.append(_value)
+        elif model_field_schema.get("type") == "string" and model_field_schema.get("format") == "binary":
+            # FileStorage
+            key, value = _get_value(form, request_files, model_field_key, model_field_value)
         else:
-            _value = request_form.get(v.alias or k) or request_form.get(k)
+            key, _value = _get_value(form, request_form, model_field_key, model_field_value)
             try:
-                value = json.loads(_value)  # type:ignore
+                value = json.loads(_value)
             except (JSONDecodeError, TypeError):
-                value = _value  # type:ignore
+                value = _value
         if value is not None and value != []:
-            form_dict[k] = value
+            form_dict[key] = value
     func_kwargs["form"] = form.model_validate(obj=form_dict)
 
 
-def _validate_body(body: Type[BaseModel], func_kwargs):
+def _validate_body(body: Type[BaseModel], func_kwargs: dict):
     obj = request.get_json(silent=True)
     if isinstance(obj, str):
         body_model = body.model_validate_json(json_data=obj)
@@ -122,7 +165,7 @@ def _validate_request(
         if cookie:
             _validate_cookie(cookie, func_kwargs)
         if path:
-            _validate_path(path, path_kwargs, func_kwargs)
+            _validate_path(path, path_kwargs or {}, func_kwargs)
         if query:
             _validate_query(query, func_kwargs)
         if form:

tests/test_populate_by_name.py

@@ -1,6 +1,8 @@
 # -*- coding: utf-8 -*-
 # @Author  : llc
 # @Time    : 2024/8/31 15:35
+from typing import Sequence, List, Tuple
+
 import pytest
 from pydantic import BaseModel, Field
 
@@ -29,17 +31,101 @@ def get_book(query: BookQuery):
     """
     get all books
     """
-    return {
-        "code": 0,
-        "message": "ok",
-        "data": [
-            {"bid": 1, "age": query.age, "author": query.author},
-            {"bid": 2, "age": query.age, "author": query.author}
-        ]
-    }
+    print(query)
+    return "ok"
+
+
+class QueryModel(BaseModel):
+    aliased_field: str = Field(alias="aliasedField")
+
+    aliased_list_field: List[str] = Field(alias="aliasedListField")
+
+
+@app.get("/query-alias-test")
+def query_alias_test(query: QueryModel):
+    assert query.aliased_field == "test"
+    return "ok"
+
+
+class HeaderModel(BaseModel):
+    Hello1: str = Field(..., alias="Hello2")
+
+    model_config = {"populate_by_name": True}
+
+
+@app.get("/header")
+def get_book_header(header: HeaderModel):
+    return header.model_dump(by_alias=True)
+
+
+class TupleModel(BaseModel):
+    values: Tuple[int, int]
+    sequence: Sequence[int] = Field(alias="Sequence")
+
+    model_config = {"populate_by_name": True}
+
+
+@app.get("/tuple-test")
+def tuple_test(query: TupleModel):
+    assert query.values == (2, 2)
+    return b"", 200
+
+
+class AliasModel(BaseModel):
+    aliased_field: str = Field(alias="aliasedField")
+
+
+@app.post("/form-alias-test")
+def alias_test(form: AliasModel):
+    assert form.aliased_field == "test"
+    return b"", 200
+
+
+def test_header(client):
+    headers = {"Hello2": "111"}
+    resp = client.get("/header", headers=headers)
+    print(resp.json)
+    assert resp.status_code == 200
+    assert resp.json == headers
+
+
+def test_tuple_query(client):
+    resp = client.get(
+        "/tuple-test",
+        query_string={"values": [2, 2], "sequence": [1, 2, 3]},
+    )
+    assert resp.status_code == 200
+
+
+def test_form_alias(client):
+    resp = client.post(
+        "/form-alias-test",
+        data={"aliasedField": "test"},
+    )
+    assert resp.status_code == 200
+
+    resp = client.post(
+        "/form-alias-test",
+        data={"aliased_field": "test"},
+    )
+    assert resp.status_code == 422
+
+
+def test_query_alias(client):
+    resp = client.get(
+        "/query-alias-test",
+        query_string={"aliasedField": "test", "aliasedListField": ["test"]},
+    )
+    assert resp.status_code == 200
+
+    resp = client.get(
+        "/query-alias-test",
+        data={"aliased_field": "test", "aliased_list_field": ["test"]},
+    )
+    assert resp.status_code == 422
 
 
-def test_openapi(client):
+def test_query_populate_by_name(client):
     resp = client.get("/book?age=1&author=aa")
     assert resp.status_code == 200
     resp = client.get("/book?age=1&author_name=aa")