text: import warnings
  
  from functools import partial
  from mimetypes import guess_type
  from os import path
  from urllib.parse import quote_plus
  
  from sanic.compat import Header, open_async
  from sanic.cookies import CookieJar
  from sanic.headers import format_http1, format_http1_response
  from sanic.helpers import has_message_body, remove_entity_headers
  
  
  try:
      from ujson import dumps as json_dumps
  except ImportError:
      # This is done in order to ensure that the JSON response is
      # kept consistent across both ujson and inbuilt json usage.
      from json import dumps
  
      json_dumps = partial(dumps, separators=(",", ":"))
  
  
  class BaseHTTPResponse:
      def _encode_body(self, data):
          return data.encode() if hasattr(data, "encode") else data
  
      def _parse_headers(self):
          return format_http1(self.headers.items())
  
      @property
      def cookies(self):
          if self._cookies is None:
              self._cookies = CookieJar(self.headers)
          return self._cookies
  
      def get_headers(
          self,
          version="1.1",
          keep_alive=False,
          keep_alive_timeout=None,
          body=b"",
      ):
          """.. deprecated:: 20.3:
             This function is not public API and will be removed."""
  
          # self.headers get priority over content_type
          if self.content_type and "Content-Type" not in self.headers:
              self.headers["Content-Type"] = self.content_type
  
          if keep_alive:
              self.headers["Connection"] = "keep-alive"
              if keep_alive_timeout is not None:
                  self.headers["Keep-Alive"] = keep_alive_timeout
          else:
              self.headers["Connection"] = "close"
  
          if self.status in (304, 412):
              self.headers = remove_entity_headers(self.headers)
  
          return format_http1_response(self.status, self.headers.items(), body)
  
  
  class StreamingHTTPResponse(BaseHTTPResponse):
      __slots__ = (
          "protocol",
          "streaming_fn",
          "status",
          "content_type",
          "headers",
          "chunked",
          "_cookies",
      )
  
      def __init__(
          self,
          streaming_fn,
          status=200,
          headers=None,
          content_type="text/plain; charset=utf-8",
          chunked=True,
      ):
          self.content_type = content_type
          self.streaming_fn = streaming_fn
          self.status = status
          self.headers = Header(headers or {})
          self.chunked = chunked
          self._cookies = None
          self.protocol = None
  
      async def write(self, data):
          """Writes a chunk of data to the streaming response.
  
          :param data: str or bytes-ish data to be written.
          """
          data = self._encode_body(data)
  
          if self.chunked:
              await self.protocol.push_data(b"%x\r\n%b\r\n" % (len(data), data))
          else:
              await self.protocol.push_data(data)
          await self.protocol.drain()
  
      async def stream(
          self, version="1.1", keep_alive=False, keep_alive_timeout=None
      ):
          """Streams headers, runs the `streaming_fn` callback that writes
          content to the response body, then finalizes the response body.
          """
          if version != "1.1":
              self.chunked = False
          headers = self.get_headers(
              version,
              keep_alive=keep_alive,
              keep_alive_timeout=keep_alive_timeout,
          )
          await self.protocol.push_data(headers)
          await self.protocol.drain()
          await self.streaming_fn(self)
          if self.chunked:
              await self.protocol.push_data(b"0\r\n\r\n")
          # no need to await drain here after this write, because it is the
          # very last thing we write and nothing needs to wait for it.
  
      def get_headers(
          self, version="1.1", keep_alive=False, keep_alive_timeout=None
      ):
          if self.chunked and version == "1.1":
              self.headers["Transfer-Encoding"] = "chunked"
              self.headers.pop("Content-Length", None)
  
          return super().get_headers(version, keep_alive, keep_alive_timeout)
  
  
  class HTTPResponse(BaseHTTPResponse):
      __slots__ = ("body", "status", "content_type", "headers", "_cookies")
  
      def __init__(
          self,
          body=None,
          status=200,
          headers=None,
          content_type=None,
          body_bytes=b"",
      ):
          self.content_type = content_type
          self.body = body_bytes if body is None else self._encode_body(body)
          self.status = status
          self.headers = Header(headers or {})
          self._cookies = None
  
          if body_bytes:
              warnings.warn(
                  "Parameter `body_bytes` is deprecated, use `body` instead",
                  DeprecationWarning,
              )
  
      def output(self, version="1.1", keep_alive=False, keep_alive_timeout=None):
          body = b""
          if has_message_body(self.status):
              body = self.body
              self.headers["Content-Length"] = self.headers.get(
                  "Content-Length", len(self.body)
              )
  
          return self.get_headers(version, keep_alive, keep_alive_timeout, body)
  
      @property
      def cookies(self):
          if self._cookies is None:
              self._cookies = CookieJar(self.headers)
          return self._cookies
  
  
  def empty(status=204, headers=None):
      """
      Returns an empty response to the client.
  
      :param status Response code.
      :param headers Custom Headers.
      """
      return HTTPResponse(body=b"", status=status, headers=headers)
  
  
  def json(
      body,
      status=200,
      headers=None,
      content_type="application/json",
      dumps=json_dumps,
      **kwargs,
  ):
      """
      Returns response object with body in json format.
  
      :param body: Response data to be serialized.
      :param status: Response code.
      :param headers: Custom Headers.
      :param kwargs: Remaining arguments that are passed to the json encoder.
      """
      return HTTPResponse(
          dumps(body, **kwargs),
          headers=headers,
          status=status,
          content_type=content_type,
      )
  
  
  def text(
      body, status=200, headers=None, content_type="text/plain; charset=utf-8"
  ):
      """
      Returns response object with body in text format.
  
      :param body: Response data to be encoded.
      :param status: Response code.
      :param headers: Custom Headers.
      :param content_type: the content type (string) of the response
      """
      if not isinstance(body, str):
          warnings.warn(
              "Types other than str will be deprecated in future versions for"
              f" response.text, got type {type(body).__name__})",
              DeprecationWarning,
          )
      # Type conversions are deprecated and quite b0rked but still supported for
      # text() until applications get fixed. This try-except should be removed.
      try:
          # Avoid repr(body).encode() b0rkage for body that is already encoded.
          # memoryview used only to test bytes-ishness.
          with memoryview(body):
              pass
      except TypeError:
          body = f"{body}"  # no-op if body is already str
      return HTTPResponse(
          body, status=status, headers=headers, content_type=content_type
      )
  
  
  def raw(
      body, status=200, headers=None, content_type="application/octet-stream"
  ):
      """
      Returns response object without encoding the body.
  
      :param body: Response data.
      :param status: Response code.
      :param headers: Custom Headers.
      :param content_type: the content type (string) of the response.
      """
      return HTTPResponse(
          body=body, status=status, headers=headers, content_type=content_type,
      )
  
  
  def html(body, status=200, headers=None):
      """
      Returns response object with body in html format.
  
      :param body: str or bytes-ish, or an object with __html__ or _repr_html_.
      :param status: Response code.
      :param headers: Custom Headers.
      """
      if hasattr(body, "__html__"):
          body = body.__html__()
      elif hasattr(body, "_repr_html_"):
          body = body._repr_html_()
      return HTTPResponse(
          body,
          status=status,
          headers=headers,
          content_type="text/html; charset=utf-8",
      )
  
  
  async def file(
      location,
      status=200,
      mime_type=None,
      headers=None,
      filename=None,
      _range=None,
  ):
      """Return a response object with file data.
  
      :param location: Location of file on system.
      :param mime_type: Specific mime_type.
      :param headers: Custom Headers.
      :param filename: Override filename.
      :param _range:
      """
      headers = headers or {}
      if filename:
          headers.setdefault(
              "Content-Disposition", f'attachment; filename="{filename}"'
          )
      filename = filename or path.split(location)[-1]
  
      async with await open_async(location, mode="rb") as f:
          if _range:
              await f.seek(_range.start)
              out_stream = await f.read(_range.size)
              headers[
                  "Content-Range"
              ] = f"bytes {_range.start}-{_range.end}/{_range.total}"
              status = 206
          else:
              out_stream = await f.read()
  
      mime_type = mime_type or guess_type(filename)[0] or "text/plain"
      return HTTPResponse(
          body=out_stream,
          status=status,
          headers=headers,
          content_type=mime_type,
      )
  
  
  async def file_stream(
      location,
      status=200,
      chunk_size=4096,
      mime_type=None,
      headers=None,
      filename=None,
      chunked=True,
      _range=None,
  ):
      """Return a streaming response object with file data.
  
      :param location: Location of file on system.
      :param chunk_size: The size of each chunk in the stream (in bytes)
      :param mime_type: Specific mime_type.
      :param headers: Custom Headers.
      :param filename: Override filename.
      :param chunked: Enable or disable chunked transfer-encoding
      :param _range:
      """
      headers = headers or {}
      if filename:
          headers.setdefault(
              "Content-Disposition", f'attachment; filename="{filename}"'
          )
      filename = filename or path.split(location)[-1]
      mime_type = mime_type or guess_type(filename)[0] or "text/plain"
      if _range:
          start = _range.start
          end = _range.end
          total = _range.total
  
          headers["Content-Range"] = f"bytes {start}-{end}/{total}"
          status = 206
  
      async def _streaming_fn(response):
          async with await open_async(location, mode="rb") as f:
              if _range:
                  await f.seek(_range.start)
                  to_send = _range.size
                  while to_send > 0:
                      content = await f.read(min((_range.size, chunk_size)))
                      if len(content) < 1:
                          break
                      to_send -= len(content)
                      await response.write(content)
              else:
                  while True:
                      content = await f.read(chunk_size)
                      if len(content) < 1:
                          break
                      await response.write(content)
  
      return StreamingHTTPResponse(
          streaming_fn=_streaming_fn,
          status=status,
          headers=headers,
          content_type=mime_type,
          chunked=chunked,
      )
  
  
  def stream(
      streaming_fn,
      status=200,
      headers=None,
      content_type="text/plain; charset=utf-8",
      chunked=True,
  ):
      """Accepts an coroutine `streaming_fn` which can be used to
      write chunks to a streaming response. Returns a `StreamingHTTPResponse`.
  
      Example usage::
  
          @app.route("/")
          async def index(request):
              async def streaming_fn(response):
                  await response.write('foo')
                  await response.write('bar')
  
              return stream(streaming_fn, content_type='text/plain')
  
      :param streaming_fn: A coroutine accepts a response and
          writes content to that response.
      :param mime_type: Specific mime_type.
      :param headers: Custom Headers.
      :param chunked: Enable or disable chunked transfer-encoding
      """
      return StreamingHTTPResponse(
          streaming_fn,
          headers=headers,
          content_type=content_type,
          status=status,
          chunked=chunked,
      )
  
  
  def redirect(
      to, headers=None, status=302, content_type="text/html; charset=utf-8"
  ):
      """Abort execution and cause a 302 redirect (by default).
  
      :param to: path or fully qualified URL to redirect to
      :param headers: optional dict of headers to include in the new request
      :param status: status code (int) of the new request, defaults to 302
      :param content_type: the content type (string) of the response
      :returns: the redirecting Response
      """
      headers = headers or {}
  
      # URL Quote the URL before redirecting
      safe_to = quote_plus(to, safe=":/%#?&=@[]!$&'()*+,;")
  
      # According to RFC 7231, a relative URI is now permitted.
      headers["Location"] = safe_to
  
      return HTTPResponse(
          status=status, headers=headers, content_type=content_type
      )
  