src/blackd/middlewares.py

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