http-server/response.py at main · ayan-04/http-server · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
"""
response.py - HTTP/1.1 Response Builder

Turns Python data (status code, headers, body) into the exact byte sequence
the HTTP/1.1 spec (RFC 7230) requires:

  HTTP/1.1 <status-code> <reason-phrase>\r\n
  <Header-Name>: <Header-Value>\r\n
  ...
  \r\n
  <body bytes>

We always send Content-Length so the client knows exactly when the message
ends.  Since we close the connection after every response (no Keep-Alive),
Content-Length is the only reliable body-delimiter available to us.
"""

import logging
from typing import Dict, Optional

logger = logging.getLogger(__name__)

# Reason phrases mandated by RFC 7231 §6
_REASON_PHRASES: Dict[int, str] = {
    200: "OK",
    201: "Created",
    204: "No Content",
    400: "Bad Request",
    404: "Not Found",
    405: "Method Not Allowed",
    500: "Internal Server Error",
}


class HTTPResponse:
    """
    Encapsulates all data needed to send an HTTP/1.1 response.

    Args:
        status_code:  Numeric HTTP status (e.g. 200, 404).
        body:         Response body as bytes.  Defaults to empty.
        content_type: Value for the Content-Type header.
        extra_headers: Any additional headers to include.
    """

    def __init__(
        self,
        status_code: int,
        body: bytes = b"",
        content_type: str = "text/plain; charset=utf-8",
        extra_headers: Optional[Dict[str, str]] = None,
    ) -> None:
        self.status_code: int = status_code
        self.body: bytes = body
        self.content_type: str = content_type
        self.extra_headers: Dict[str, str] = extra_headers or {}

    def to_bytes(self) -> bytes:
        """
        Serialise the response to the wire format expected by HTTP clients.

        The two \\r\\n sequences at the end of the header block are critical:
        the first terminates the last header line and the second is the blank
        line that signals «headers are done, body follows».  Omitting either
        one causes every HTTP client in existence to hang waiting for more data.
        """
        reason = _REASON_PHRASES.get(self.status_code, "Unknown")
        status_line = f"HTTP/1.1 {self.status_code} {reason}"

        # Build header lines
        headers: Dict[str, str] = {
            "Content-Type": self.content_type,
            # Content-Length is mandatory when we close the connection; it lets
            # the client determine body boundaries without chunked encoding.
            "Content-Length": str(len(self.body)),
            # Explicitly advertise that we will close; this matches our
            # single-request-per-connection design and prevents clients from
            # attempting to reuse the socket.
            "Connection": "close",
        }
        headers.update(self.extra_headers)

        header_lines = "\r\n".join(f"{k}: {v}" for k, v in headers.items())
        # Blank line (\r\n\r\n) separates headers from body per RFC 7230 §3
        preamble = f"{status_line}\r\n{header_lines}\r\n\r\n"

        logger.debug("Sending response: %s %s (%d bytes body)", self.status_code, reason, len(self.body))
        return preamble.encode("iso-8859-1") + self.body


# ── Convenience constructors ───────────────────────────────────────────────

def make_text_response(text: str, status_code: int = 200) -> HTTPResponse:
    """Create a plain-text HTTP response."""
    return HTTPResponse(
        status_code=status_code,
        body=text.encode("utf-8"),
        content_type="text/plain; charset=utf-8",
    )


def make_json_response(body_bytes: bytes, status_code: int = 200) -> HTTPResponse:
    """Create an application/json HTTP response from pre-serialised bytes."""
    return HTTPResponse(
        status_code=status_code,
        body=body_bytes,
        content_type="application/json; charset=utf-8",
    )


def make_html_response(html_bytes: bytes, status_code: int = 200) -> HTTPResponse:
    """Create a text/html HTTP response."""
    return HTTPResponse(
        status_code=status_code,
        body=html_bytes,
        content_type="text/html; charset=utf-8",
    )


def make_error_response(status_code: int, message: str = "") -> HTTPResponse:
    """Create a plain-text error response."""
    reason = _REASON_PHRASES.get(status_code, "Error")
    body_text = message or f"{status_code} {reason}"
    return HTTPResponse(
        status_code=status_code,
        body=body_text.encode("utf-8"),
        content_type="text/plain; charset=utf-8",
    )