feat: base server code

src/params.py

@@ -0,0 +1,46 @@
+import argparse
+import dataclasses
+import os
+import os.path
+
+from . import project
+
+
+@dataclasses.dataclass
+class Parameters:
+    port: int
+    host: str
+    data_dir: str
+    bind: str
+
+    @classmethod
+    def from_namespace(cls, args: argparse.Namespace) -> "Parameters":
+        return Parameters(**vars(args))
+
+
+def parse_parameters() -> Parameters:
+    parser = argparse.ArgumentParser(
+        project.get_name(), description=project.get_description()
+    )
+    parser.add_argument(
+        "-p", "--port", type=int, default=8080, help="server port (default: 8080)"
+    )
+    parser.add_argument(
+        "--host",
+        default="localhost",
+        help="server default host (default: localhost)",
+    )
+    parser.add_argument(
+        "-d",
+        "--data-dir",
+        help="directory where files are/will be stored",
+        default=os.path.join(os.getcwd(), "data"),
+    )
+    parser.add_argument(
+        "-b",
+        "--bind",
+        default="0.0.0.0",
+        help="server bind address (default: 0.0.0.0)",
+    )
+    args = parser.parse_args()
+    return Parameters.from_namespace(args)

src/project.py

@@ -0,0 +1,40 @@
+import os.path
+import toml
+import typing
+
+__project_data = None
+
+
+def __get_project_data() -> None | dict[str, typing.Any]:
+    global __project_data
+    if __project_data is None:
+        pyproject_toml_file = os.path.join(
+            os.path.dirname(__file__), "..", "pyproject.toml"
+        )
+        if os.path.exists(pyproject_toml_file) and os.path.isfile(pyproject_toml_file):
+            try:
+                data = toml.load(pyproject_toml_file)
+                if "project" in data:
+                    __project_data = data["project"]
+            except TypeError, toml.TomlDecodeError, FileNotFoundError:
+                pass
+    return __project_data
+
+
+def __get_str_value(key: str) -> str:
+    project_data = __get_project_data()
+    if project_data is not None and key in project_data:
+        return project_data[key]
+    return "unknown"
+
+
+def get_version() -> str:
+    return __get_str_value("version")
+
+
+def get_name() -> str:
+    return __get_str_value("name")
+
+
+def get_description() -> str:
+    return __get_str_value("description")

src/server.py

@@ -0,0 +1,57 @@
+import http.server
+import http
+
+from . import project, params
+
+
+class _StaplerRequestHandler(http.server.SimpleHTTPRequestHandler):
+    protocol_version = "HTTP/2.0"
+    server_version = "StaplerServer/" + project.get_version()
+
+    def __init__(self, *args, params: params.Parameters, **kwargs):
+        self.default_host = params.host
+        super().__init__(*args, directory=params.data_dir, **kwargs)
+
+    def list_directory(self, *_, **__):
+        """Disable default directory listing"""
+        self.send_error(http.HTTPStatus.NOT_FOUND, "File not found")
+
+    def do_GET(self):
+        if self.path == "/" and self.get_host() == self.default_host:
+            self.server_index()
+            return
+        super().do_GET()
+
+    def get_host(self) -> str:
+        return self.headers["Host"].split(":")[0]
+
+    def server_index(self):
+        self.send_basic_body(self.server_version)
+
+    def send_basic_body(self, body: str, content_type: str = "text/plain"):
+        encoded: bytes = body.encode()
+        self.send_response(http.HTTPStatus.OK)
+        self.send_header("Content-type", f"{content_type}; charset=UTF-8")
+        self.send_header("Content-Length", str(len(encoded)))
+        self.end_headers()
+        self.wfile.write(encoded)
+
+
+class StaplerServer:
+    def __init__(self, params: params.Parameters):
+        self.default_host = params.host
+        self.server = http.server.ThreadingHTTPServer(
+            (params.bind, params.port),
+            lambda req, client, server: _StaplerRequestHandler(
+                req, client, server, params=params
+            ),
+        )
+
+    def start(self):
+        print(
+            f"{_StaplerRequestHandler.server_version} serving on http://{self.default_host}:{self.server.server_port}..."
+        )
+        try:
+            self.server.serve_forever()
+        except KeyboardInterrupt:
+            pass