Compare commits

...
2 Commits
2 changed files with 83 additions and 67 deletions

No files matched your search

+78 -62
View File
@@ -90,7 +90,8 @@ def better_connection_actuallyworking(connectionSettings: dict, conn_name="conn"
try:
with sqlcon.connect(**connectionSettings) as conn:
kwargs[conn_name] = conn
return func(*args, **kwargs)
r = func(*args, **kwargs)
return r
except sqlcon.Error as e:
print("SQL error from wrapper:", e)
raise
@@ -103,6 +104,36 @@ def better_connection_actuallyworking(connectionSettings: dict, conn_name="conn"
return decorator
# 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",
*,
conn=None,
):
"every thing passes to execute, return fetchone. lda test"
# print("i seqeralr", conn)
with conn.cursor(**cursor_settings) as r:
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 _:
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)
def funnying_funnytheFunnier(
*,
@@ -113,26 +144,26 @@ def funnying_funnytheFunnier(
return conn # WHAT NOT TO DO
@better_connection_actuallyworking(connectionSettings, conn_name="_conn")
# @better_connection_actuallyworking(connectionSettings, conn_name="_conn")
def fulltable(
tbaleName: str,
returnType: Literal["zip", "dict"] = "dict",
*,
_conn: Optional[
sqlcon.MySQLConnection | sqlcon.pooling.PooledMySQLConnection
] = None,
# *,
# _conn: Optional[
# sqlcon.MySQLConnection | sqlcon.pooling.PooledMySQLConnection
# ] = None,
) -> dict | tuple[list[tuple], list[str]]: # connection is inserted in by the thing
print(tbaleName)
if not (tableName := ifTableInDB(db_name, tbaleName)):
raise sqlcon.DatabaseError(f"{tbaleName} where tELL ME")
ifdict = returnType == "dict"
try:
with _conn.cursor(dictionary=ifdict, prepared=True) as cur:
_res = cur.execute(f"select * from {tableName}")
r = cur.fetchall()
r = query_once_fetch_lda(
(f"select * from {tableName}",),
{"dictionary": ifdict, "prepared": True},
fetch_many="all",
)
if ifdict:
return r
this = getColumename(tableName)
@@ -145,22 +176,24 @@ def fulltable(
raise e
@better_connection_actuallyworking(connectionSettings, conn_name="_conn")
# @better_connection_actuallyworking(connectionSettings, conn_name="_conn")
def getColumename(
tbaleName: str,
*,
_conn: Optional[
sqlcon.MySQLConnection | sqlcon.pooling.PooledMySQLConnection
] = None,
# *,
# _conn: Optional[
# sqlcon.MySQLConnection | sqlcon.pooling.PooledMySQLConnection
# ] = None,
) -> list[str]:
print(tbaleName)
try:
with _conn.cursor(prepared=True) as cur:
_res = cur.execute(
coln = query_once_fetch_lda(
execute_settings=(
"select column_name from INFORMATION_SCHEMA.COLUMNS where TABLE_NAME = ?",
(tbaleName,),
),
cursor_settings={"prepared": True},
fetch_many="all",
)
coln = cur.fetchall()
for a, i in enumerate(coln):
coln[a] = i[0]
return coln
@@ -172,21 +205,23 @@ def getColumename(
raise e
@better_connection_actuallyworking(connectionSettings, conn_name="_conn")
# @better_connection_actuallyworking(connectionSettings, conn_name="_conn")
def getTableNmaes(
db_name: str,
*,
_conn: Optional[
sqlcon.MySQLConnection | sqlcon.pooling.PooledMySQLConnection
] = None,
# *,
# _conn: Optional[
# sqlcon.MySQLConnection | sqlcon.pooling.PooledMySQLConnection
# ] = None,
) -> list[str]:
try:
with _conn.cursor(prepared=True) as cur:
_res = cur.execute(
coln = query_once_fetch_lda(
execute_settings=(
"select table_name from INFORMATION_SCHEMA.TABLES where TABLE_SCHEMA = ?",
(db_name,),
),
cursor_settings={"prepared": True},
fetch_many="all",
)
coln = cur.fetchall()
for a, i in enumerate(coln):
coln[a] = i[0]
return coln
@@ -209,28 +244,28 @@ csTrtb = dict(zip(clietColum, severColum))
print(scTrtb, csTrtb)
@better_connection_actuallyworking(connectionSettings, "_conn")
# @better_connection_actuallyworking(connectionSettings, "_conn")
def ifTableInDB(
db_name: str,
tabN: str,
*,
_conn: Optional[
sqlcon.MySQLConnection | sqlcon.pooling.PooledMySQLConnection
] = None,
# *,
# _conn: Optional[
# sqlcon.MySQLConnection | sqlcon.pooling.PooledMySQLConnection
# ] = None,
) -> str | None:
"""you dont believe it but it works like a true false, as python is good enough
if (vari := ifTableInDB(dbname, tablename)):"""
try:
with _conn.cursor(prepared=True) as cur:
_res = cur.execute(
r = query_once_fetch_lda(
(
"select table_name from information_schema.tables where table_schema = ? and table_name = ? Limit 1;",
(db_name, tabN),
),
{"prepared": True},
)
a = cur.fetchall()
if a:
return a[0][0] # deadass dont understand why this shi weird like that
if r:
return r[0]
return
except sqlcon.Error as e:
print("getTableNmaes has a", e)
@@ -275,7 +310,6 @@ class ServerLogic(serverhttp.BaseHTTPRequestHandler):
# tableli = getTableNmaes(db_name)
# print(getColumename(tableli[0])) # i know damn well
# print(fulltable(tableli[0]))
return
def close(self):
try:
@@ -433,9 +467,7 @@ def post_endp_lda(self):
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))
f"({','.join(wwwww)}) values ({','.join('?' for _ in wwwww)})"
)
print(
wwwww,
@@ -473,7 +505,6 @@ def post_endp_lda(self):
# what the fuck is this
def openFileManually(
theGuy: ServerLogic,
webDir: str,
@@ -493,7 +524,7 @@ def openFileManually(
theGuy.header_setup(rescode.NOT_FOUND)
theGuy.wfile.write(notfound404)
return
elif path.is_dir():
if path.is_dir():
path /= defaultFile
if not path.is_file():
theGuy.header_setup(rescode.NOT_FOUND)
@@ -575,29 +606,14 @@ def openFileSemiAutomatically(
theGuy.wfile.write(c)
# 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)()
def get_from_db(self, pathObj):
ee = fetchonce_lda(
ee = query_once_fetch_lda(
execute_settings=(
f"select neme, kewod, cod from {tabletbale} where neme = ?;",
(self.path.lstrip("/"),),
),
cursor_settings={"prepared": True},
fetch_many="once",
# fetch_many="one",
)
if not ee:
+5 -5
View File
@@ -105,7 +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]] = (
None,
@@ -203,7 +203,7 @@ def funny_trycatch_wrapper(
for oneThing in elses:
res = function_splitter(oneThing)
superreturn["relses"].append(res)
finally:
# finally:
for oneThing in finallies:
res = function_splitter(oneThing)
superreturn["rfinallies"].append(res)
@@ -239,7 +239,7 @@ def funny_singular_Simple_trycatch(Onetry: tuple):
this = function_splitter(Onetry)
except BaseException as e:
er = e
finally:
# finally:
return (this, er)
@@ -312,7 +312,7 @@ def funny_tuples_to_dicts_processor(colname: list[tuple], coldata: list[tuple]):
return new_dicts
getnewdictl = lambda d, l: {k: d[k] for k in d if k in l}
# getnewdictl = lambda d, l: {k: d[k] for k in d if k in l}
def getnewdict(d: dict, l: list) -> dict:
@@ -322,7 +322,7 @@ def getnewdict(d: dict, l: list) -> dict:
return {k: d[k] for k in d if k in l}
getnewdicttranslatel = lambda d, tltb: {tltb[k]: d[k] for k in d if k in tltb}
# getnewdicttranslatel = lambda d, tltb: {tltb[k]: d[k] for k in d if k in tltb}
def getnewdicttranslate(d: dict, tltb: dict) -> dict: