wsgi.py 5.2 KB

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