102 lines
3.1 KiB
Python
102 lines
3.1 KiB
Python
import logging
|
|
from dotenv import load_dotenv
|
|
import os
|
|
from models import Items
|
|
from sqlalchemy import create_engine, func
|
|
from sqlalchemy.orm import sessionmaker
|
|
from urllib.parse import quote_plus
|
|
|
|
#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')
|
|
|
|
#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)
|
|
|
|
engine = create_engine(f"mariadb+mariadbconnector://{db_user}:{quote_plus(db_psswd)}@{db_host}:{db_port}/{db_name}")
|
|
Session = sessionmaker(bind=engine)
|
|
session = Session()
|
|
|
|
|
|
def new_part(cloth_type, weight, pieces):
|
|
existing = session.query(Items).filter_by(clothes=cloth_type).first()
|
|
if existing:
|
|
logging.debug("Item already inserted")
|
|
else:
|
|
if type(pieces) is not int:
|
|
print("only ints allowed")
|
|
logging.error("only ints allowed")
|
|
else:
|
|
new_item = Items(clothes=cloth_type, weight=weight, pieces=pieces)
|
|
session.add(new_item)
|
|
session.commit()
|
|
logging.info("new Insert is in DB")
|
|
|
|
def update_weight(cloth_type, weight):
|
|
item = session.query(Items).filter_by(clothes=cloth_type).first()
|
|
item.weight = weight
|
|
session.commit()
|
|
logging.info("weight updated")
|
|
|
|
def update_pieces(cloth_type, pieces):
|
|
if type(pieces) is not int:
|
|
logging.error("only ints allowed")
|
|
else:
|
|
item = session.query(Items).filter_by(clothes=cloth_type).first()
|
|
item.pieces = pieces
|
|
session.commit()
|
|
logging.info("pieces updated")
|
|
|
|
def delete(cloth_type):
|
|
logging.critical(f"{cloth_type} was deleted")
|
|
session.query(Items).filter_by(clothes=cloth_type).delete()
|
|
session.commit()
|
|
|
|
|
|
def fetch_clothes():
|
|
logging.info("fetched clothes")
|
|
return [row[0] for row in session.query(Items.clothes).all()]
|
|
|
|
def fetch_weight(cloth_type):
|
|
logging.info("fetched weight")
|
|
result = session.query(Items.weight).filter_by(clothes=cloth_type).first()
|
|
return result[0] if result else None
|
|
|
|
def fetch_luggage_weight():
|
|
logging.info("fetched whole weight")
|
|
result = session.query(func.sum(Items.weight)).scalar()
|
|
return result if result else None
|
|
|
|
def fetch_piece(cloth_type):
|
|
logging.info("fetched pieces")
|
|
result = session.query(Items.pieces).filter_by(clothes=cloth_type).first()
|
|
return result[0] if result else None
|
|
|
|
def fetch_all_pieces():
|
|
logging.info("fetched all pieces")
|
|
result = session.query(Items.pieces).all()
|
|
return [row[0] for row in session.query(Items.pieces).all()]
|
|
|
|
def check_clothe(clothe_type):
|
|
exists = session.query(Items.clothes).filter_by(clothes=clothe_type).first() is not None
|
|
return exists
|
|
|