[main] WIP Restructuring the codebase

src/common/constants.py

@@ -0,0 +1,16 @@
+from typing import Dict, List
+
+PROTOCOL: str = "HTCPCP/1.0"
+
+COFFEE_SCHEMES: List[str] = [
+    "coffee", "koffie", "%D9%82%D9%87%D9%88%D8%A9", "caf%C3%E8", "%E5%92%96%E5%95%A1",
+    "k%C3%A1va", "kaffe", "kafo", "kohv", "kahvi", "%CE%BA%CE%B1%CF%86%CE%AD",
+    "%E0%A4%95%E0%A5%8C%E0%A4%AB%E0%A5%80", "%EC%BB%A4%ED%94%BC", "%D0%BA%D0%BE%D1%84%D0%B5"
+]
+
+RESPONSES: Dict[int, str] = {
+    200: "OK",
+    400: "BAD REQUEST"
+}
+
+ALLOWED_METHODS: List[str] = ["GET", "BREW", "POST"]

src/common/htcpcp_error.py

@@ -0,0 +1,4 @@
+class HtpcpError(Exception):
+    def __init__(self, message, status):
+        super().__init__(message)
+        self.status = status

src/common/request.py

@@ -0,0 +1,80 @@
+from typing import Dict, Optional
+from .htcpcp_error import HtpcpError
+from .constants import PROTOCOL, COFFEE_SCHEMES, ALLOWED_METHODS
+
+
+class Request:
+    def __init__(
+        self,
+        method: Optional[str] = None,
+        url: Optional[str] = None,
+        protocol: str = PROTOCOL,
+        headers: Optional[Dict[str, str]] = None,
+        payload: Optional[str] = None,
+        raw_request: Optional[str] = None
+    ) -> None:
+        self.method = method or ""
+        self.url = url or ""
+        self.protocol = protocol
+        self.headers: Dict[str, str] = headers or {}
+        self.payload = payload
+
+        if raw_request:  # If a raw request is provided, parse it.
+            self._parse_request(raw_request)
+        # Ensure required fields for new requests.
+        elif not (self.method and self.url):
+            raise ValueError(
+                "Either 'raw_request' or 'method' and 'url' must be provided.")
+
+    def _parse_request(self, raw_request: str) -> None:
+        """Parse a raw request string into components."""
+        lines = raw_request.split("\n")
+
+        if len(lines) < 1:
+            raise HtpcpError("Invalid request: Request is empty.", 400)
+
+        request_line = lines[0].strip()
+        self._parse_request_line(request_line)
+
+        header_and_body = raw_request.split("\n\n", 1)
+        self._parse_headers(header_and_body[0])
+
+        if len(header_and_body) == 2:
+            self.payload = header_and_body[1].strip()
+
+    def _parse_request_line(self, request_line: str) -> None:
+        """Parse the request line (e.g., 'GET /coffee HTCPCP/1.0')."""
+        parts = request_line.split(" ")
+
+        if len(parts) != 3:
+            raise HtpcpError("Invalid request line format.", 400)
+
+        self.method, self.url, self.protocol = parts
+
+        if self.protocol != PROTOCOL:
+            raise HtpcpError(f"Invalid protocol: {self.protocol}", 400)
+
+    def _parse_headers(self, header_block: str) -> None:
+        """Parse headers from the header block."""
+        header_lines = header_block.split("\n")[1:]
+
+        for line in header_lines:
+            if ":" in line:
+                key, value = map(str.strip, line.split(":", 1))
+                self.headers[key.lower()] = value
+
+    def validate(self) -> None:
+        """Validate the parsed or constructed request."""
+        if not any(self.url.startswith(f"{scheme}://") for scheme in COFFEE_SCHEMES):
+            raise HtpcpError("Invalid URL scheme.", 400)
+
+        if self.method not in ALLOWED_METHODS:
+            raise HtpcpError(f"Unsupported method: {self.method}", 400)
+
+    def __str__(self) -> str:
+        """Convert the request to its raw string representation."""
+        request_line = f"{self.method} {self.url} {self.protocol}"
+        headers = "\n".join(f"{key}: {value}" for key,
+                            value in self.headers.items())
+        body = f"\n\n{self.payload}" if self.payload else ""
+        return f"{request_line}\n{headers}{body}"

src/common/response.py

@@ -0,0 +1,84 @@
+from typing import Dict, Optional
+from .constants import RESPONSES, PROTOCOL
+
+
+class Response:
+    def __init__(
+        self,
+        status: Optional[int] = None,
+        body: Optional[str] = None,
+        headers: Optional[Dict[str, str]] = None,
+        raw_response: Optional[str] = None
+    ) -> None:
+        self.status = status or 200
+        self.body = body or ""
+        self.headers: Dict[str, str] = headers or {}
+
+        if raw_response:  # Parse the raw response string if provided.
+            self._parse_response(raw_response)
+        elif self.status not in RESPONSES:
+            raise ValueError(f"Invalid status code: {self.status}")
+
+    def _parse_response(self, raw_response: str) -> None:
+        """Parse a raw HTTP/HTCPCP response string."""
+        lines = raw_response.split("\r\n")
+
+        if not lines or len(lines) < 1:
+            raise ValueError(
+                "Invalid response: Response is empty or malformed.")
+
+        # Parse status line
+        status_line = lines[0]
+        self._parse_status_line(status_line)
+
+        # Separate headers and body
+        header_and_body = raw_response.split("\r\n\r\n", 1)
+        self._parse_headers(header_and_body[0])
+
+        if len(header_and_body) == 2:
+            self.body = header_and_body[1].strip()
+
+    def _parse_status_line(self, status_line: str) -> None:
+        """Parse the status line (e.g., 'HTCPCP/1.0 200 OK')."""
+        parts = status_line.split(" ", 2)
+
+        if len(parts) < 3 or parts[0] != PROTOCOL:
+            raise ValueError(f"Invalid status line: {status_line}")
+
+        try:
+            self.status = int(parts[1])
+        except ValueError as e:
+            print(e)
+            raise e
+
+        if self.status not in RESPONSES:
+            raise ValueError(f"Unknown status code: {self.status}")
+
+    def _parse_headers(self, header_block: str) -> None:
+        """Parse headers from the header block."""
+        header_lines = header_block.split("\r\n")[1:]
+
+        for line in header_lines:
+            if ":" in line:
+                key, value = map(str.strip, line.split(":", 1))
+                self.headers[key.lower()] = value
+
+    def validate(self) -> None:
+        """Validate the constructed or parsed response."""
+        if self.status not in RESPONSES:
+            raise ValueError(f"Invalid status code: {self.status}")
+
+        if not self.headers:
+            raise ValueError("Response must have at least one header.")
+
+        if "content-length" in self.headers and not self.body:
+            raise ValueError(
+                "Content-Length header specified but no body provided.")
+
+    def __str__(self) -> str:
+        """Convert the response to its raw string representation."""
+        status_line = f"{PROTOCOL} {self.status} {RESPONSES[self.status]}"
+        headers = "\r\n".join(f"{key}: {value}" for key,
+                              value in self.headers.items())
+        body = f"\r\n\r\n{self.body}" if self.body else ""
+        return f"{status_line}\r\n{headers}{body}"

src/server/main.py

@@ -46,26 +46,22 @@
 #    - Log incoming requests, responses, and errors for debugging.
 import re
 import socket
-from typing import Dict, Optional
-from datetime import datetime, timezone
+
+from utils.http_date import get_http_date
+from ..common.htcpcp_error import HtpcpError
+
+from ..common.constants import PROTOCOL, RESPONSES
+from ..common.request import Request
+from ..common.response import Response
 
 # Constants
-PROTOCOL = "HTCPCP/1.0"
-ALLOWED_METHODS = ["GET", "BREW", "POST"]
+
+
 NAME = "CoffeeMaker"
 HOST = '127.0.0.1'
 PORT = 6969
 
-COFFEE_SCHEMES = [
-    "coffee", "koffie", "%D9%82%D9%87%D9%88%D8%A9", "caf%C3%E8", "%E5%92%96%E5%95%A1",
-    "k%C3%A1va", "kaffe", "kafo", "kohv", "kahvi", "%CE%BA%CE%B1%CF%86%CE%AD",
-    "%E0%A4%95%E0%A5%8C%E0%A4%AB%E0%A5%80", "%EC%BB%A4%ED%94%BC", "%D0%BA%D0%BE%D1%84%D0%B5"
-]
 
-RESPONSES: Dict[int, str] = {
-    200: "OK",
-    400: "BAD REQUEST"
-}
 
 ERRORS = {
     "SCHEME_MISMATCH": "HTTP/1.1 400 Bad Request\r\nContent-Type: text/plain\r\n\r\nInvalid scheme. Use coffee://.",
@@ -73,90 +69,6 @@ ERRORS = {
 }
 
 
-# region Utils
-
-def get_http_date() -> str:
-    """
-    Returns the current date and time in HTTP-date format (RFC 7231).
-    Example: Thu, 19 Dec 2024 06:56:48 GMT
-    """
-    now = datetime.now(timezone.utc)
-    return now.strftime("%a, %d %b %Y %H:%M:%S GMT")
-
-# endregion
-
-
-class HtpcpError(Exception):
-    def __init__(self, message, status):
-        super().__init__(message)
-        self.status = status
-
-
-class Request:
-    def __init__(self, raw_request: str):
-        self.method = ""
-        self.url = ""
-        self.protocol = ""
-        self.headers: Dict[str, str] = {}
-        self.payload: Optional[str] = None
-
-        self._parse_request(raw_request)
-
-    def _parse_request(self, raw_request: str):
-        lines = raw_request.split("\n")
-
-        if len(lines) < 1:
-            raise HtpcpError("Invalid request: Request is empty.", 400)
-
-        request_line = lines[0].strip()
-        self._parse_request_line(request_line)
-
-        header_and_body = raw_request.split("\n\n", 1)
-        self._parse_headers(header_and_body[0])
-
-        if len(header_and_body) == 2:
-            self.payload = header_and_body[1].strip()
-
-    def _parse_request_line(self, request_line: str):
-        parts = request_line.split(" ")
-
-        if len(parts) != 3:
-            raise HtpcpError("Invalid request line format.", 400)
-
-        self.method, self.url, self.protocol = parts
-
-        if self.protocol != PROTOCOL:
-            raise HtpcpError(f"Invalid protocol: {self.protocol}", 400)
-
-    def _parse_headers(self, header_block: str):
-        header_lines = header_block.split("\n")[1:]
-
-        for line in header_lines:
-            if ":" in line:
-                key, value = map(str.strip, line.split(":", 1))
-                self.headers[key.lower()] = value.lower()
-
-    def validate(self):
-        if not any(self.url.startswith(f"{scheme}://") for scheme in COFFEE_SCHEMES):
-            raise HtpcpError("Invalid URL scheme.", 400)
-
-        if self.method not in ALLOWED_METHODS:
-            raise HtpcpError(f"Unsupported method: {self.method}", 400)
-
-
-class Response():
-    status: int = 200
-    
-    body: str = ""
-
-    def __str__(self):
-        status_line = f"{PROTOCOL} {self.status} {RESPONSES[self.status]}"
-        date_line = get_http_date()
-        server_line = f"Server: {NAME}"
-
-        return f"{status_line}\r\nDate: {date_line}\r\n{server_line}"
-
-
 class Server():
     def __init__(self, host: str, port: int):
         self.host = host

src/server/utils/http_date.py

@@ -0,0 +1,9 @@
+from datetime import datetime, timezone
+
+def get_http_date() -> str:
+    """
+    Returns the current date and time in HTTP-date format (RFC 7231).
+    Example: Thu, 19 Dec 2024 06:56:48 GMT
+    """
+    now = datetime.now(timezone.utc)
+    return now.strftime("%a, %d %b %Y %H:%M:%S GMT")