127 lines
3.9 KiB
Python
127 lines
3.9 KiB
Python
import mariadb
|
|
import logging
|
|
from dotenv import load_dotenv
|
|
import os
|
|
|
|
#Logging
|
|
|
|
logging.basicConfig(
|
|
filename='app.log', # Log to a file
|
|
level=logging.DEBUG, # Log everything (DEBUG and above)
|
|
format='%(asctime)s %(levelname)s %(message)s'
|
|
)
|
|
|
|
logging.info('Application started')
|
|
try:
|
|
logging.debug('Debug info here')
|
|
except Exception as e:
|
|
logging.exception('An error occurred')
|
|
|
|
|
|
def connect(db_user, db_psswd, db_host, db_port, db_name):
|
|
try:
|
|
global connection
|
|
connection = mariadb.connect(user=db_user,
|
|
password=db_psswd,
|
|
host=db_host,
|
|
port=db_port,
|
|
database=db_name)
|
|
print("Connected to MariaDB successfully!")
|
|
logging.info("Connected to MariaDB successfully!")
|
|
|
|
except mariadb.Error as e:
|
|
print(f"Error connecting to DB: {e}")
|
|
logging.error("Error connecting")
|
|
logging.debug(f"Error connecting to DB: {e}")
|
|
|
|
def new_part(cloth_type, weight, pieces):
|
|
if type(pieces) is not int:
|
|
print("only ints allowed")
|
|
logging.error("only ints allowed")
|
|
else:
|
|
cursor = connection.cursor()
|
|
executed = "INSERT INTO luggage (clothes, weight, pieces) VALUES (?, ?, ?)"
|
|
executed_data = (cloth_type, weight, pieces)
|
|
cursor.execute(executed, executed_data)
|
|
connection.commit()
|
|
logging.info("new Insert is in DB")
|
|
|
|
def fetch_clothes():
|
|
cursor = connection.cursor()
|
|
cursor.execute("SELECT clothes FROM luggage")
|
|
clothes = [str(row[0]) for row in cursor.fetchall()]
|
|
logging.info("fetched clothes")
|
|
return clothes
|
|
|
|
def update_weight(cloth_type, weight):
|
|
cursor = connection.cursor()
|
|
executed = "UPDATE luggage SET weight = ? WHERE clothes = ?"
|
|
executed_data = (weight, cloth_type)
|
|
cursor.execute(executed, executed_data)
|
|
connection.commit()
|
|
logging.info("weight updated")
|
|
|
|
def update_pieces(cloth_type, pieces):
|
|
if type(pieces) is not int:
|
|
logging.error("only ints allowed")
|
|
else:
|
|
cursor = connection.cursor()
|
|
executed = "UPDATE luggage SET pieces = ? WHERE clothes = ?"
|
|
executed_data = (pieces, cloth_type)
|
|
cursor.execute(executed, executed_data)
|
|
connection.commit()
|
|
logging.info("pieces updated")
|
|
|
|
def fetch_weight(cloth_type):
|
|
cursor = connection.cursor()
|
|
cursor.execute("SELECT weight FROM luggage WHERE clothes = ?", (cloth_type,))
|
|
weight = cursor.fetchone()[0]
|
|
logging.info("fetched weight")
|
|
return weight
|
|
|
|
def fetch_luggage_weight():
|
|
cursor = connection.cursor()
|
|
cursor.execute("SELECT sum(weight) FROM luggage")
|
|
weight = cursor.fetchone()[0]
|
|
logging.info("fetched luggage weight")
|
|
return weight
|
|
|
|
def fetch_piece(cloth_type):
|
|
cursor = connection.cursor()
|
|
cursor.execute("SELECT pieces FROM luggage WHERE clothes = ?", (cloth_type,))
|
|
pieces = cursor.fetchone()[0]
|
|
logging.info("fetched luggage pieces")
|
|
return pieces
|
|
|
|
def fetch_all_pieces():
|
|
cursor = connection.cursor()
|
|
cursor.execute("SELECT sum(pieces) FROM luggage")
|
|
pieces = cursor.fetchone()[0]
|
|
logging.info("fetched all luggage pieces")
|
|
return pieces
|
|
|
|
def delete(cloth_type):
|
|
cursor = connection.cursor()
|
|
cursor.execute("DELETE FROM luggage WHERE clothes = ?;", (cloth_type,))
|
|
logging.critical(f"{cloth_type} was deleted")
|
|
connection.commit()
|
|
|
|
def check_clothe(clothe_type):
|
|
cursor = connection.cursor()
|
|
cursor.execute("SELECT (SELECT clothes FROM luggage WHERE clothes = ? LIMIT 1) AS clothes", (clothe_type,))
|
|
check = cursor.fetchone()[0]
|
|
return check
|
|
|
|
|
|
#using .env
|
|
load_dotenv()
|
|
db_name = os.getenv("DB_NAME")
|
|
db_psswd = os.getenv("DB_PASSWORD")
|
|
db_user = os.getenv("DB_USER")
|
|
db_port = os.getenv("DB_PORT") or "3306"
|
|
db_host = os.getenv("DB_HOST")
|
|
if db_host == None:
|
|
logging.error("no hostname provided")
|
|
os._exit(1)
|
|
connect(db_user, db_psswd, db_host, int(db_port), db_name)
|