src/blackd/middlewares.py

   1 from typing import Iterable, Awaitable, Callable
   2 from aiohttp.web_response import StreamResponse
   3 from aiohttp.web_request import Request
   4 from aiohttp.web_middlewares import middleware
   5
   6 Handler = Callable[[Request], Awaitable[StreamResponse]]
   7 Middleware = Callable[[Request, Handler], Awaitable[StreamResponse]]
   8
   9
  10 def cors(allow_headers: Iterable[str]) -> Middleware:
  11     @middleware
  12     async def impl(request: Request, handler: Handler) -> StreamResponse:
  13         is_options = request.method == "OPTIONS"
  14         is_preflight = is_options and "Access-Control-Request-Method" in request.headers
  15         if is_preflight:
  16             resp = StreamResponse()
  17         else:
  18             resp = await handler(request)
  19
  20         origin = request.headers.get("Origin")
  21         if not origin:
  22             return resp
  23
  24         resp.headers["Access-Control-Allow-Origin"] = "*"
  25         resp.headers["Access-Control-Expose-Headers"] = "*"
  26         if is_options:
  27             resp.headers["Access-Control-Allow-Headers"] = ", ".join(allow_headers)
  28             resp.headers["Access-Control-Allow-Methods"] = ", ".join(
  29                 ("OPTIONS", "POST")
  30             )
  31
  32         return resp
  33
  34     return impl  # type: ignore