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

semi broken. some changes to the conn thing

Browse Source
This commit is contained in:
lda committed 2025-05-01 19:16:36 +07:00
1 parent dbebf03250
commit 517707b79c
3 files changed
+118 -95
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

THE_SERVER.py
+7 -6
View File
@@ -4,14 +4,15 @@ go dam
""" """
import http.server as serverhttp 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 # import mysql.connector as sqlcon
import sys
if not (sys.version_info >= (3, 11)): if not sys.version_info >= (3, 11):
raise Exception(f"python 3.11 up, u running {sys.version}") raise Exception(f"python 3.11 up, u running {sys.version}")
PORT = 8088 PORT = 8088
@@ -25,11 +26,11 @@ def runServer():
server = serverhttp.HTTPServer(("", PORT), ServerLogic) server = serverhttp.HTTPServer(("", PORT), ServerLogic)
try: try:
server.serve_forever() server.serve_forever()
except KeyboardInterrupt: except KeyboardInterrupt as c:
server.server_close() server.server_close()
# connectio.close() # connectio.close()
raise SystemExit("\nctrl c") raise SystemExit("\nctrl c") from c
except BaseException as e: except Exception as e:
print(e) print(e)
# finally: # finally:
# print("why am i printing") # print("why am i printing")
the_server_handler.py
+87 -79
View File
@@ -1,27 +1,29 @@
import http.server as serverhttp import http.server as serverhttp
# import urllib.parse # import urllib
import warnings import json
# import chardet
from http import HTTPStatus as rescode
import mimetypes import mimetypes
# import traceback # import traceback
import os import os
from typing import Literal
import mysql.connector as sqlcon # import urllib.parse
import warnings
from functools import wraps
# import chardet
from http import HTTPStatus as rescode
from pathlib import Path
from typing import Literal, Optional
# from mysql.connector import errorcode # from mysql.connector import errorcode
from dotenv import dotenv_values
from urllib import parse as urlparse # lmao the normal person from urllib import parse as urlparse # lmao the normal person
# import urllib import mysql.connector as sqlcon
import json
from functools import wraps
from pathlib import Path
from dotenv import dotenv_values from dotenv import dotenv_values
# from dotenv import dotenv_values
# import urllib, sql # import urllib, sql
class ServerLogic(serverhttp.BaseHTTPRequestHandler): ... class ServerLogic(serverhttp.BaseHTTPRequestHandler): ...
@@ -42,11 +44,9 @@ if os.path.isdir(webrootdir): # unfuckable
os.chdir(webrootdir) # move to safe part of world os.chdir(webrootdir) # move to safe part of world
curdir = os.path.curdir or os.getcwd() curdir = os.path.curdir or os.getcwd()
print(os.path.curdir, os.getcwd()) print(os.path.curdir, os.getcwd())
from things import ( from things import ( # process_datetime_tuples,; funny_tuples_to_dicts_processor,
process_datetime_DICTS,
# process_datetime_tuples,
# funny_tuples_to_dicts_processor,
getnewdicttranslate, getnewdicttranslate,
process_datetime_DICTS,
) )
dotenv_values1 = dotenv_values() ## change so fast dotenv_values1 = dotenv_values() ## change so fast
@@ -78,39 +78,37 @@ def get_that_connection(
return (connectio, None) return (connectio, None)
def better_connection_actuallyworking(connectionSettings: dict): def better_connection_actuallyworking(connectionSettings: dict, conn_name="conn"):
""" """
arg for func: Wraps a function and injects a SQL connection as a kwarg.
1. conn (do you hate life) The kwarg is named via conn_name, default is 'conn'.
2. thingElse
""" """
def ONETHING(func): def decorator(func):
@wraps(func) @wraps(func)
def huh(*a, **kw): def wrapper(*args, **kwargs):
# conn = sqlcon.connect(**connectionSettings)
try: try:
with sqlcon.connect( with sqlcon.connect(**connectionSettings) as conn:
**connectionSettings kwargs[conn_name] = conn
) as conn: # HOLY SHIT THANK YOU CHAT GPT return func(*args, **kwargs)
return func(conn, *a, **kw)
except sqlcon.Error as e: except sqlcon.Error as e:
print("oh hey errorSql FromtheWrapper", e) print("SQL error from wrapper:", e)
raise e raise
except Exception as e: except Exception as e:
print("betterConnection NoWorked", e) print("Connection wrapper failed:", e)
raise e raise
# finally: # they run even after return! try finally gon go crazy
# conn.close()
return huh return wrapper
return ONETHING return decorator
@better_connection_actuallyworking(connectionSettings) @better_connection_actuallyworking(connectionSettings)
def funnying_funnytheFunnier( def funnying_funnytheFunnier(
conn: sqlcon.MySQLConnection | sqlcon.pooling.PooledMySQLConnection, *,
conn: Optional[
sqlcon.MySQLConnection | sqlcon.pooling.PooledMySQLConnection
] = None,
): ):
return conn # WHAT NOT TO DO return conn # WHAT NOT TO DO
@@ -135,19 +133,22 @@ def funnying_funnytheFunnier(
# print(cur.fetchall()) # print(cur.fetchall())
@better_connection_actuallyworking(connectionSettings) @better_connection_actuallyworking(connectionSettings, conn_name="_conn")
def fulltable( def fulltable(
conn: sqlcon.MySQLConnection | sqlcon.pooling.PooledMySQLConnection,
tbaleName: str, tbaleName: str,
returnType: Literal["zip", "dict"] = "dict", returnType: Literal["zip", "dict"] = "dict",
*,
_conn: Optional[
sqlcon.MySQLConnection | sqlcon.pooling.PooledMySQLConnection
] = 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: with _conn.cursor(dictionary=ifdict, prepared=True) as cur:
res = cur.execute(f"select * from {tableName}") _res = cur.execute(f"select * from {tableName}")
r = cur.fetchall() r = cur.fetchall()
if ifdict: if ifdict:
return r return r
@@ -161,14 +162,18 @@ def fulltable(
raise e raise e
@better_connection_actuallyworking(connectionSettings) @better_connection_actuallyworking(connectionSettings, conn_name="_conn")
def getColumename( def getColumename(
conn: sqlcon.MySQLConnection | sqlcon.pooling.PooledMySQLConnection, tbaleName: str tbaleName: str,
*,
_conn: Optional[
sqlcon.MySQLConnection | sqlcon.pooling.PooledMySQLConnection
] = None,
) -> list[str]: ) -> list[str]:
print(tbaleName) print(tbaleName)
try: try:
with conn.cursor(prepared=True) as cur: with _conn.cursor(prepared=True) as cur:
res = cur.execute( _res = cur.execute(
"select column_name from INFORMATION_SCHEMA.COLUMNS where TABLE_NAME = ?", "select column_name from INFORMATION_SCHEMA.COLUMNS where TABLE_NAME = ?",
(tbaleName,), (tbaleName,),
) )
@@ -184,13 +189,17 @@ def getColumename(
raise e raise e
@better_connection_actuallyworking(connectionSettings) @better_connection_actuallyworking(connectionSettings, conn_name="_conn")
def getTableNmaes( def getTableNmaes(
conn: sqlcon.MySQLConnection | sqlcon.pooling.PooledMySQLConnection, db_name: str db_name: str,
*,
_conn: Optional[
sqlcon.MySQLConnection | sqlcon.pooling.PooledMySQLConnection
] = None,
) -> list[str]: ) -> list[str]:
try: try:
with conn.cursor(prepared=True) as cur: with _conn.cursor(prepared=True) as cur:
res = cur.execute( _res = cur.execute(
"select table_name from INFORMATION_SCHEMA.TABLES where TABLE_SCHEMA = ?", "select table_name from INFORMATION_SCHEMA.TABLES where TABLE_SCHEMA = ?",
(db_name,), (db_name,),
) )
@@ -217,19 +226,22 @@ csTrtb = dict(zip(clietColum, severColum))
print(scTrtb, csTrtb) print(scTrtb, csTrtb)
@better_connection_actuallyworking(connectionSettings) @better_connection_actuallyworking(connectionSettings, "_conn")
def ifTableInDB( def ifTableInDB(
conn: sqlcon.MySQLConnection | sqlcon.pooling.PooledMySQLConnection,
db_name: str, db_name: str,
tabN: str, tabN: str,
*,
_conn: Optional[
sqlcon.MySQLConnection | sqlcon.pooling.PooledMySQLConnection
] = 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: with _conn.cursor(prepared=True) as cur:
res = cur.execute( _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),
) )
@@ -253,13 +265,13 @@ def openFileManually(
theGuy: ServerLogic, theGuy: ServerLogic,
webDir: str, webDir: str,
defaultFile: str | None = "index.html", defaultFile: str | None = "index.html",
dir: str | None = None, dir_: str | None = None,
headers: dict | None = None, headers: dict | None = None,
rootDir: str | None = curdir, rootDir: str | None = curdir,
) -> None: ) -> None:
print(webDir, dir, headers) print(webDir, dir_, headers)
if dir: if dir_:
filename = dir filename = dir_
else: else:
filename = webDir.lstrip("/") filename = webDir.lstrip("/")
path = Path(rootDir, filename) path = Path(rootDir, filename)
@@ -295,28 +307,26 @@ def openFileManually(
with open(path, "rb") as file: with open(path, "rb") as file:
asfd = file.read() asfd = file.read()
headers |= {(gay or "content-type"): finalMime, headers |= {(gay or "content-type"): finalMime, "content-length": str(len(asfd))}
"content-length": str(len(asfd))
}
theGuy.header_setup(headers=headers) theGuy.header_setup(headers=headers)
return theGuy.wfile.write(asfd)
def openFileSemiAutomatically( def openFileSemiAutomatically(
theGuy: ServerLogic, theGuy: ServerLogic,
webDir: str, webDir: str,
defaultFile: str | list[str] | tuple[str] | None = "index.html", defaultFile: str | list[str] | tuple[str] | None = "index.html",
dir: str | None = None, dir_: str | None = None,
headers: dict | None = None, headers: dict | None = None,
rootDir: str | None = curdir, rootDir: str | None = curdir,
): ):
c = bytes() c = bytes()
pf = Path() pf = Path()
a = Path(dir) if dir else Path(webDir.lstrip("/")) a = Path(dir_) if dir_ else Path(webDir.lstrip("/"))
if not a.exists(): if not a.exists():
c = notfound404 c = notfound404
elif a.is_dir(): elif a.is_dir():
if type(defaultFile) == str: if isinstance(defaultFile, str):
p = a / defaultFile p = a / defaultFile
if p.is_file(): if p.is_file():
with open(p, "rb") as f: with open(p, "rb") as f:
@@ -338,19 +348,17 @@ def openFileSemiAutomatically(
else: else:
c = notfound404 c = notfound404
mt, enct = mimetypes.guess_type(pf) mt, _enct = mimetypes.guess_type(pf)
if not headers: if not headers:
headers = {} headers = {}
if mt.split("/")[0] == "text": if mt.split("/")[0] == "text":
fg = mt + "; charset=utf-8" mt = mt + "; charset=utf-8"
theGuy.header_setup( if mt:
200, headers |= {
headers "content-type": mt,
| {
"content-type": fg or mt,
# "content-length": str(len(c)) # "content-length": str(len(c))
}, }
) theGuy.header_setup(200, headers)
theGuy.wfile.write(c) theGuy.wfile.write(c)
@@ -360,8 +368,8 @@ def openFileSemiAutomatically(
class ServerLogic(serverhttp.BaseHTTPRequestHandler): class ServerLogic(serverhttp.BaseHTTPRequestHandler):
def __init__(self, *a, **kw): # AHAHHAHAHAHAHAHAHA # def __init__(self, *a, **kw): # AHAHHAHAHAHAHAHAHA
super().__init__(*a, **kw) # super().__init__(*a, **kw)
def reallyEveryTime(self) -> tuple: def reallyEveryTime(self) -> tuple:
print("all headers:\n", self.headers) print("all headers:\n", self.headers)
@@ -401,8 +409,8 @@ class ServerLogic(serverhttp.BaseHTTPRequestHandler):
self.send_response(code) self.send_response(code)
for key, value in headers.items(): for key, value in headers.items():
self.send_header(key, value) # why! self.send_header(key, value) # why!
else: # else:
pass # pass
self.end_headers() self.end_headers()
def do_GET(self): def do_GET(self):
@@ -519,11 +527,11 @@ class ServerLogic(serverhttp.BaseHTTPRequestHandler):
""" """
super().__init__(rc, case, *a, **kw) super().__init__(rc, case, *a, **kw)
self.stupid = case self.stupid = case
if type(rc) == int: if isinstance(rc, int):
print("what") print("what")
self.ldarescode = rescode(rc) # why do i have to do this self.ldarescode = rescode(rc) # why do i have to do this
print(self.ldarescode) print(self.ldarescode)
elif type(rc) == rescode: elif isinstance(rc, rescode):
print("good") print("good")
self.ldarescode = rc self.ldarescode = rc
print(self.ldarescode) print(self.ldarescode)
@@ -620,7 +628,7 @@ class ServerLogic(serverhttp.BaseHTTPRequestHandler):
) )
conn.commit() # WHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWH conn.commit() # WHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWHYWH
cur.execute( cur.execute(
f"select * from tbale where neme = ?", "select * from tbale where neme = ?",
(data_dict["neme"],), (data_dict["neme"],),
) )
print("res of:", cur.fetchone()) print("res of:", cur.fetchone())
things.py
+24 -10
View File
@@ -1,10 +1,11 @@
from collections import defaultdict import datetime
import random import random
import string import string
from typing import Callable, Dict, Any, List, Literal, Tuple, Union from collections import defaultdict
from itertools import zip_longest from itertools import zip_longest
import datetime from typing import Any, Callable, Dict, List, Literal, Tuple, Union
import copy
# import copy
# import os, sys # import os, sys
@@ -142,12 +143,19 @@ def function_splitter(
def funny_trycatch_wrapper( def funny_trycatch_wrapper(
tries: list[tuple[Callable | str, tuple[Any, ...] | Any, Dict[str, Any]]], tries: list[tuple[Callable | str, tuple[Any, ...] | Any, Dict[str, Any]]],
excepts: dict[ excepts: (
dict[
BaseException, BaseException,
list[tuple[Callable | str, tuple[Any, ...] | Any, Dict[str, Any]]], list[tuple[Callable | str, tuple[Any, ...] | Any, Dict[str, Any]]],
] = {BaseException: [(lambda x: print(f"Error: {x}"))]}, ]
elses: list[tuple[Callable | str, tuple[Any, ...] | Any, Dict[str, Any]]] = [], | None
finallies: list[tuple[Callable | str, tuple[Any, ...] | Any, Dict[str, Any]]] = [], ) = None,
elses: (
list[tuple[Callable | str, tuple[Any, ...] | Any, Dict[str, Any]]] | None
) = None,
finallies: (
list[tuple[Callable | str, tuple[Any, ...] | Any, Dict[str, Any]]] | None
) = None,
): ):
""" """
Try, except, else, and finally wrapper for multiple functions or exec strings. Try, except, else, and finally wrapper for multiple functions or exec strings.
@@ -165,6 +173,11 @@ def funny_trycatch_wrapper(
"rfinallies": [], "rfinallies": [],
} }
""" """
# tries = tries or []
excepts = excepts or {BaseException: [lambda x: print(f"Error: {x}")]}
elses = elses or []
finallies = finallies or []
superreturn = { superreturn = {
"rtries": [], "rtries": [],
"rexcepts": {}, "rexcepts": {},
@@ -273,9 +286,10 @@ def process_datetime_DICTS(
dict[k] = v.strftime("%Y-%m-%d %H:%M:%S") dict[k] = v.strftime("%Y-%m-%d %H:%M:%S")
elif action == "timestamp": elif action == "timestamp":
dict[k] = v.timestamp() dict[k] = v.timestamp()
else: else:
raise ValueError(f"Unsupported action {action}. Use 'string' or 'timestamp'.") # bet this will run raise ValueError(
f"Unsupported action {action}. Use 'string' or 'timestamp'."
) # bet this will run
return dicts # in place processing return dicts # in place processing