huge changes again!!! my bad habits!!!!!!
This commit is contained in:
lda committed
1 parent
517707b79c
commit
e2b1112e0f
3 files changed
+276
-460
No files matched your search
+1
-70
@@ -11,6 +11,7 @@ from the_server_handler import \
|
||||
|
||||
# import mysql.connector as sqlcon
|
||||
|
||||
# pylint: disable=invalid-name, line-too-long, missing-function-docstring, missing-class-docstring, missing-module-docstring
|
||||
|
||||
if not sys.version_info >= (3, 11):
|
||||
raise Exception(f"python 3.11 up, u running {sys.version}")
|
||||
@@ -40,73 +41,3 @@ if __name__ == "__main__":
|
||||
runServer()
|
||||
|
||||
|
||||
# import sql
|
||||
|
||||
# import datetime
|
||||
# import os
|
||||
# from dotenv import dotenv_values
|
||||
|
||||
# from typing import Callable, Dict, Any
|
||||
|
||||
# from things import (
|
||||
# funny_trycatch_wrapper,
|
||||
# process_tuples,
|
||||
# funny_tuples_to_dicts_processor,
|
||||
# ) # no way its the funny trycatch wrapper
|
||||
|
||||
# dotenv_values1 = dotenv_values()
|
||||
# db_name = dotenv_values1["db_name"]
|
||||
|
||||
# he really did that
|
||||
# from mysql.connector import errorcode
|
||||
|
||||
# comectomKeys = ["user", "password", "host", "database"]
|
||||
# comectomtrk = {
|
||||
# "db_user": "user",
|
||||
# "db_password": "password",
|
||||
# "db_host": "host",
|
||||
# "db_name": "database",
|
||||
# }
|
||||
|
||||
# def getnewdict(d:dict, l:list) -> dict:
|
||||
# r = dict()
|
||||
# for k in d:
|
||||
# if k in l:
|
||||
# r[k] = d[k]
|
||||
# return r
|
||||
# getnewdict = lambda d, l: {k: d[k] for k in d if k in l}
|
||||
|
||||
# def getnewdicttranslate(d:dict, tltb:dict) -> dict:
|
||||
# r = dict()
|
||||
# for k in d:
|
||||
# if k in tltb:
|
||||
# r[tltb[k]] = d[k]
|
||||
# return r
|
||||
# getnewdicttranslate = lambda d, tltb: {tltb[k]: d[k] for k in d if k in tltb}
|
||||
|
||||
|
||||
# connectionSettings = {
|
||||
# **getnewdicttranslate(dotenv_values1, comectomtrk),
|
||||
# "raise_on_warnings": True,
|
||||
# }
|
||||
|
||||
|
||||
# quer = "insert into tbale (kewod, id) values (%s, %s);" # do not support custom key Only if
|
||||
# vals = (
|
||||
# r"faf",
|
||||
# r"24",
|
||||
# )
|
||||
# tablesql = sql.Table(name="tbale", database=db_name)
|
||||
|
||||
# tablesql.select()
|
||||
|
||||
|
||||
# def get_that_connection(
|
||||
# connectionSettings: dict,
|
||||
# ) -> tuple: # the giving up is real
|
||||
# try:
|
||||
# connectio = sqlcon.connect(**connectionSettings)
|
||||
# except sqlcon.Error as e:
|
||||
# return (None, e)
|
||||
# else:
|
||||
# return (connectio, None)
|
||||
+270
-386
@@ -22,12 +22,16 @@ from urllib import parse as urlparse # lmao the normal person
|
||||
import mysql.connector as sqlcon
|
||||
from dotenv import dotenv_values
|
||||
|
||||
from things import getnewdicttranslate, process_datetime_DICTS
|
||||
|
||||
# process_datetime_tuples,; funny_tuples_to_dicts_processor,
|
||||
# from dotenv import dotenv_values
|
||||
|
||||
|
||||
# import urllib, sql
|
||||
class ServerLogic(serverhttp.BaseHTTPRequestHandler): ...
|
||||
# class ServerLogic(serverhttp.BaseHTTPRequestHandler): ...
|
||||
|
||||
# pylint: disable=invalid-name, line-too-long, missing-function-docstring, missing-class-docstring, missing-module-docstring
|
||||
|
||||
webrootdir = "webRoot"
|
||||
curdir = "/home/opc/Desktop/pythonprac"
|
||||
@@ -44,10 +48,6 @@ if os.path.isdir(webrootdir): # unfuckable
|
||||
os.chdir(webrootdir) # move to safe part of world
|
||||
curdir = os.path.curdir or os.getcwd()
|
||||
print(os.path.curdir, os.getcwd())
|
||||
from things import ( # process_datetime_tuples,; funny_tuples_to_dicts_processor,
|
||||
getnewdicttranslate,
|
||||
process_datetime_DICTS,
|
||||
)
|
||||
|
||||
dotenv_values1 = dotenv_values() ## change so fast
|
||||
db_name = dotenv_values1["db_name"]
|
||||
@@ -113,24 +113,7 @@ def funnying_funnytheFunnier(
|
||||
return conn # WHAT NOT TO DO
|
||||
|
||||
|
||||
# if __name__ == "__main__":
|
||||
# conn = funnying_funnytheFunnier()
|
||||
# print(f"Connection object: {conn}")
|
||||
# print(f"Internal connection (_cnx): {getattr(conn, '_cnx', 'Not available')}")
|
||||
# try:
|
||||
# with conn.cursor() as cur:
|
||||
# cur.execute('SELECT "Hello world"')
|
||||
# print(cur.fetchall())
|
||||
# except Exception as e:
|
||||
# traceback.print_exc()
|
||||
# if __name__ == "__main__":
|
||||
# print("a separator")
|
||||
# with sqlcon.connect( pool_name = "ig", **connectionSettings) as conn:
|
||||
# print(f"Connection object: {conn}")
|
||||
# print(f"Internal connection (_cnx): {getattr(conn, '_cnx', 'Not available')}")
|
||||
# with conn.cursor() as cur:
|
||||
# cur.execute('SELECT "Hello world!~"') #gayà
|
||||
# print(cur.fetchall())
|
||||
|
||||
|
||||
|
||||
@better_connection_actuallyworking(connectionSettings, conn_name="_conn")
|
||||
@@ -261,6 +244,236 @@ notfound404 = b"""<h1>404</h1><p>where u at</p>"""
|
||||
notallowed403 = b"""<h1>403</h1><p>tf u doin</p>"""
|
||||
|
||||
|
||||
class ServerLogic(serverhttp.BaseHTTPRequestHandler):
|
||||
|
||||
# def __init__(self, *a, **kw): # AHAHHAHAHAHAHAHAHA
|
||||
# super().__init__(*a, **kw)
|
||||
access_table = {
|
||||
"/": {
|
||||
# {"dir" : "index.html",}
|
||||
},
|
||||
"/index/style.css": {
|
||||
# "headers": {"Content-Type": "text/css"}
|
||||
},
|
||||
"/index/script.js": {
|
||||
"headers": {"Content-Type": "text/javascript"} # bout Stupid
|
||||
},
|
||||
}
|
||||
|
||||
def reallyEveryTime(self) -> tuple:
|
||||
print("all headers:\n", self.headers)
|
||||
print("the path is : ", self.path)
|
||||
print(
|
||||
"connected at: ",
|
||||
self.headers.get("X-Forwarded-For", "directly").split(", ")[0],
|
||||
)
|
||||
print(urlparse.urlparse(self.path))
|
||||
# openFileSemiAutomatically(self, self.path)
|
||||
|
||||
# table = 'tbale'
|
||||
# print("there is no" ,table,ifTableInDB(db_name, table))
|
||||
# tableli = getTableNmaes(db_name)
|
||||
# print(getColumename(tableli[0])) # i know damn well
|
||||
# print(fulltable(tableli[0]))
|
||||
return
|
||||
|
||||
def close(self):
|
||||
try:
|
||||
self.server.server_close()
|
||||
except BaseException as e:
|
||||
print(e)
|
||||
finally:
|
||||
print("gato")
|
||||
raise SystemExit("truly the /close")
|
||||
|
||||
def header_setup( # please not be bad
|
||||
self,
|
||||
code: int | rescode = rescode.OK,
|
||||
headers: Optional[dict[str, str]] = None,
|
||||
):
|
||||
"""
|
||||
code as int and headers as keyvalue pairs (of strings btw)
|
||||
"""
|
||||
headers = headers or {"Content-type": "text/html"}
|
||||
self.send_response(code)
|
||||
for key, value in headers.items():
|
||||
self.send_header(key, value) # why!
|
||||
# else:
|
||||
# pass
|
||||
self.end_headers()
|
||||
|
||||
def do_GET(self):
|
||||
self.reallyEveryTime()
|
||||
pathObj: urlparse.ParseResult = urlparse.urlparse(self.path)
|
||||
match pathObj.path:
|
||||
case "/close":
|
||||
return self.close()
|
||||
case a if a in (self.access_table):
|
||||
return openFileManually(
|
||||
self, a, **self.access_table[a]
|
||||
) # i can waste objects!!!! simple or empty dicts have same id
|
||||
case "/endp":
|
||||
return endp_query(self)
|
||||
case _:
|
||||
return get_from_db(self, pathObj)
|
||||
|
||||
def do_POST(self):
|
||||
|
||||
self.reallyEveryTime()
|
||||
match self.path:
|
||||
case "/close":
|
||||
return self.close()
|
||||
|
||||
case p if p != "/endp": # self.path
|
||||
self.header_setup(rescode.FORBIDDEN)
|
||||
self.wfile.write(notallowed403)
|
||||
return
|
||||
case _:
|
||||
return post_endp_lda(self)
|
||||
|
||||
|
||||
def post_endp_lda(self):
|
||||
|
||||
content_type = self.headers.get("Content-Type", str())
|
||||
content_length = int(self.headers.get("Content-Length", int()))
|
||||
print("cccc", content_type, content_length)
|
||||
with self.rfile as r:
|
||||
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()
|
||||
+ (
|
||||
post_data_bytes
|
||||
or b"nothing??? what? (no fr nothing fr ALSO YOU KNOW YOUR OWN POST????????????)"
|
||||
) # why so hectic
|
||||
+ 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())
|
||||
|
||||
try:
|
||||
data_dict: dict = json.loads(post_data_bytes)
|
||||
##
|
||||
except json.JSONDecodeError as e:
|
||||
return lmao(self, rescode.BAD_REQUEST, 2, e)
|
||||
|
||||
if data_dict.get("from") == "ldlda":
|
||||
data_dict = getnewdicttranslate(
|
||||
# ogdata_dict[
|
||||
data_dict["data"], ## this would actually change ogdatadict.data ## OR NOT
|
||||
csTrtb,
|
||||
)
|
||||
|
||||
wwwww = tuple(key for key in data_dict if key in colnmaes)
|
||||
# hmmm
|
||||
# repated cod? huh?
|
||||
|
||||
updated_Verybreakablequery_insert_into = (
|
||||
f"insert into {tabletbale} "
|
||||
+ "({0}) values ({0})".format( ## LEARN PYTHON SQL this is nasty
|
||||
",".join(("{}",) * len(wwwww)) # incredibly real update
|
||||
).format(*wwwww, *("?",) * len(wwwww))
|
||||
)
|
||||
print(
|
||||
wwwww,
|
||||
tuple(data_dict[w] for w in wwwww),
|
||||
updated_Verybreakablequery_insert_into,
|
||||
sep="\n",
|
||||
)
|
||||
##
|
||||
try:
|
||||
with sqlcon.connect(**connectionSettings) as conn:
|
||||
# a = connectio.autocommit # archiver
|
||||
conn.autocommit = True # where is my ASDFCKjugsadkfjywgefdklajegdlagf
|
||||
try:
|
||||
with conn.cursor(prepared=True) as cur:
|
||||
conn.autocommit = True # WHERE!!!! IS COMMIT!!!!!!!
|
||||
cur.execute(
|
||||
updated_Verybreakablequery_insert_into,
|
||||
tuple(data_dict[w] for w in wwwww),
|
||||
# multi=True,
|
||||
)
|
||||
conn.commit() # WHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWH
|
||||
cur.execute(
|
||||
"select * from tbale where neme = ?",
|
||||
(data_dict["neme"],),
|
||||
)
|
||||
print("res of:", cur.fetchone())
|
||||
|
||||
except sqlcon.Error as er:
|
||||
return lmao(self, rescode.BAD_REQUEST, 3, er)
|
||||
return lmao(self, rescode.CREATED, 4)
|
||||
|
||||
except sqlcon.Error as eror:
|
||||
return lmao(self, rescode.INTERNAL_SERVER_ERROR, 5, eror)
|
||||
|
||||
# what the fuck is this
|
||||
|
||||
|
||||
|
||||
def openFileManually(
|
||||
theGuy: ServerLogic,
|
||||
webDir: str,
|
||||
@@ -362,136 +575,36 @@ def openFileSemiAutomatically(
|
||||
theGuy.wfile.write(c)
|
||||
|
||||
|
||||
## USE THE @ if you know how
|
||||
### i surely dont know how
|
||||
# i have to use my own tech
|
||||
@better_connection_actuallyworking(connectionSettings)
|
||||
def fetchonce_lda(
|
||||
execute_settings: tuple,
|
||||
cursor_settings: dict,
|
||||
fetch_many: Literal["one", "many", "all", "warnings"] = "once",
|
||||
*,
|
||||
conn=None,
|
||||
):
|
||||
"every thing passes to execute, return fetchone. lda test"
|
||||
with conn.cursor(**cursor_settings) as r:
|
||||
r.execute(*execute_settings)
|
||||
return getattr(r, "fetch" + fetch_many)()
|
||||
|
||||
|
||||
class ServerLogic(serverhttp.BaseHTTPRequestHandler):
|
||||
|
||||
# def __init__(self, *a, **kw): # AHAHHAHAHAHAHAHAHA
|
||||
# super().__init__(*a, **kw)
|
||||
|
||||
def reallyEveryTime(self) -> tuple:
|
||||
print("all headers:\n", self.headers)
|
||||
print("the path is : ", self.path)
|
||||
print(
|
||||
"connected at: ",
|
||||
self.headers.get("X-Forwarded-For", "directly").split(", ")[0],
|
||||
)
|
||||
print(urlparse.urlparse(self.path))
|
||||
# openFileSemiAutomatically(self, self.path)
|
||||
|
||||
# table = 'tbale'
|
||||
# print("there is no" ,table,ifTableInDB(db_name, table))
|
||||
# tableli = getTableNmaes(db_name)
|
||||
# print(getColumename(tableli[0])) # i know damn well
|
||||
# print(fulltable(tableli[0]))
|
||||
return
|
||||
|
||||
def close(self):
|
||||
try:
|
||||
self.server.server_close()
|
||||
except BaseException as e:
|
||||
print(e)
|
||||
finally:
|
||||
print("gato")
|
||||
raise SystemExit("truly the /close")
|
||||
|
||||
def header_setup( # please not be bad
|
||||
self,
|
||||
code: int | rescode = rescode.OK,
|
||||
headers: dict[str, str] = {"Content-type": "text/html"},
|
||||
):
|
||||
"""
|
||||
code as int and headers as keyvalue pairs (of strings btw)
|
||||
"""
|
||||
|
||||
self.send_response(code)
|
||||
for key, value in headers.items():
|
||||
self.send_header(key, value) # why!
|
||||
# else:
|
||||
# pass
|
||||
self.end_headers()
|
||||
|
||||
def do_GET(self):
|
||||
self.reallyEveryTime()
|
||||
pathObj: urlparse.ParseResult = urlparse.urlparse(self.path)
|
||||
match pathObj.path:
|
||||
case "/close":
|
||||
return self.close()
|
||||
case a if a in (
|
||||
thisATable := {
|
||||
"/": {
|
||||
# {"dir" : "index.html",}
|
||||
},
|
||||
"/index/style.css": {
|
||||
# "headers": {"Content-Type": "text/css"}
|
||||
},
|
||||
"/index/script.js": {
|
||||
"headers": {"Content-Type": "text/javascript"} # bout Stupid
|
||||
},
|
||||
}
|
||||
):
|
||||
return openFileManually(
|
||||
self, a, **thisATable[a]
|
||||
) # i can waste objects!!!! simple or empty dicts have same id
|
||||
|
||||
# (connectio, eror) = get_that_connection(connectionSettings)
|
||||
# if eror:
|
||||
# print("coneti eror: ", eror)
|
||||
# return
|
||||
# # HE. MUST. AVOID. THE. TRYCATCH
|
||||
# with connectio.cursor(buffered=True) as cur:
|
||||
# try:
|
||||
# res = cur.execute("select * from tbale")
|
||||
# w = cur.fetchall()
|
||||
# res = cur.execute(
|
||||
# "select column_name from INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME = 'tbale'"
|
||||
# )
|
||||
# anotherone2 = cur.fetchall()
|
||||
# except Exception as e:
|
||||
# print(e)
|
||||
# return # return ealy no else
|
||||
|
||||
case "/endp":
|
||||
ncnw = process_datetime_DICTS(fulltable(tabletbale))
|
||||
efg = json.dumps(
|
||||
[ # haha yeah right why cant you do it client side
|
||||
## BIGGER QUESTION is that this is 2 of the same data
|
||||
getnewdicttranslate(
|
||||
i,
|
||||
scTrtb,
|
||||
)
|
||||
for i in ncnw
|
||||
]
|
||||
)
|
||||
self.header_setup(headers={"Content-type": "application/json"})
|
||||
# self.wfile.write(str(c).encode())
|
||||
# self.wfile.write(str(anotherone2).encode())
|
||||
self.wfile.write(str(efg).encode())
|
||||
return
|
||||
# wd = [(a["neme"], a["kewod"], a["cod"]) for a in ncnw]
|
||||
# print(wd)
|
||||
# npnd = {i["neme"]: i for i in ncnw} # simple as
|
||||
# # for p in npnd:
|
||||
# a = npnd.get(self.path.lstrip("/"))
|
||||
with sqlcon.connect(**connectionSettings) as n:
|
||||
with n.cursor(prepared=True) as r:
|
||||
r.execute(
|
||||
def get_from_db(self, pathObj):
|
||||
ee = fetchonce_lda(
|
||||
execute_settings=(
|
||||
f"select neme, kewod, cod from {tabletbale} where neme = ?;",
|
||||
(self.path.lstrip("/"),),
|
||||
),
|
||||
cursor_settings={"prepared": True},
|
||||
fetch_many="once",
|
||||
)
|
||||
ee = r.fetchone()
|
||||
|
||||
if not ee:
|
||||
self.header_setup(rescode.NOT_FOUND)
|
||||
self.wfile.write(notfound404)
|
||||
return
|
||||
v, c = ee[1], ee[2] # value, code
|
||||
# print(a,v)
|
||||
# print(self.path, a)
|
||||
# if self.path.lstrip("/") == p["neme"]:
|
||||
# a, v, c
|
||||
# print(v, c)
|
||||
if v:
|
||||
if c == rescode.OK:
|
||||
self.header_setup(c, {"Content-type": "text/plain;charset=utf-8"})
|
||||
@@ -513,253 +626,24 @@ class ServerLogic(serverhttp.BaseHTTPRequestHandler):
|
||||
self.wfile.write(b"<p>this shouldve been a text file</p>")
|
||||
elif c // 100 == 3:
|
||||
self.wfile.write(b"<p>this shouldve been a redirect</p>")
|
||||
return
|
||||
|
||||
def do_POST(self):
|
||||
class lmao(Exception):
|
||||
def __init__(
|
||||
self, rc: int | rescode | None = None, case: int | None = None, *a, **kw
|
||||
):
|
||||
"""
|
||||
rc: rescode
|
||||
case: special thingy. truly the hax
|
||||
msg: msg!!!!
|
||||
"""
|
||||
super().__init__(rc, case, *a, **kw)
|
||||
self.stupid = case
|
||||
if isinstance(rc, int):
|
||||
print("what")
|
||||
self.ldarescode = rescode(rc) # why do i have to do this
|
||||
print(self.ldarescode)
|
||||
elif isinstance(rc, rescode):
|
||||
print("good")
|
||||
self.ldarescode = rc
|
||||
print(self.ldarescode)
|
||||
|
||||
@property
|
||||
def ldacause(self):
|
||||
return self.__cause__
|
||||
|
||||
self.reallyEveryTime()
|
||||
match self.path:
|
||||
case "/close":
|
||||
return self.close()
|
||||
|
||||
case p if p != "/endp": # self.path
|
||||
self.header_setup(rescode.FORBIDDEN)
|
||||
self.wfile.write(notallowed403)
|
||||
return
|
||||
try:
|
||||
content_type = self.headers.get("Content-Type", str())
|
||||
content_length = int(self.headers.get("Content-Length"), int())
|
||||
print("cccc", content_type, content_length)
|
||||
with self.rfile as r:
|
||||
post_data_bytes = r.read(content_length)
|
||||
## read how many bytes is the content length
|
||||
|
||||
if content_type != "application/json":
|
||||
raise lmao(rescode.OK, 1)
|
||||
# self.wfile.write(b"its of the type: ")
|
||||
# self.wfile.write(content_type.encode())
|
||||
|
||||
try:
|
||||
data_dict: dict = json.loads(post_data_bytes)
|
||||
##
|
||||
except json.JSONDecodeError as e:
|
||||
raise lmao(rescode.BAD_REQUEST, 2) from e
|
||||
# print(e)
|
||||
# self.header_setup(
|
||||
# code=rescode.BAD_REQUEST,
|
||||
# headers={"Content-type": "text/plain"},
|
||||
# )
|
||||
# self.wfile.write(
|
||||
# b"haha your data sucks (json.loads is not necessarily strict/bad, its just u sent bad): "
|
||||
# + str(e).encode()
|
||||
# )
|
||||
|
||||
##
|
||||
# print((*data_dict.keys(), *data_dict.values()))
|
||||
##
|
||||
if data_dict.get("from") == "ldlda":
|
||||
# ogdata_dict = data_dict.copy() ## as this isnt deep copy also because the thing is short lived i guess i dont ever need this for anything
|
||||
# print("b4: ", data_dict, ogdata_dict) ## initial state for comp
|
||||
data_dict = getnewdicttranslate(
|
||||
# ogdata_dict[
|
||||
data_dict[
|
||||
"data"
|
||||
], ## this would actually change ogdatadict.data ## OR NOT
|
||||
csTrtb,
|
||||
## USE THE @ if you know how
|
||||
### i surely dont know how
|
||||
def endp_query(self):
|
||||
ncnw = process_datetime_DICTS(fulltable(tabletbale))
|
||||
efg = json.dumps(
|
||||
[ # haha yeah right why cant you do it client side
|
||||
## BIGGER QUESTION is that this is 2 of the same data
|
||||
getnewdicttranslate(
|
||||
i,
|
||||
scTrtb,
|
||||
)
|
||||
# print(
|
||||
# "aftr: ", data_dict, ogdata_dict
|
||||
# ) ## notice that the thing changed.
|
||||
##
|
||||
|
||||
wwwww = tuple(key for key in data_dict if key in colnmaes)
|
||||
# hmmm
|
||||
# repated cod? huh?
|
||||
|
||||
updated_Verybreakablequery_insert_into = (
|
||||
f"insert into {tabletbale} "
|
||||
+ "({0}) values ({0})".format( ## LEARN PYTHON SQL this is nasty
|
||||
",".join(("{}",) * len(wwwww)) # incredibly real update
|
||||
).format(*wwwww, *("?",) * len(wwwww))
|
||||
for i in ncnw
|
||||
]
|
||||
)
|
||||
print(
|
||||
wwwww,
|
||||
tuple(data_dict[w] for w in wwwww),
|
||||
updated_Verybreakablequery_insert_into,
|
||||
sep="\n",
|
||||
)
|
||||
##
|
||||
try:
|
||||
with sqlcon.connect(**connectionSettings) as conn:
|
||||
# a = connectio.autocommit # archiver
|
||||
conn.autocommit = (
|
||||
True # where is my ASDFCKjugsadkfjywgefdklajegdlagf
|
||||
)
|
||||
try:
|
||||
with conn.cursor(prepared=True) as cur:
|
||||
conn.autocommit = True # WHERE!!!! IS COMMIT!!!!!!!
|
||||
cur.execute(
|
||||
updated_Verybreakablequery_insert_into,
|
||||
tuple(data_dict[w] for w in wwwww),
|
||||
# multi=True,
|
||||
)
|
||||
conn.commit() # WHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWH
|
||||
cur.execute(
|
||||
"select * from tbale where neme = ?",
|
||||
(data_dict["neme"],),
|
||||
)
|
||||
print("res of:", cur.fetchone())
|
||||
# e = cur.fetchall()
|
||||
# print("insert into qure: ")
|
||||
# e = str(e)
|
||||
# map(lambda a, b: e.replace(a, b), severColum, clietColum)
|
||||
# with self.wfile as wf:
|
||||
# wf.write(b"uh what you mightve submitted SuperSsdajfhgaswjkdfaw "+e.encode())
|
||||
# another = cur.execute("select * from tbale")
|
||||
# w2f = cur.fetchall()
|
||||
##
|
||||
except sqlcon.Error as er:
|
||||
raise lmao(rescode.BAD_REQUEST, 3) from er
|
||||
# print("execute err: ", er)
|
||||
# q = str(er)
|
||||
# map(
|
||||
# lambda a, b: q.replace(a, b),
|
||||
# severColum,
|
||||
# clietColum,
|
||||
# )
|
||||
|
||||
# self.header_setup(
|
||||
# code=rescode.BAD_REQUEST,
|
||||
# headers={"Content-type": "text/plain"},
|
||||
# )
|
||||
# self.wfile.write(
|
||||
# b"is it very much Invalid: "
|
||||
# + q.encode()
|
||||
# + b"\n funny Stuff: q=str(error);map(lambda a, b: q.replace(a, b), severColum, clietColum)"
|
||||
# )
|
||||
|
||||
##
|
||||
else:
|
||||
raise lmao(rescode.CREATED, 4)
|
||||
# self.header_setup(
|
||||
# code=rescode.CREATED,
|
||||
# headers={"Content-type": "text/plain"},
|
||||
# )
|
||||
|
||||
# print(cur)
|
||||
# cur.execute("select * from tbale")
|
||||
# aq142 = cur.fetchall()
|
||||
# self.do_GET()
|
||||
# pass
|
||||
##
|
||||
# finally:
|
||||
# connectio.autocommit = a
|
||||
except sqlcon.Error as eror:
|
||||
raise lmao(rescode.INTERNAL_SERVER_ERROR, 5) from eror
|
||||
# print("coneti eror: ", eror)
|
||||
# self.header_setup(
|
||||
# code=rescode.INTERNAL_SERVER_ERROR,
|
||||
# headers={"Content-type": "text/plain"},
|
||||
# )
|
||||
# self.wfile.write(
|
||||
# b"the thing cant even start like what f: " + str(eror).encode()
|
||||
# )
|
||||
except lmao as e:
|
||||
print(e.args)
|
||||
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()
|
||||
+ (
|
||||
post_data_bytes
|
||||
or b"nothing??? what? (no fr nothing fr ALSO YOU KNOW YOUR OWN POST????????????)"
|
||||
) # why so hectic
|
||||
+ b"\n"
|
||||
)
|
||||
headerWhat = {"Content-type": "text/plain;charset=utf-8"} # tf
|
||||
match e.stupid:
|
||||
case 1:
|
||||
print("cccc", content_type, content_length)
|
||||
|
||||
self.header_setup(
|
||||
code=e.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 = e.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 = e.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 = e.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()
|
||||
)
|
||||
# self.wfile.write(b"\n We have this in the this: ")
|
||||
# self.wfile.write(str(process_datetime_tuples(w2f)).encode())
|
||||
|
||||
## what is a get to endp
|
||||
# elif content-type != json
|
||||
self.header_setup(headers={"Content-type": "application/json"})
|
||||
self.wfile.write(str(efg).encode())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -9,6 +9,7 @@ from typing import Any, Callable, Dict, List, Literal, Tuple, Union
|
||||
|
||||
# import os, sys
|
||||
|
||||
# pylint: disable=invalid-name, line-too-long, missing-function-docstring, missing-class-docstring, missing-module-docstring
|
||||
# class FunnyStuff:
|
||||
# @staticmethod
|
||||
# def funny(
|
||||
@@ -279,13 +280,13 @@ def process_datetime_DICTS(
|
||||
dicts: List[Dict[str, Any]], action: Literal["timestamp", "string"] = "timestamp"
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""check process datetime tuples"""
|
||||
for dict in dicts:
|
||||
for k, v in dict.items():
|
||||
for dic2t in dicts:
|
||||
for k, v in dic2t.items():
|
||||
if isinstance(v, datetime.datetime):
|
||||
if action == "string":
|
||||
dict[k] = v.strftime("%Y-%m-%d %H:%M:%S")
|
||||
dic2t[k] = v.strftime("%Y-%m-%d %H:%M:%S")
|
||||
elif action == "timestamp":
|
||||
dict[k] = v.timestamp()
|
||||
dic2t[k] = v.timestamp()
|
||||
else:
|
||||
raise ValueError(
|
||||
f"Unsupported action {action}. Use 'string' or 'timestamp'."
|
||||
|
||||
Reference in new issue
Block a user