Method send

uvicorn/protocols/http/httptools_impl.py:462–560

(self, message: ASGISendEvent)

    class="cm"># ASGI interface
    async def send(self, message: ASGISendEvent) -> None:
        if self.flow.write_paused and not self.disconnected:
            await self.flow.drain()  class="cm"># pragma: full coverage

        if self.disconnected:
            return  class="cm"># pragma: full coverage

        if not self.response_started:
            class="cm"># Sending response status line and headers
            if message[class="st">"type"] != class="st">"http.response.start":
                raise RuntimeError(fclass="st">"Expected ASGI message &class="cm">#x27;http.response.start', but got '{message['type']}'.")

            self.response_started = True
            self.waiting_for_100_continue = False

            status_code = message[class="st">"status"]
            headers = self.default_headers + list(message.get(class="st">"headers", []))

            if CLOSE_HEADER in self.scope[class="st">"headers"] and CLOSE_HEADER not in headers:
                headers = headers + [CLOSE_HEADER]

            if self.access_log:
                self.access_logger.info(
                    &class="cm">#x27;%s - class="st">"%s %s HTTP/%s" %d',
                    get_client_addr(self.scope),
                    self.scope[class="st">"method"],
                    get_path_with_query_string(self.scope),
                    self.scope[class="st">"http_version"],
                    status_code,
                )

            class="cm"># Write response status line and headers
            content = [STATUS_LINE[status_code]]

            for name, value in headers:
                if HEADER_RE.search(name):
                    raise RuntimeError(class="st">"Invalid HTTP header name.")  class="cm"># pragma: full coverage
                if HEADER_VALUE_RE.search(value):
                    raise RuntimeError(class="st">"Invalid HTTP header value.")

                name = name.lower()
                if name == bclass="st">"content-length" and self.chunked_encoding is None:
                    self.expected_content_length = int(value.decode())
                    self.chunked_encoding = False
                elif name == bclass="st">"transfer-encoding" and value.lower() == bclass="st">"chunked":
                    self.expected_content_length = 0
                    self.chunked_encoding = True
                elif name == bclass="st">"connection" and value.lower() == bclass="st">"close":
                    self.keep_alive = False
                content.extend([name, bclass="st">": ", value, bclass="st">"\r\n"])

            if self.chunked_encoding is None and self.scope[class="st">"method"] != class="st">"HEAD" and status_code not in (204, 304):
                class="cm"># Neither content-length nor transfer-encoding specified
                self.chunked_encoding = True
                content.append(bclass="st">"transfer-encoding: chunked\r\n")

            content.append(bclass="st">"\r\n")
            self.transport.write(bclass="st">"".join(content))