Clever Geek Handbook

Service API Json sur aiohttp: middleware et validation

Dans cet article, je décrirai l'une des approches pour créer un service API json avec validation des données.







Le service sera implémenté sur aiohttp . C'est un framework python moderne et en constante évolution qui utilise asyncio



.







À propos des annotations:







L'introduction d'annotations dans le python



code a facilité sa compréhension. De plus, les annotations ouvrent des possibilités supplémentaires. Ce sont les annotations qui jouent un rôle clé dans la validation des données pour les gestionnaires de méthodes API dans cet article.







Bibliothèques utilisées:







  • aiohttp - un cadre pour la création d'applications Web
  • pydantic - classes qui vous permettent de décrire de manière déclarative les données et de les valider
  • valdec - décorateur pour valider les arguments et renvoyer les valeurs des fonctions


Table des matières:









1. 



- sources -    
    - data_classes -     
        - base.py -   
        - person.py -    
        - wraps.py -     /
    - handlers -     
        - kwargs.py -      `KwargsHandler.middleware`
        - simple.py -      `SimpleHandler.middleware`
        - wraps.py -      `WrapsKwargsHandler.middleware`
    - middlewares -     middlewares
        - exceptions.py -  
        - kwargs_handler.py -  `KwargsHandler`
        - simple_handler.py -  `SimpleHandler`
        - utils.py -      middlewares
        - wraps_handler.py -  `WrapsKwargsHandler`
    - requirements.txt -  
    - run_kwargs.py -   `KwargsHandler.middleware`
    - run_simple.py -  c `SimpleHandler.middleware`
    - run_wraps.py -  c `WrapsKwargsHandler.middleware`
    - settings.py -    
- Dockerfile -


: https://github.com/EvgeniyBurdin/api_service







2. json middlewares



middleware



aiohttp.web.Application()



.







middleware



, , . . middleware



.







middleware



, .







middleware



"" "" web.Request



web.Response



. .







, middleware



/, .







, .







2.1. middleware json



, aiohttp.web.Application()



, , :







from aiohttp import web

async def some_handler(request: web.Request) -> web.Response:
    data = await request.json()
    ...
    text = json.dumps(some_data)
    ...
    return web.Response(text=text, ...)


"" web.Request



, json. , . json "" web.Response



( web.json_response()



).







2.1.1.



. , middleware



, , :







from aiohttp import web

async def some_handler(request: web.Request, data: Any) -> Any:
    ...
    return some_data


. web.Request



( ), — python, .







, : data: Any



. ( ), , "" . .







, , :







from aiohttp import web
from typing import Union, List

async def some_handler(
    request: web.Request, data: Union[str, List[str]]
) -> List[int]:
    ...
    return some_data


2.1.2. SimpleHandler



middleware



SimpleHandler



middleware



, / middleware



( ).







.







2.1.2.1. middleware 





    @web.middleware
    async def middleware(self, request: web.Request, handler: Callable):
        """ middleware  json-.
        """
        if not self.is_json_service_handler(request, handler):
            return await handler(request)

        try:
            request_body = await self.get_request_body(request, handler)

        except Exception as error:
            response_body = self.get_error_body(request, error)
            status = 400

        else:
            #  
            response_body, status = await self.get_response_body_and_status(
                request, handler, request_body
            )

        finally:
            #     python (  
            # response_body)   json.
            text, status = await self.get_response_text_and_status(
                request, response_body, status
            )

        return web.Response(
            text=text, status=status, content_type="application/json",
        )


middlewares



.







, :







    ...
    app = web.Application()
    service_handler = SimpleHandler()
    app.middlewares.append(service_handler.middleware)
    ...


2.1.2.2.


json , , , ( 400), ( 500), json.







"" :







    def get_error_body(self, request: web.Request, error: Exception) -> dict:
        """       .
        """
        return {"error_type": str(type(error)), "error_message": str(error)}


, , json. , json .







2.1.2.3.


:







    async def run_handler(
        self, request: web.Request, handler: Callable, request_body: Any
    ) -> Any:
        """   ,     .
        """
        return await handler(request, request_body)


, / .







2.1.3.



:







async def some_handler(request: web.Request, data: dict) -> dict:
    return data


url .







...
2.1.3.1. Répondre avec le code 200


Demande POST



de /some_handler



:







{
    "name": "test",
    "age": 25
}


… 200:







{
    "name": "test",
    "age": 25
}


2.1.3.2. 400


.







POST



/some_handler



:







{
    "name": "test", 111111111111
    "age": 25
}


:







{
    "error_type": "<class 'json.decoder.JSONDecodeError'>",
    "error_message": "Expecting property name enclosed in double quotes: line 2 column 21 (char 22)"
}


2.1.3.3. 500


( ).







async def handler500(request: web.Request, data: dict) -> dict:
    raise Exception("  500")
    return data


POST



/handler500



:







{
    "name": "test",
    "age": 25
}


:







{
    "error_type": "<class 'Exception'>",
    "error_message": "  500"
}


2.2. middleware "kwargs-"



middleware



.







.







:







async def some_handler(request: web.Request, data: dict) -> dict:

    storage = request.app["storage"]
    logger = request.app["logger"]
    user_id = request.match_info["user_id"]
    #  ..  ....

    return data


storage



, logger



( - ), , "" .







2.2.1.



, , , :







async def some_handler_1(data: dict) -> int:
    # ...
    return some_data

async def some_handler_2(storage: StorageClass, data: List[int]) -> dict:
    # ...
    return some_data

async def some_handler_3(
    data: Union[dict, List[str]], logger: LoggerClass, request: web.Request
) -> str:
    # ...
    return some_data


, .







2.2.2. ArgumentsManager





middleware



, "" "" .







, "" ArgumentsManager



. middlewares/utils.py



( ).







" " — " " , " ", — , " ".







, :







@dataclass
class RawDataForArgument:

    request: web.Request
    request_body: Any
    arg_name: Optional[str] = None

class ArgumentsManager:
    """    .

            ,    
          .
    """

    def __init__(self) -> None:

        self.getters: Dict[str, Callable] = {}

    #  json  ------------------------------------------------------

    def reg_request_body(self, arg_name) -> None:
        """      .
        """
        self.getters[arg_name] = self.get_request_body

    def get_request_body(self, raw_data: RawDataForArgument):
        return raw_data.request_body

    #   request --------------------------------------------------------

    def reg_request_key(self, arg_name) -> None:
        """       request.
        """
        self.getters[arg_name] = self.get_request_key

    def get_request_key(self, raw_data: RawDataForArgument):
        return raw_data.request[raw_data.arg_name]

    #   request.app ----------------------------------------------------

    def reg_app_key(self, arg_name) -> None:
        """       app.
        """
        self.getters[arg_name] = self.get_app_key

    def get_app_key(self, raw_data: RawDataForArgument):
        return raw_data.request.app[raw_data.arg_name]

    #   ------------------------------------------------------

    def reg_match_info_key(self, arg_name) -> None:
        """        .
        """
        self.getters[arg_name] = self.get_match_info_key

    def get_match_info_key(self, raw_data: RawDataForArgument):
        return raw_data.request.match_info[raw_data.arg_name]

    #     ...


web.Application()



:








# ...

app = web.Application()

arguments_manager = ArgumentsManager()

#    ,    
#    json- 
arguments_manager.reg_request_body("data")

#    ,    
#      request.match_info
arguments_manager.reg_match_info_key("info_id")

#     
# (  "  "     )
app["storage"] = SomeStorageClass(login="user", password="123")
#    ,    
#  
arguments_manager.reg_app_key("storage")

# ...


ArgumentsManager



. middleware



:







...
service_handler = KwargsHandler(arguments_manager=arguments_manager)
app.middlewares.append(service_handler.middleware)
...


. , , … , , .







2.2.3. KwargsHandler



middleware



KwargsHandler



SimpleHandler



, .2.2.1.







— run_handler



, — make_handler_kwargs



build_error_message_for_invalid_handler_argument



( ).







2.2.3.1.


:







    async def run_handler(
        self, request: web.Request, handler: Callable, request_body: Any
    ) -> Any:
        """   ,     .

            (   ,   
             //)
        """
        kwargs = self.make_handler_kwargs(request, handler, request_body)

        return await handler(**kwargs)


, . , . .







2.2.3.2.


make_handler_kwargs



. , . ArgumentsManager



.







, , ArgumentsManager



.







. , web.Request



, web.Request



(, r: web.Request



req: web.Request



request: web.Request



). , web.Request



"" , .







: .







build_error_message_for_invalid_handler_argument



— . .







2.2.4.



:







async def create(
    data: Union[dict, List[dict]], storage: dict,
) -> Union[dict, List[dict]]:
    # ...

async def read(storage: dict, data: str) -> dict:
    # ...

async def info(info_id: int, request: web.Request) -> str:
    # ...


POST



, — GET



(, )







...
2.2.4.1. /create


:







[
    {
        "name": "Ivan"
    },
    {
        "name": "Oleg"
    }
]


:







[
    {
        "id": "5730bab1-9c1b-4b01-9979-9ad640ea5fc1",
        "name": "Ivan"
    },
    {
        "id": "976d821a-e871-41b4-b5a2-2875795d6166",
        "name": "Oleg"
    }
]


2.2.4.2. /read


:







"5730bab1-9c1b-4b01-9979-9ad640ea5fc1"


:







{
    "id": "5730bab1-9c1b-4b01-9979-9ad640ea5fc1",
    "name": "Ivan"
}


: UUID



, 500



— PersonNotFound



.







2.2.4.3. /info/{info_id}


GET



/info/123



:







"any json"


:







"info_id=123 and request=<Request GET /info/123 >"


2.3. middleware c /



, api- .







, create



:







{
    "data": [
        {
            "name": "Ivan"
        },
        {
            "name": "Oleg"
        }
    ],
    "id": 11
}


:







{

    "success": true,
    "result": [
        {
            "id": "9738d8b8-69da-40b2-8811-b33652f92f1d",
            "name": "Ivan"
        },
        {
            "id": "df0fdd43-4adc-43cd-ac17-66534529d440",
            "name": "Oleg"
        }
    ],
    "id": 11
}


, data



result



.







id



, .







success



.







, :







read



:







{
    "data":  "ddb0f2b1-0179-44b7-b94d-eb2f3b69292d",
    "id": 3
}


:







{
    "success": false,
    "result": {
        "error_type": "<class 'handlers.PersonNotFound'>",
        "error_message": "Person whith id=ddb0f2b1-0179-44b7-b94d-eb2f3b69292d not found!"
    },
    "id": 3
}


json middleware



middleware



. run_handler



, ( ) get_error_body



.







, "" , ( data



). ( result



). middleware



.







, , .







" ", . .







2.3.1. pydantic.BaseModel





pydantic.BaseModel



.







( ). — .







:







from pydantic import BaseModel
from typing import Union, List

class Info(BaseModel):
    foo: int

class Person(BaseModel):
    name: str
    info: Union[Info, List[Info]]

kwargs = {"name": "Ivan", "info": {"foo": 0}}
person = Person(**kwargs)
assert person.info.foo == 0

kwargs = {"name": "Ivan", "info": [{"foo": 0}, {"foo": 1}]}
person = Person(**kwargs)
assert person.info[1].foo == 1

kwargs = {"name": "Ivan", "info": {"foo": "bar"}}  # <- , str  int
person = Person(**kwargs)
#  :
# ...
# pydantic.error_wrappers.ValidationError: 2 validation errors for Person
# info -> foo
#  value is not a valid integer (type=type_error.integer)
# info
#  value is not a valid list (type=type_error.list)


, , . , , .







typing



.







- pydantic.BaseModel



, "" ( … , "" — ).







. , : info.foo



int



, info



list



, .







pydantic.BaseModel



, .







2.3.1.1.


, , :







kwargs = {"name": "Ivan", "info": {"foo": "0"}}
person = Person(**kwargs)
assert person.info.foo == 0


, . str->int



( pydantic



. ).







, , , UUID



-> UUID



. , , , Strict...



. , pydantic.StrictInt



, pydantic.StrictStr



, ....







2.3.1.2.


, , :







kwargs = {"name": "Ivan", "info": {"foo": 0}, "bar": "BAR"}
person = Person(**kwargs)


.







, .







, , :







from pydantic import BaseModel, Extra, StrictInt, StrictStr
from typing import Union, List

class BaseApi(BaseModel):
    class Config:
        #    (ignore),  (allow) 
        #  (forbid)     
        # , :
        # https://pydantic-docs.helpmanual.io/usage/model_config/
        extra = Extra.forbid

class Info(BaseApi):
    foo: StrictInt

class Person(BaseApi):
    name: StrictStr
    info: Union[Info, List[Info]]

kwargs = {"name": "Ivan", "info": {"foo": 0}, "bar": "BAR"}
person = Person(**kwargs)
# ...
# pydantic.error_wrappers.ValidationError: 1 validation error for Person
# bar
#   extra fields not permitted (type=value_error.extra)


— , .







2.3.2. valdec.validate





valdec.validate / .







, .







, None



( -> None:



).







/:







from valdec.decorators import validate

@validate  #     ,  
def foo(i: int, s: str) -> int:
    return i

@validate("i", "s")  #   "i"  "s"
def bar(i: int, s: str) -> int:
    return i


… .







#   
from valdec.decorators import async_validate as validate

@validate("s", "return", exclude=True)  #   "i"
async def foo(i: int, s: str) -> int:
    return int(i)

@validate("return")  #   
async def bar(i: int, s: str) -> int:
    return int(i)


2.3.2.1. -


/ , - ( , ), , , .







-:







def validator(
    annotations: Dict[str, Any],
    values: Dict[str, Any],
    is_replace: bool,
    extra: dict
) -> Optional[Dict[str, Any]]:


:







  • annotations



    — , .
  • values



    — , .
  • is_replace



    — , -, — .

    • True



      , . , , BaseModel



      , BaseModel



      , " ".
    • False



      , None



      , ( , , , BaseModel



      ).
  • extra



    — .


, validate



- pydantic.BaseModel



.







:







  • ( pydantic.BaseModel



    )
  • . .
  • ( ), , is_replace



    .


, , , . , , , .







- ( valdec



ValidatedDC



). : , pydantic.BaseModel



. , , "" .







2.3.2.2.


, "" :







from typing import List, Optional

from pydantic import BaseModel, StrictInt, StrictStr
from valdec.decorators import validate

class Profile(BaseModel):
    age: StrictInt
    city: StrictStr

class Student(BaseModel):
    name: StrictStr
    profile: Profile

@validate("group")
def func(group: Optional[List[Student]] = None):
    for student in group:
        assert isinstance(student, Student)
        assert isinstance(student.name, str)
        assert isinstance(student.profile.age, int)

data = [
    {"name": "Peter", "profile": {"age": 22, "city": "Samara"}},
    {"name": "Elena", "profile": {"age": 20, "city": "Kazan"}},
]

func(data)


assert'



.







:







@validate  #  
def func(group: Optional[List[Student]] = None, i: int) -> List[Student]:
    #...
    return [
        {"name": "Peter", "profile": {"age": 22, "city": "Samara"}},
        {"name": "Elena", "profile": {"age": 20, "city": "Kazan"}},
    ]


, , return



, Student



( ).







… . , , (, , ). :







from valdec.data_classes import Settings
from valdec.decorators import validate as _validate
from valdec.validator_pydantic import validator

custom_settings = Settings(
    validator=validator,     # -.
    is_replace_args=False,   #     
    is_replace_result=False, #     
    extra={}                 #  ,  
                             #   -
)
#   
def validate_without_replacement(*args, **kwargs):
    kwargs["settings"] = custom_settings
    return _validate(*args, **kwargs)

# 
@validate_without_replacement
def func(group: Optional[List[Student]] = None, i: int) -> List[Student]:
    #...
    return [
        {"name": "Peter", "profile": {"age": 22, "city": "Samara"}},
        {"name": "Elena", "profile": {"age": 20, "city": "Kazan"}},
    ]


func



, is_replace_result=False



. , is_replace_args=False



.







, .







— , , "" . , . , .







— , , , , ? — .







, -, — .







2.3.2.3.


:







from valdec.decorators import validate

@validate
def foo(i: int):
    assert isinstance(i, int)

foo("1")


. , .







, , validate



, - pydantic.BaseModel



. .2.3.1.1. .







, ( ), :







from valdec.decorators import validate
from pydantic import StrictInt

@validate
def foo(i: StrictInt):
    pass

foo("1")
# ...
# valdec.errors.ValidationArgumentsError: Validation error
# <class 'valdec.errors.ValidationError'>: 1 validation error for
# argument with the name of:
# i
#  value is not a valid integer (type=type_error.integer).


: , .







.







2.3.2.4.


  • valdec.errors.ValidationArgumentsError



    — ""
  • valdec.errors.ValidationReturnError



    —


. pydantic.BaseModel



.







2.3.3.



, - pydantic.BaseModel



.







C :







data_classes/base.py 









from pydantic import BaseModel, Extra

class BaseApi(BaseModel):
    """     api.
    """
    class Config:
        extra = Extra.forbid


2.3.4.



middleware, , , , :







from typing import List, Union
from valdec.decorators import async_validate as validate
from data_classes.person import PersonCreate, PersonInfo

@validate("data", "return")
async def create(
    data: Union[PersonCreate, List[PersonCreate]], storage: dict,
) -> Union[PersonInfo, List[PersonInfo]]:
    # ...
    return result


( ):







  • validate



    , ""
  • .


/ , .







: web.Request()



, , wep.Aplication()



. , , - , web.Request()



.







, :







data_classes/person.py 









from uuid import UUID
from pydantic import Field, StrictStr
from data_classes.base import BaseApi

class PersonCreate(BaseApi):
    """    .
    """
    name: StrictStr = Field(description=".", example="Oleg")

class PersonInfo(BaseApi):
    """   .
    """
    id: UUID = Field(description=".")
    name: StrictStr = Field(description=".")


2.3.5.



.2.3. .







.







data_classes/wraps.py 









from typing import Any, Optional
from pydantic import Field, StrictInt
from data_classes.base import BaseApi

_ID_DESCRIPTION = "   ."

class WrapRequest(BaseApi):
    """ .
    """
    data: Any = Field(description=" .", default=None)
    id: Optional[StrictInt] = Field(description=_ID_DESCRIPTION)

class WrapResponse(BaseApi):
    """ .
    """
    success: bool = Field(description=" .", default=True)
    result: Any = Field(description=" .")
    id: Optional[StrictInt] = Field(description=_ID_DESCRIPTION)


middleware



.







2.3.6. WrapsKwargsHandler



middleware



WrapsKwargsHandler



KwargsHandler



, ( ).







— run_handler



get_error_body



.







2.3.6.1.


:







async def run_handler(
        self, request: web.Request, handler: Callable, request_body: Any
    ) -> dict:

        id_ = None

        try:
            #    
            wrap_request = WrapRequest(**request_body)

        except Exception as error:
            message = f"{type(error).__name__} - {error}"
            raise InputDataValidationError(message)

        #   id  
        id_ = wrap_request.id
        request[KEY_NAME_FOR_ID] = id_

        try:
            result = await super().run_handler(
                request, handler, wrap_request.data
            )
        except ValidationArgumentsError as error:
            message = f"{type(error).__name__} - {error}"
            raise InputDataValidationError(message)

        #    
        wrap_response = WrapResponse(success=True, result=result, id=id_)

        return wrap_response.dict()


. InputDataValidationError



:







  • ( )
  • data



    id



  • id



    StrictInt



    None





id



, wrap_request.id



None



. data



. , , wrap_request.data



None



.







wrap_request.id



request



. ( ).







, wrap_request.data



(, wrap_request.data



python , json). , InputDataValidationError



valdec.errors.ValidationArgumentsError



.







, , WrapResponse



.







, . wrap_response



, ( ). , , , , , BaseApi



. , json. , "" WrapResponse.result



wrap_response



wrap_response.dict()



( ).







2.3.6.2.


:







def get_error_body(self, request: web.Request, error: Exception) -> dict:
        """         .
        """
        result = dict(error_type=str(type(error)), error_message=str(error))
        #         ,   
        #   ""
        response = dict(
            #   id    request .
            success=False, result=result, id=request.get(KEY_NAME_FOR_ID)
        )
        return response


( super()



result



), . .







2.3.7.



:








@validate("data", "return")
async def create(
    data: Union[PersonCreate, List[PersonCreate]], storage: dict,
) -> Union[PersonInfo, List[PersonInfo]]:
    # ...

@validate("data", "return")
async def read(storage: dict, req: web.Request, data: UUID) -> PersonInfo:
    # ...

@validate("info_id")
async def info(info_id: int, request: web.Request) -> Any:
    return f"info_id={info_id} and request={request}"


POST , — GET (, )







...
2.3.7.1. /create


  • №1:


{
    "data": [
        {
            "name": "Ivan"
        },
        {
            "name": "Oleg"
        }
    ],
    "id": 1
}


:







{
    "success": true,
    "result": [
        {
            "id": "af908a90-9157-4231-89f6-560eb6a8c4c0",
            "name": "Ivan"
        },
        {
            "id": "f7d554a0-1be9-4a65-bfc2-b89dbf70bb3c",
            "name": "Oleg"
        }
    ],
    "id": 1
}


  • №2:


{
    "data": {
        "name": "Eliza"
    },
    "id": 2
}


:







{
    "success": true,
    "result": {
        "id": "f3a45a19-acd0-4939-8e0c-e10743ff8e55",
        "name": "Eliza"
    },
    "id": 2
}


  • №3:


data 









{
    "data": 123,
    "id": 3
}


:







{
    "success": false,
    "result": {
        "error_type": "<class 'middlewares.exceptions.InputDataValidationError'>",
        "error_message": "ValidationArgumentsError - Validation error <class 'valdec.errors.ValidationError'>: 2 validation errors for argument with the name of:\ndata\n  value is not a valid dict (type=type_error.dict)\ndata\n  value is not a valid list (type=type_error.list)."
    },
    "id": 3
}


2.3.7.2. /read


  • №1:


{
    "data": "f3a45a19-acd0-4939-8e0c-e10743ff8e55",
    "id": 4
}


:







{
    "success": true,
    "result": {
        "id": "f3a45a19-acd0-4939-8e0c-e10743ff8e55",
        "name": "Eliza"
    },
    "id": 4


  • №2:


.







{
    "some_key": "f3a45a19-acd0-4939-8e0c-e10743ff8e55",
    "id": 5
}


:







{
    "success": false,
    "result": {
        "error_type": "<class 'middlewares.exceptions.InputDataValidationError'>",
        "error_message": "ValidationError - 1 validation error for WrapRequest\nsome_key\n  extra fields not permitted (type=value_error.extra)"
    },
    "id": null
}


2.3.7.3. /info/{info_id}


  • GET



    /info/123



    :


{}


:







{
    "success": true,
    "result": "info_id=123 and request=<Request GET /info/123 >",
    "id": null
}


3.



, WrapsKwargsHandler



, , . . pydantic.BaseModel



json-schema, ( , : swagger-, json- ).







. . , swagger



aiohttp



, ( ).







, aiohttp-swagger



( ), Union



.







aiohttp-swagger3



, , , sub_app



.







- , , , - , — .







4.



json middleware . . .







Vous pouvez créer des wrappers pour le contenu des demandes et des réponses. En outre, vous pouvez personnaliser la validation de manière flexible et l'appliquer uniquement là où elle est vraiment nécessaire.







Je suis convaincu que les exemples que j'ai proposés peuvent être mis en œuvre d'une autre manière. Mais j'espère que mes solutions, si elles ne sont pas totalement utiles, contribueront à en trouver d'autres, plus adaptées.







Merci pour votre temps. Je serais heureux de recevoir des commentaires et des clarifications.







Utilisation de MarkConv lors de la publication de l'article







More articles:


All Articles