wsgi.py 5.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154
  1. import io
  2. import math
  3. import sys
  4. import typing
  5. import warnings
  6. import anyio
  7. from anyio.abc import ObjectReceiveStream, ObjectSendStream
  8. from starlette.types import Receive, Scope, Send
  9. warnings.warn(
  10. "starlette.middleware.wsgi is deprecated and will be removed in a future release. "
  11. "Please refer to https://github.com/abersheeran/a2wsgi as a replacement.",
  12. DeprecationWarning,
  13. )
  14. def build_environ(scope: Scope, body: bytes) -> typing.Dict[str, typing.Any]:
  15. """
  16. Builds a scope and request body into a WSGI environ object.
  17. """
  18. script_name = scope.get("root_path", "").encode("utf8").decode("latin1")
  19. path_info = scope["path"].encode("utf8").decode("latin1")
  20. if path_info.startswith(script_name):
  21. path_info = path_info[len(script_name) :]
  22. environ = {
  23. "REQUEST_METHOD": scope["method"],
  24. "SCRIPT_NAME": script_name,
  25. "PATH_INFO": path_info,
  26. "QUERY_STRING": scope["query_string"].decode("ascii"),
  27. "SERVER_PROTOCOL": f"HTTP/{scope['http_version']}",
  28. "wsgi.version": (1, 0),
  29. "wsgi.url_scheme": scope.get("scheme", "http"),
  30. "wsgi.input": io.BytesIO(body),
  31. "wsgi.errors": sys.stdout,
  32. "wsgi.multithread": True,
  33. "wsgi.multiprocess": True,
  34. "wsgi.run_once": False,
  35. }
  36. # Get server name and port - required in WSGI, not in ASGI
  37. server = scope.get("server") or ("localhost", 80)
  38. environ["SERVER_NAME"] = server[0]
  39. environ["SERVER_PORT"] = server[1]
  40. # Get client IP address
  41. if scope.get("client"):
  42. environ["REMOTE_ADDR"] = scope["client"][0]
  43. # Go through headers and make them into environ entries
  44. for name, value in scope.get("headers", []):
  45. name = name.decode("latin1")
  46. if name == "content-length":
  47. corrected_name = "CONTENT_LENGTH"
  48. elif name == "content-type":
  49. corrected_name = "CONTENT_TYPE"
  50. else:
  51. corrected_name = f"HTTP_{name}".upper().replace("-", "_")
  52. # HTTPbis say only ASCII chars are allowed in headers, but we latin1 just in
  53. # case
  54. value = value.decode("latin1")
  55. if corrected_name in environ:
  56. value = environ[corrected_name] + "," + value
  57. environ[corrected_name] = value
  58. return environ
  59. class WSGIMiddleware:
  60. def __init__(self, app: typing.Callable[..., typing.Any]) -> None:
  61. self.app = app
  62. async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
  63. assert scope["type"] == "http"
  64. responder = WSGIResponder(self.app, scope)
  65. await responder(receive, send)
  66. class WSGIResponder:
  67. stream_send: ObjectSendStream[typing.MutableMapping[str, typing.Any]]
  68. stream_receive: ObjectReceiveStream[typing.MutableMapping[str, typing.Any]]
  69. def __init__(self, app: typing.Callable[..., typing.Any], scope: Scope) -> None:
  70. self.app = app
  71. self.scope = scope
  72. self.status = None
  73. self.response_headers = None
  74. self.stream_send, self.stream_receive = anyio.create_memory_object_stream(
  75. math.inf
  76. )
  77. self.response_started = False
  78. self.exc_info: typing.Any = None
  79. async def __call__(self, receive: Receive, send: Send) -> None:
  80. body = b""
  81. more_body = True
  82. while more_body:
  83. message = await receive()
  84. body += message.get("body", b"")
  85. more_body = message.get("more_body", False)
  86. environ = build_environ(self.scope, body)
  87. async with anyio.create_task_group() as task_group:
  88. task_group.start_soon(self.sender, send)
  89. async with self.stream_send:
  90. await anyio.to_thread.run_sync(self.wsgi, environ, self.start_response)
  91. if self.exc_info is not None:
  92. raise self.exc_info[0].with_traceback(self.exc_info[1], self.exc_info[2])
  93. async def sender(self, send: Send) -> None:
  94. async with self.stream_receive:
  95. async for message in self.stream_receive:
  96. await send(message)
  97. def start_response(
  98. self,
  99. status: str,
  100. response_headers: typing.List[typing.Tuple[str, str]],
  101. exc_info: typing.Any = None,
  102. ) -> None:
  103. self.exc_info = exc_info
  104. if not self.response_started:
  105. self.response_started = True
  106. status_code_string, _ = status.split(" ", 1)
  107. status_code = int(status_code_string)
  108. headers = [
  109. (name.strip().encode("ascii").lower(), value.strip().encode("ascii"))
  110. for name, value in response_headers
  111. ]
  112. anyio.from_thread.run(
  113. self.stream_send.send,
  114. {
  115. "type": "http.response.start",
  116. "status": status_code,
  117. "headers": headers,
  118. },
  119. )
  120. def wsgi(
  121. self,
  122. environ: typing.Dict[str, typing.Any],
  123. start_response: typing.Callable[..., typing.Any],
  124. ) -> None:
  125. for chunk in self.app(environ, start_response):
  126. anyio.from_thread.run(
  127. self.stream_send.send,
  128. {"type": "http.response.body", "body": chunk, "more_body": True},
  129. )
  130. anyio.from_thread.run(
  131. self.stream_send.send, {"type": "http.response.body", "body": b""}
  132. )