lda/python_web_server
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

best refactor youve seen in a bit

Browse Source
This commit is contained in:
lda committed 2025-05-01 22:10:24 +07:00
1 parent e2b1112e0f
commit 19d28994a2
1 file changed
+79 -61
Download Patch File Download Diff File Expand all files Collapse all files
the_server_handler.py
+79 -61
View File
@@ -90,7 +90,8 @@ def better_connection_actuallyworking(connectionSettings: dict, conn_name="conn"
try: try:
with sqlcon.connect(**connectionSettings) as conn: with sqlcon.connect(**connectionSettings) as conn:
kwargs[conn_name] = conn kwargs[conn_name] = conn
return func(*args, **kwargs) r = func(*args, **kwargs)
return r
except sqlcon.Error as e: except sqlcon.Error as e:
print("SQL error from wrapper:", e) print("SQL error from wrapper:", e)
raise raise
@@ -103,6 +104,36 @@ def better_connection_actuallyworking(connectionSettings: dict, conn_name="conn"
return decorator return decorator
# 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"] = "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) @better_connection_actuallyworking(connectionSettings)
def funnying_funnytheFunnier( def funnying_funnytheFunnier(
*, *,
@@ -113,26 +144,26 @@ def funnying_funnytheFunnier(
return conn # WHAT NOT TO DO return conn # WHAT NOT TO DO
# @better_connection_actuallyworking(connectionSettings, conn_name="_conn")
@better_connection_actuallyworking(connectionSettings, conn_name="_conn")
def fulltable( def fulltable(
tbaleName: str, tbaleName: str,
returnType: Literal["zip", "dict"] = "dict", returnType: Literal["zip", "dict"] = "dict",
*, # *,
_conn: Optional[ # _conn: Optional[
sqlcon.MySQLConnection | sqlcon.pooling.PooledMySQLConnection # sqlcon.MySQLConnection | sqlcon.pooling.PooledMySQLConnection
] = None, # ] = None,
) -> dict | tuple[list[tuple], list[str]]: # connection is inserted in by the thing ) -> dict | tuple[list[tuple], list[str]]: # connection is inserted in by the thing
print(tbaleName) print(tbaleName)
if not (tableName := ifTableInDB(db_name, tbaleName)): if not (tableName := ifTableInDB(db_name, tbaleName)):
raise sqlcon.DatabaseError(f"{tbaleName} where tELL ME") raise sqlcon.DatabaseError(f"{tbaleName} where tELL ME")
ifdict = returnType == "dict" ifdict = returnType == "dict"
try: try:
with _conn.cursor(dictionary=ifdict, prepared=True) as cur:
_res = cur.execute(f"select * from {tableName}") r = fetchonce_lda(
r = cur.fetchall() (f"select * from {tableName}",),
{"dictionary": ifdict, "prepared": True},
fetch_many="all",
)
if ifdict: if ifdict:
return r return r
this = getColumename(tableName) this = getColumename(tableName)
@@ -145,22 +176,24 @@ def fulltable(
raise e raise e
@better_connection_actuallyworking(connectionSettings, conn_name="_conn") # @better_connection_actuallyworking(connectionSettings, conn_name="_conn")
def getColumename( def getColumename(
tbaleName: str, tbaleName: str,
*, # *,
_conn: Optional[ # _conn: Optional[
sqlcon.MySQLConnection | sqlcon.pooling.PooledMySQLConnection # sqlcon.MySQLConnection | sqlcon.pooling.PooledMySQLConnection
] = None, # ] = None,
) -> list[str]: ) -> list[str]:
print(tbaleName) print(tbaleName)
try: try:
with _conn.cursor(prepared=True) as cur: coln = fetchonce_lda(
_res = cur.execute( execute_settings=(
"select column_name from INFORMATION_SCHEMA.COLUMNS where TABLE_NAME = ?", "select column_name from INFORMATION_SCHEMA.COLUMNS where TABLE_NAME = ?",
(tbaleName,), (tbaleName,),
) ),
coln = cur.fetchall() cursor_settings={"prepared": True},
fetch_many="all",
)
for a, i in enumerate(coln): for a, i in enumerate(coln):
coln[a] = i[0] coln[a] = i[0]
return coln return coln
@@ -172,21 +205,23 @@ def getColumename(
raise e raise e
@better_connection_actuallyworking(connectionSettings, conn_name="_conn") # @better_connection_actuallyworking(connectionSettings, conn_name="_conn")
def getTableNmaes( def getTableNmaes(
db_name: str, db_name: str,
*, # *,
_conn: Optional[ # _conn: Optional[
sqlcon.MySQLConnection | sqlcon.pooling.PooledMySQLConnection # sqlcon.MySQLConnection | sqlcon.pooling.PooledMySQLConnection
] = None, # ] = None,
) -> list[str]: ) -> list[str]:
try: try:
with _conn.cursor(prepared=True) as cur: coln = fetchonce_lda(
_res = cur.execute( execute_settings=(
"select table_name from INFORMATION_SCHEMA.TABLES where TABLE_SCHEMA = ?", "select table_name from INFORMATION_SCHEMA.TABLES where TABLE_SCHEMA = ?",
(db_name,), (db_name,),
) ),
coln = cur.fetchall() cursor_settings={"prepared": True},
fetch_many="all",
)
for a, i in enumerate(coln): for a, i in enumerate(coln):
coln[a] = i[0] coln[a] = i[0]
return coln return coln
@@ -209,29 +244,29 @@ csTrtb = dict(zip(clietColum, severColum))
print(scTrtb, csTrtb) print(scTrtb, csTrtb)
@better_connection_actuallyworking(connectionSettings, "_conn") # @better_connection_actuallyworking(connectionSettings, "_conn")
def ifTableInDB( def ifTableInDB(
db_name: str, db_name: str,
tabN: str, tabN: str,
*, # *,
_conn: Optional[ # _conn: Optional[
sqlcon.MySQLConnection | sqlcon.pooling.PooledMySQLConnection # sqlcon.MySQLConnection | sqlcon.pooling.PooledMySQLConnection
] = None, # ] = None,
) -> str | None: ) -> str | None:
"""you dont believe it but it works like a true false, as python is good enough """you dont believe it but it works like a true false, as python is good enough
if (vari := ifTableInDB(dbname, tablename)):""" if (vari := ifTableInDB(dbname, tablename)):"""
try: try:
with _conn.cursor(prepared=True) as cur: r = fetchonce_lda(
(
_res = cur.execute(
"select table_name from information_schema.tables where table_schema = ? and table_name = ? Limit 1;", "select table_name from information_schema.tables where table_schema = ? and table_name = ? Limit 1;",
(db_name, tabN), (db_name, tabN),
) ),
a = cur.fetchall() {"prepared": True},
if a: )
return a[0][0] # deadass dont understand why this shi weird like that if r:
return return r[0]
return
except sqlcon.Error as e: except sqlcon.Error as e:
print("getTableNmaes has a", e) print("getTableNmaes has a", e)
raise e raise e
@@ -275,7 +310,6 @@ class ServerLogic(serverhttp.BaseHTTPRequestHandler):
# tableli = getTableNmaes(db_name) # tableli = getTableNmaes(db_name)
# print(getColumename(tableli[0])) # i know damn well # print(getColumename(tableli[0])) # i know damn well
# print(fulltable(tableli[0])) # print(fulltable(tableli[0]))
return
def close(self): def close(self):
try: try:
@@ -473,7 +507,6 @@ def post_endp_lda(self):
# what the fuck is this # what the fuck is this
def openFileManually( def openFileManually(
theGuy: ServerLogic, theGuy: ServerLogic,
webDir: str, webDir: str,
@@ -575,21 +608,6 @@ def openFileSemiAutomatically(
theGuy.wfile.write(c) 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): def get_from_db(self, pathObj):
ee = fetchonce_lda( ee = fetchonce_lda(
execute_settings=( execute_settings=(
@@ -597,7 +615,7 @@ def get_from_db(self, pathObj):
(self.path.lstrip("/"),), (self.path.lstrip("/"),),
), ),
cursor_settings={"prepared": True}, cursor_settings={"prepared": True},
fetch_many="once", # fetch_many="one",
) )
if not ee: if not ee: