typing

THE_SERVER.py

@@ -6,8 +6,7 @@ go dam
 import http.server as serverhttp
 import sys
 
-from the_server_handler import \
-    ServerLogic  # , connectionSettings, get_that_connection
+from the_server_handler import ServerLogic  # , connectionSettings, get_that_connection
 
 # import mysql.connector as sqlcon
 
@@ -39,5 +38,3 @@ def runServer():
 
 if __name__ == "__main__":
     runServer()
-
-

the_server_handler.py

@@ -14,7 +14,8 @@ from functools import wraps
 # import chardet
 from http import HTTPStatus as rescode
 from pathlib import Path
-from typing import Literal, Optional
+from collections.abc import Callable, Collection
+from typing import Literal, Optional, TypeAlias
 
 # from mysql.connector import errorcode
 from urllib import parse as urlparse  # lmao the normal person
@@ -104,29 +105,33 @@ def better_connection_actuallyworking(connectionSettings: dict, conn_name="conn"
     return decorator
 
 
+fn: TypeAlias = Literal["one", "many", "all", "warnings"]
+
+
 # i have to use my own tech
 @better_connection_actuallyworking(connectionSettings)
 def query_once_fetch_lda(
     execute_settings: tuple,
     cursor_settings: dict,
-    fetch_many: Literal["one", "many", "all", "warnings"] = "one",
+    fetch_many: fn | Collection[fn] = "one",
     *,
     conn=None,
 ):
     "every thing passes to execute, return fetchone. lda test"
     # print("i seqeralr", conn)
     with conn.cursor(**cursor_settings) as r:
+        rf: dict[fn, Callable] = {
+            "one": r.fetchone,
+            "all": r.fetchall,
+            "many": r.fetchmany,
+            "warnings": r.fetchwarnings,
+        }
         r.execute(*execute_settings)
-        match fetch_many:
-            case "one":
-                return r.fetchone()
-            case "all":
-                return r.fetchall()
-            case "many":
-                return r.fetchmany()
-            case "warnings":
-                return r.fetchwarnings()
-            case _:
+
+        if isinstance(fetch_many, fn):
+            return rf[fetch_many]()
+        if isinstance(fetch_many, Collection[fn]):
+            return tuple(rf[i]() for i in fetch_many)
         raise AttributeError(
             f"please put one of the one, many, all, warnings in ty (you put {fetch_many})",
             name=fetch_many,
@@ -336,6 +341,12 @@ class ServerLogic(serverhttp.BaseHTTPRequestHandler):
         #     pass
         self.end_headers()
 
+    def print_setup(self, code=rescode.OK, headers=None, content=None):
+        "header_setup + wfile.write"
+        self.header_setup(code, headers)
+        if content is not None:
+            self.wfile.write(content)
+
     def do_GET(self):
         self.reallyEveryTime()
         pathObj: urlparse.ParseResult = urlparse.urlparse(self.path)
@@ -358,13 +369,12 @@ class ServerLogic(serverhttp.BaseHTTPRequestHandler):
             case "/close":
                 return self.close()
 
-            case p if p != "/endp":  # self.path
-                self.header_setup(rescode.FORBIDDEN)
-                self.wfile.write(notallowed403)
-                return
-            case _:
+            case "/endp":
                 return post_endp_lda(self)
 
+            case _:  # self.path
+                return self.print_setup(rescode.FORBIDDEN, content=notallowed403)
+
 
 def post_endp_lda(self):
 
@@ -375,7 +385,6 @@ def post_endp_lda(self):
         post_data_bytes = r.read(content_length)
         ## read how many bytes is the content length
 
-    def lmao(self, ldarescode, stupid, ldacause=None):
     efg = (
         f'Ive got a POST at {self.headers.get("Origin", "_nowhere lmao_")} from {self.headers.get("Referer", "_who?_")} with the following data:\n'.encode()
         + (
@@ -385,74 +394,29 @@ def post_endp_lda(self):
         + b"\n"
     )
     headerWhat = {"Content-type": "text/plain;charset=utf-8"}  # tf
-        match stupid:
-            case 1:
-                print("cccc", content_type, content_length)
-
-                self.header_setup(
-                    code=ldarescode,
-                    headers=headerWhat,
-                )
-                self.wfile.write(efg)
-                self.wfile.write(b"its of the type: ")
-                self.wfile.write(content_type.encode() or b"fdahjklfdashkjlasdf")
-            case 2:
-                cc = ldacause
-                print(cc)
-                self.header_setup(
-                    code=rescode.BAD_REQUEST,
-                    headers=headerWhat,
-                )
-                self.wfile.write(efg)
-                self.wfile.write(
-                    b"haha your data sucks (json.loads is not necessarily strict/bad, its just u sent bad): "
-                    + str(cc).encode()
-                )
-            case 3:
-                cc = ldacause
-                # print("execute err: ", cc)
-                q = str(cc)
-                for a, b in zip(severColum, clietColum):
-                    q = q.replace(a, b)
-                # print("execute errd: ", q)
-                self.header_setup(
-                    code=rescode.BAD_REQUEST,
-                    headers=headerWhat,
-                )
-                self.wfile.write(efg)
-                self.wfile.write(
-                    b"is it very much Invalid: "
-                    + q.encode()
-                    + b"\n funny Stuff: # replace strings found in list old to same index in list new. used for, zip, string.replace"
-                )
-            case 4:
-                self.header_setup(
-                    code=rescode.CREATED,
-                    headers=headerWhat,
-                )
-                self.wfile.write(efg)
-            case 5:
-                cc = ldacause
-                print("coneti eror: ", cc)
-                self.header_setup(
-                    code=rescode.INTERNAL_SERVER_ERROR,
-                    headers=headerWhat,
-                )
-                self.wfile.write(efg)
-                self.wfile.write(
-                    b"the thing cant even start like what f: " + str(cc).encode()
-                )
 
     if content_type != "application/json":
-        return lmao(self, rescode.OK, 1)
-        # self.wfile.write(b"its of the type: ")
-        # self.wfile.write(content_type.encode())
+        print("cccc", content_type, content_length)
+        return self.print_setup(
+            code=rescode.OK,
+            headers=headerWhat,
+            content=efg
+            + b"its of the type: "
+            + (content_type.encode() or b"fdahjklfdashkjlasdf"),
+        )
 
     try:
         data_dict: dict = json.loads(post_data_bytes)
     ##
-    except json.JSONDecodeError as e:
-        return lmao(self, rescode.BAD_REQUEST, 2, e)
+    except json.JSONDecodeError as cc:
+        print(cc)
+        return self.print_setup(
+            code=rescode.BAD_REQUEST,
+            headers=headerWhat,
+            content=efg
+            + b"haha your data sucks (json.loads is not necessarily strict/bad, its just u sent bad): "
+            + str(cc).encode(),
+        )
 
     if data_dict.get("from") == "ldlda":
         data_dict = getnewdicttranslate(
@@ -495,12 +459,33 @@ def post_endp_lda(self):
                     )
                     print("res of:", cur.fetchone())
 
+                return self.print_setup(
+                    code=rescode.CREATED, headers=headerWhat, content=efg
+                )
+
             except sqlcon.Error as er:
-                return lmao(self, rescode.BAD_REQUEST, 3, er)
-            return lmao(self, rescode.CREATED, 4)
+                q = str(er)
+                for a, b in zip(severColum, clietColum):
+                    q = q.replace(a, b)
+                # print("execute errd: ", q)
+                return self.print_setup(
+                    code=rescode.BAD_REQUEST,
+                    headers=headerWhat,
+                    content=efg
+                    + b"is it very much Invalid: "
+                    + q.encode()
+                    + b"\n funny Stuff: # replace strings found in list old to same index in list new. used for, zip, string.replace",
+                )
 
     except sqlcon.Error as eror:
-        return lmao(self, rescode.INTERNAL_SERVER_ERROR, 5, eror)
+        print("coneti eror: ", eror)
+        return self.print_setup(
+            code=rescode.INTERNAL_SERVER_ERROR,
+            headers=headerWhat,
+            content=efg
+            + b"the thing cant even start like what f: "
+            + str(eror).encode(),
+        )
 
     # what the fuck is this
 
@@ -521,15 +506,12 @@ def openFileManually(
     path = Path(rootDir, filename)
     print(path)
     if not path.exists():
-        theGuy.header_setup(rescode.NOT_FOUND)
-        theGuy.wfile.write(notfound404)
-        return
+        return theGuy.print_setup(rescode.NOT_FOUND, content=notfound404)
+
     if path.is_dir():
         path /= defaultFile
         if not path.is_file():
-            theGuy.header_setup(rescode.NOT_FOUND)
-            theGuy.wfile.write(notfound404)
-            return
+            return theGuy.print_setup(rescode.NOT_FOUND, content=notfound404)
 
     print(path)
     fileMime, fileEncoding = mimetypes.guess_type(path)
@@ -552,8 +534,7 @@ def openFileManually(
         asfd = file.read()
 
     headers |= {(gay or "content-type"): finalMime, "content-length": str(len(asfd))}
-    theGuy.header_setup(headers=headers)
-    theGuy.wfile.write(asfd)
+    theGuy.print_setup(headers=headers, content=asfd)
 
 
 def openFileSemiAutomatically(
@@ -602,11 +583,10 @@ def openFileSemiAutomatically(
             "content-type": mt,
             # "content-length": str(len(c))
         }
-    theGuy.header_setup(200, headers)
-    theGuy.wfile.write(c)
+    theGuy.print_setup(200, headers, c)
 
 
-def get_from_db(self, pathObj):
+def get_from_db(self: ServerLogic, pathObj: urlparse.ParseResult):
     ee = query_once_fetch_lda(
         execute_settings=(
             f"select neme, kewod, cod from {tabletbale} where neme = ?;",
@@ -617,26 +597,25 @@ def get_from_db(self, pathObj):
     )
 
     if not ee:
-        self.header_setup(rescode.NOT_FOUND)
-        self.wfile.write(notfound404)
-        return
+        return self.print_setup(rescode.NOT_FOUND, content=notfound404)
+
     v, c = ee[1], ee[2]  # value, code
     if v:
         if c == rescode.OK:
-            self.header_setup(c, {"Content-type": "text/plain;charset=utf-8"})
-            self.wfile.write(v.encode())
+            return self.print_setup(
+                c, {"Content-type": "text/plain;charset=utf-8"}, v.encode()
+            )
             ## i mean crashes are handled anyway things keep running as is
             # what bro sayin
-            return
-        elif c // 100 == 2:
+
+        if c // 100 == 3:  # redirects
+            return self.header_setup(c, {"Location": v})
+
+        if c // 100 == 2:
             pass
-        elif c // 100 == 3:  # redirects
-            self.header_setup(c, {"Location": v})
-            return
     # case what????
-    self.header_setup(rescode.OK)
-    self.wfile.write(
-        f'<p>u at <a href=".{pathObj.path}">{pathObj.path}</a>. probably content is None python said so idfk</p>'.encode()
+    self.print_setup(
+        content=f'<p>u at <a href=".{pathObj.path}">{pathObj.path}</a>. probably content is None python said so idfk</p>'.encode(),
     )
     if c == rescode.OK:
         self.wfile.write(b"<p>this shouldve been a text file</p>")
@@ -646,7 +625,7 @@ def get_from_db(self, pathObj):
 
 ## USE THE @ if you know how
 ### i surely dont know how
-def endp_query(self):
+def endp_query(self: ServerLogic):
     ncnw = process_datetime_DICTS(fulltable(tabletbale))
     efg = json.dumps(
         [  # haha yeah right why cant you do it client side
@@ -658,8 +637,9 @@ def endp_query(self):
             for i in ncnw
         ]
     )
-    self.header_setup(headers={"Content-type": "application/json"})
-    self.wfile.write(str(efg).encode())
+    self.print_setup(
+        headers={"Content-type": "application/json"}, content=str(efg).encode()
+    )
 
 
 if __name__ == "__main__":

things.py

@@ -105,6 +105,7 @@ def unpack_with_defaults(t, defaults=(None, (), {})):
 # for i in test_subject:
 #     print( f"{i = },\n{unpack_with_defaults(i) = },\n{maybe_safer_unpack(i) = }\n")
 
+
 # no dont touch this ever
 def function_splitter(
     t: tuple[Union[Callable, str], tuple[Any, ...], Dict[str, Any]] = (