GitLocker: The Coding Marketplace

Description:

middletools 0.1.2

Middletools

This is a python library that allows you to integrate middlewares-based system to your project. It contains base tools for creating and running middlewares with async-await style
Installation
PyPI
python -m pip install middletools

GitHub
python -m pip install https://github.com/deknowny/middlewares/archive/main.zip

Usage
The main idea is give an ability just passing the middlewares and inbox/outbox payload values in a few methods instead of running and saving middlewares state by hand
Standard case: a function runs RESTful API routers and requires a middleware that checks
a header in client's request

There are 2 endpoints for an abstract GET and POST methods
# Some abstract router
@router.get("/")
async def route_get(request):
return 200, {"response": "some content"}

@router.post("/")
async def route_post(request):
return 201, {"response": "ok"}

In the core of web framework you used a function like this that just call all routers
class Router:
...
...

async def call_routers(self, request):
for router in self.routers:
... # Pass request to routers and check it matched

middlewares library allows you easy integrate middleware system to your call_routers

Create middleware function
import middletools

...
...

# Adding a middleware handler to an abstract
@router.add_middleware
async def my_middleware(
request: SomeRequest, call_next: middletools.types.CallNext
) -> SomeResponse:
# Just check if header exists, id not set the default value
if "X-Required-Header" not in request.headers:
request.header["X-Required-Header"] = "default"
response = await call_next()
return response

Here we add a header to client request if clint didn't do it. Then await call_next() give control to other middlewares or to our call_routers handler and response from this is the value call_next() returns

call_routers should looks like this
import typing

import middletools

class Router:
# You can use generics to describe middleware hand;er
middlewares: typing.List[
middletools.types.MiddlewareHandler[
SomeRequest, SomeResponse
]
]
...
...

async def call_routers(self, request):
read_afterwords = await middletools.read_forewords(
*self.middlewares, inbox_value=request
)
for router in self.routers:
... # Pass request to routers and check it matched
response = ...
await read_afterwords(response)
break

middlewares.read_forewords run middlewares until every one of them give control with await call_next().
When we do all our stuff and get the router response we can call await read_afterwords(response) and run all middlewares completely.
Notes
If a middleware doesn't call call_next() it raises middlewares.CallNextNotUsedError. It means that the middleware forcibly decline middlewares handlers and response should be sent immediately without routers running. call_routers should looks like this:
import middletools

async def call_routers(self, request):
try:
read_afterwords = await middletools.read_forewords(
*self.middlewares, inbox_value=request
)
for router in self.routers:
... # Pass request to routers and check it matched
response = ...
await read_afterwords(response)
return response
except middletools.CallNextNotUsedError:
return SomeBadResponseBecauseNotRouted(400, "Require a header!")

If a middleware doesn't return anything, middlewares dispatching declined forcibly too but after routers handled. (Return nothing means there isn't any return or return None used). It raises middlewares.NothingReturnedError
import middletools

async def call_routers(self, request):
try:
read_afterwords = await middletools.read_forewords(
*self.middlewares, inbox_value=request
)
for router in self.routers:
... # Pass request to routers and check it matched
response = ...
await read_afterwords(response)
return response
except middletools.CallNextNotUsedError:
return SomeBadResponseBecauseNotRouted(400, "Require a header!")
except middletools.NothingReturnedError:
return SomeBadResponseBecauseMiddlewareDntReturnResponse(
500, "Oops, internal server error"
)