typing
This commit is contained in:
1 parent
0c967a4f9c
commit
508be518e9
3 files changed
+67
-74
No files matched your search
+1
-4
@@ -6,8 +6,7 @@ go dam
|
|||||||
import http.server as serverhttp
|
import http.server as serverhttp
|
||||||
import sys
|
import sys
|
||||||
|
|
||||||
from the_server_handler import \
|
from the_server_handler import ServerLogic # , connectionSettings, get_that_connection
|
||||||
ServerLogic # , connectionSettings, get_that_connection
|
|
||||||
|
|
||||||
# import mysql.connector as sqlcon
|
# import mysql.connector as sqlcon
|
||||||
|
|
||||||
@@ -39,5 +38,3 @@ def runServer():
|
|||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
runServer()
|
runServer()
|
||||||
|
|
||||||
|
|
||||||
+65
-70
@@ -14,7 +14,8 @@ from functools import wraps
|
|||||||
# import chardet
|
# import chardet
|
||||||
from http import HTTPStatus as rescode
|
from http import HTTPStatus as rescode
|
||||||
from pathlib import Path
|
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 mysql.connector import errorcode
|
||||||
from urllib import parse as urlparse # lmao the normal person
|
from urllib import parse as urlparse # lmao the normal person
|
||||||
@@ -104,34 +105,38 @@ def better_connection_actuallyworking(connectionSettings: dict, conn_name="conn"
|
|||||||
return decorator
|
return decorator
|
||||||
|
|
||||||
|
|
||||||
|
fn: TypeAlias = Literal["one", "many", "all", "warnings"]
|
||||||
|
|
||||||
|
|
||||||
# i have to use my own tech
|
# i have to use my own tech
|
||||||
@better_connection_actuallyworking(connectionSettings)
|
@better_connection_actuallyworking(connectionSettings)
|
||||||
def query_once_fetch_lda(
|
def query_once_fetch_lda(
|
||||||
execute_settings: tuple,
|
execute_settings: tuple,
|
||||||
cursor_settings: dict,
|
cursor_settings: dict,
|
||||||
fetch_many: Literal["one", "many", "all", "warnings"] = "one",
|
fetch_many: fn | Collection[fn] = "one",
|
||||||
*,
|
*,
|
||||||
conn=None,
|
conn=None,
|
||||||
):
|
):
|
||||||
"every thing passes to execute, return fetchone. lda test"
|
"every thing passes to execute, return fetchone. lda test"
|
||||||
# print("i seqeralr", conn)
|
# print("i seqeralr", conn)
|
||||||
with conn.cursor(**cursor_settings) as r:
|
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)
|
r.execute(*execute_settings)
|
||||||
match fetch_many:
|
|
||||||
case "one":
|
if isinstance(fetch_many, fn):
|
||||||
return r.fetchone()
|
return rf[fetch_many]()
|
||||||
case "all":
|
if isinstance(fetch_many, Collection[fn]):
|
||||||
return r.fetchall()
|
return tuple(rf[i]() for i in fetch_many)
|
||||||
case "many":
|
raise AttributeError(
|
||||||
return r.fetchmany()
|
f"please put one of the one, many, all, warnings in ty (you put {fetch_many})",
|
||||||
case "warnings":
|
name=fetch_many,
|
||||||
return r.fetchwarnings()
|
obj=r,
|
||||||
case _:
|
)
|
||||||
raise AttributeError(
|
|
||||||
f"please put one of the one, many, all, warnings in ty (you put {fetch_many})",
|
|
||||||
name=fetch_many,
|
|
||||||
obj=r,
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
@better_connection_actuallyworking(connectionSettings)
|
@better_connection_actuallyworking(connectionSettings)
|
||||||
@@ -336,9 +341,11 @@ class ServerLogic(serverhttp.BaseHTTPRequestHandler):
|
|||||||
# pass
|
# pass
|
||||||
self.end_headers()
|
self.end_headers()
|
||||||
|
|
||||||
def print_setup(self, code, headers, content):
|
def print_setup(self, code=rescode.OK, headers=None, content=None):
|
||||||
|
"header_setup + wfile.write"
|
||||||
self.header_setup(code, headers)
|
self.header_setup(code, headers)
|
||||||
self.wfile.write(content)
|
if content is not None:
|
||||||
|
self.wfile.write(content)
|
||||||
|
|
||||||
def do_GET(self):
|
def do_GET(self):
|
||||||
self.reallyEveryTime()
|
self.reallyEveryTime()
|
||||||
@@ -366,9 +373,7 @@ class ServerLogic(serverhttp.BaseHTTPRequestHandler):
|
|||||||
return post_endp_lda(self)
|
return post_endp_lda(self)
|
||||||
|
|
||||||
case _: # self.path
|
case _: # self.path
|
||||||
self.header_setup(rescode.FORBIDDEN)
|
return self.print_setup(rescode.FORBIDDEN, content=notallowed403)
|
||||||
self.wfile.write(notallowed403)
|
|
||||||
return
|
|
||||||
|
|
||||||
|
|
||||||
def post_endp_lda(self):
|
def post_endp_lda(self):
|
||||||
@@ -380,7 +385,6 @@ def post_endp_lda(self):
|
|||||||
post_data_bytes = r.read(content_length)
|
post_data_bytes = r.read(content_length)
|
||||||
## read how many bytes is the content length
|
## read how many bytes is the content length
|
||||||
|
|
||||||
# def lmao(self: ServerLogic, ldarescode, stupid, ldacause=None):
|
|
||||||
efg = (
|
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()
|
f'Ive got a POST at {self.headers.get("Origin", "_nowhere lmao_")} from {self.headers.get("Referer", "_who?_")} with the following data:\n'.encode()
|
||||||
+ (
|
+ (
|
||||||
@@ -391,19 +395,15 @@ def post_endp_lda(self):
|
|||||||
)
|
)
|
||||||
headerWhat = {"Content-type": "text/plain;charset=utf-8"} # tf
|
headerWhat = {"Content-type": "text/plain;charset=utf-8"} # tf
|
||||||
|
|
||||||
|
|
||||||
if content_type != "application/json":
|
if content_type != "application/json":
|
||||||
print("cccc", content_type, content_length)
|
print("cccc", content_type, content_length)
|
||||||
return self.print_setup(
|
return self.print_setup(
|
||||||
code=rescode.OK,
|
code=rescode.OK,
|
||||||
headers=headerWhat,
|
headers=headerWhat,
|
||||||
content=efg
|
content=efg
|
||||||
+ b"its of the type: "
|
+ b"its of the type: "
|
||||||
+ (content_type.encode() or b"fdahjklfdashkjlasdf"),
|
+ (content_type.encode() or b"fdahjklfdashkjlasdf"),
|
||||||
)
|
)
|
||||||
# return lmao(self, rescode.OK, 1)
|
|
||||||
# self.wfile.write(b"its of the type: ")
|
|
||||||
# self.wfile.write(content_type.encode())
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
data_dict: dict = json.loads(post_data_bytes)
|
data_dict: dict = json.loads(post_data_bytes)
|
||||||
@@ -417,7 +417,6 @@ def post_endp_lda(self):
|
|||||||
+ b"haha your data sucks (json.loads is not necessarily strict/bad, its just u sent bad): "
|
+ b"haha your data sucks (json.loads is not necessarily strict/bad, its just u sent bad): "
|
||||||
+ str(cc).encode(),
|
+ str(cc).encode(),
|
||||||
)
|
)
|
||||||
# return lmao(self, rescode.BAD_REQUEST, 2, e)
|
|
||||||
|
|
||||||
if data_dict.get("from") == "ldlda":
|
if data_dict.get("from") == "ldlda":
|
||||||
data_dict = getnewdicttranslate(
|
data_dict = getnewdicttranslate(
|
||||||
@@ -460,6 +459,10 @@ def post_endp_lda(self):
|
|||||||
)
|
)
|
||||||
print("res of:", cur.fetchone())
|
print("res of:", cur.fetchone())
|
||||||
|
|
||||||
|
return self.print_setup(
|
||||||
|
code=rescode.CREATED, headers=headerWhat, content=efg
|
||||||
|
)
|
||||||
|
|
||||||
except sqlcon.Error as er:
|
except sqlcon.Error as er:
|
||||||
q = str(er)
|
q = str(er)
|
||||||
for a, b in zip(severColum, clietColum):
|
for a, b in zip(severColum, clietColum):
|
||||||
@@ -473,19 +476,16 @@ def post_endp_lda(self):
|
|||||||
+ q.encode()
|
+ q.encode()
|
||||||
+ b"\n funny Stuff: # replace strings found in list old to same index in list new. used for, zip, string.replace",
|
+ b"\n funny Stuff: # replace strings found in list old to same index in list new. used for, zip, string.replace",
|
||||||
)
|
)
|
||||||
# return lmao(self, rescode.BAD_REQUEST, 3, er)
|
|
||||||
return self.print_setup(code=rescode.CREATED, headers=headerWhat, content=efg)
|
|
||||||
|
|
||||||
except sqlcon.Error as eror:
|
except sqlcon.Error as eror:
|
||||||
print("coneti eror: ", eror)
|
print("coneti eror: ", eror)
|
||||||
return self.print_setup(
|
return self.print_setup(
|
||||||
code=rescode.INTERNAL_SERVER_ERROR,
|
code=rescode.INTERNAL_SERVER_ERROR,
|
||||||
headers=headerWhat,
|
headers=headerWhat,
|
||||||
content=efg
|
content=efg
|
||||||
+ b"the thing cant even start like what f: "
|
+ b"the thing cant even start like what f: "
|
||||||
+ str(eror).encode(),
|
+ str(eror).encode(),
|
||||||
)
|
)
|
||||||
# return lmao(self, rescode.INTERNAL_SERVER_ERROR, 5, eror)
|
|
||||||
|
|
||||||
# what the fuck is this
|
# what the fuck is this
|
||||||
|
|
||||||
@@ -506,15 +506,12 @@ def openFileManually(
|
|||||||
path = Path(rootDir, filename)
|
path = Path(rootDir, filename)
|
||||||
print(path)
|
print(path)
|
||||||
if not path.exists():
|
if not path.exists():
|
||||||
theGuy.header_setup(rescode.NOT_FOUND)
|
return theGuy.print_setup(rescode.NOT_FOUND, content=notfound404)
|
||||||
theGuy.wfile.write(notfound404)
|
|
||||||
return
|
|
||||||
if path.is_dir():
|
if path.is_dir():
|
||||||
path /= defaultFile
|
path /= defaultFile
|
||||||
if not path.is_file():
|
if not path.is_file():
|
||||||
theGuy.header_setup(rescode.NOT_FOUND)
|
return theGuy.print_setup(rescode.NOT_FOUND, content=notfound404)
|
||||||
theGuy.wfile.write(notfound404)
|
|
||||||
return
|
|
||||||
|
|
||||||
print(path)
|
print(path)
|
||||||
fileMime, fileEncoding = mimetypes.guess_type(path)
|
fileMime, fileEncoding = mimetypes.guess_type(path)
|
||||||
@@ -537,8 +534,7 @@ def openFileManually(
|
|||||||
asfd = file.read()
|
asfd = file.read()
|
||||||
|
|
||||||
headers |= {(gay or "content-type"): finalMime, "content-length": str(len(asfd))}
|
headers |= {(gay or "content-type"): finalMime, "content-length": str(len(asfd))}
|
||||||
theGuy.header_setup(headers=headers)
|
theGuy.print_setup(headers=headers, content=asfd)
|
||||||
theGuy.wfile.write(asfd)
|
|
||||||
|
|
||||||
|
|
||||||
def openFileSemiAutomatically(
|
def openFileSemiAutomatically(
|
||||||
@@ -587,11 +583,10 @@ def openFileSemiAutomatically(
|
|||||||
"content-type": mt,
|
"content-type": mt,
|
||||||
# "content-length": str(len(c))
|
# "content-length": str(len(c))
|
||||||
}
|
}
|
||||||
theGuy.header_setup(200, headers)
|
theGuy.print_setup(200, headers, c)
|
||||||
theGuy.wfile.write(c)
|
|
||||||
|
|
||||||
|
|
||||||
def get_from_db(self, pathObj):
|
def get_from_db(self: ServerLogic, pathObj: urlparse.ParseResult):
|
||||||
ee = query_once_fetch_lda(
|
ee = query_once_fetch_lda(
|
||||||
execute_settings=(
|
execute_settings=(
|
||||||
f"select neme, kewod, cod from {tabletbale} where neme = ?;",
|
f"select neme, kewod, cod from {tabletbale} where neme = ?;",
|
||||||
@@ -602,26 +597,25 @@ def get_from_db(self, pathObj):
|
|||||||
)
|
)
|
||||||
|
|
||||||
if not ee:
|
if not ee:
|
||||||
self.header_setup(rescode.NOT_FOUND)
|
return self.print_setup(rescode.NOT_FOUND, content=notfound404)
|
||||||
self.wfile.write(notfound404)
|
|
||||||
return
|
|
||||||
v, c = ee[1], ee[2] # value, code
|
v, c = ee[1], ee[2] # value, code
|
||||||
if v:
|
if v:
|
||||||
if c == rescode.OK:
|
if c == rescode.OK:
|
||||||
self.header_setup(c, {"Content-type": "text/plain;charset=utf-8"})
|
return self.print_setup(
|
||||||
self.wfile.write(v.encode())
|
c, {"Content-type": "text/plain;charset=utf-8"}, v.encode()
|
||||||
|
)
|
||||||
## i mean crashes are handled anyway things keep running as is
|
## i mean crashes are handled anyway things keep running as is
|
||||||
# what bro sayin
|
# 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
|
pass
|
||||||
elif c // 100 == 3: # redirects
|
|
||||||
self.header_setup(c, {"Location": v})
|
|
||||||
return
|
|
||||||
# case what????
|
# case what????
|
||||||
self.header_setup(rescode.OK)
|
self.print_setup(
|
||||||
self.wfile.write(
|
content=f'<p>u at <a href=".{pathObj.path}">{pathObj.path}</a>. probably content is None python said so idfk</p>'.encode(),
|
||||||
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:
|
if c == rescode.OK:
|
||||||
self.wfile.write(b"<p>this shouldve been a text file</p>")
|
self.wfile.write(b"<p>this shouldve been a text file</p>")
|
||||||
@@ -631,7 +625,7 @@ def get_from_db(self, pathObj):
|
|||||||
|
|
||||||
## USE THE @ if you know how
|
## USE THE @ if you know how
|
||||||
### i surely dont know how
|
### i surely dont know how
|
||||||
def endp_query(self):
|
def endp_query(self: ServerLogic):
|
||||||
ncnw = process_datetime_DICTS(fulltable(tabletbale))
|
ncnw = process_datetime_DICTS(fulltable(tabletbale))
|
||||||
efg = json.dumps(
|
efg = json.dumps(
|
||||||
[ # haha yeah right why cant you do it client side
|
[ # haha yeah right why cant you do it client side
|
||||||
@@ -643,8 +637,9 @@ def endp_query(self):
|
|||||||
for i in ncnw
|
for i in ncnw
|
||||||
]
|
]
|
||||||
)
|
)
|
||||||
self.header_setup(headers={"Content-type": "application/json"})
|
self.print_setup(
|
||||||
self.wfile.write(str(efg).encode())
|
headers={"Content-type": "application/json"}, content=str(efg).encode()
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
|
|||||||
@@ -105,6 +105,7 @@ def unpack_with_defaults(t, defaults=(None, (), {})):
|
|||||||
# for i in test_subject:
|
# for i in test_subject:
|
||||||
# print( f"{i = },\n{unpack_with_defaults(i) = },\n{maybe_safer_unpack(i) = }\n")
|
# print( f"{i = },\n{unpack_with_defaults(i) = },\n{maybe_safer_unpack(i) = }\n")
|
||||||
|
|
||||||
|
|
||||||
# no dont touch this ever
|
# no dont touch this ever
|
||||||
def function_splitter(
|
def function_splitter(
|
||||||
t: tuple[Union[Callable, str], tuple[Any, ...], Dict[str, Any]] = (
|
t: tuple[Union[Callable, str], tuple[Any, ...], Dict[str, Any]] = (
|
||||||
|
|||||||
Reference in new issue
Block a user