AdmMabe/py_luggage
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request 'Api' (#2) from api into master

Browse Source 
Reviewed-on: #2
This commit was merged in pull request #2.
This commit is contained in:
AdmMabe
2026-08-28 21:07:58 +02:00
parent 1581538448 c1931fae20
commit 5a692f9dd9
11 changed files with 454 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1
View File
@@ -1,2 +1,3 @@
__pycache__/
*.log
.env
api/__init__.py
View File
api/db.py
+93
View File
@@ -0,0 +1,93 @@
from fastapi import APIRouter, HTTPException
import logging
import db
from pydantic import BaseModel
#Logging
logging.basicConfig(
filename='api.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')
router = APIRouter(prefix="/db", tags=["db"])
@router.get("/help")
def get_help():
return {"name of clothes": "example",
"weight": "sample weight (only floats)",
"pieces": "sample pieces (only ints)"}
class Item_name(BaseModel):
name_of_clothes: str
class Config:
fields = {
"name_of_clothes": "name of clothes"
}
class Item_weight(BaseModel):
weight: float
class Item_piece(BaseModel):
pieces: int
@router.post("/insert", status_code=201)
def insert(item_name: Item_name, item_weight: Item_weight, item_pieces: Item_piece):
name = item_name.name_of_clothes.upper()
weight = item_weight.weight
pieces = item_pieces.pieces
db.new_part(name, weight, pieces)
logging.info("new Item created in DB")
return {"message": "Item inserted"}
@router.put("/update/{clothe_name}/pieces", status_code=202)
def update_pieces(clothe_name: str, item_pieces: Item_piece):
name = clothe_name.upper()
pieces = item_pieces.pieces
if db.check_clothe(name) != None:
db.update_pieces(name, pieces)
logging.info(f"Item {name} was set a new count")
return {"message": "pieces updated"}
else:
logging.error(f"Item {name} not found")
raise HTTPException(status_code=404, detail=f"Item {name} not found")
@router.put("/update/{clothe_name}/weight", status_code=202)
def update_weight(clothe_name: str, item: Item_weight):
name = clothe_name.upper()
weight = item.weight
if db.check_clothe(name) != None:
db.update_weight(name, weight)
logging.info(f"Item {name} was set a new weight")
return {"message": "weight updated"}
else:
logging.error(f"Item {name} not found")
raise HTTPException(status_code=404, detail=f"Item {name} not found")
@router.delete("/{clothe_name}/delete", status_code=204)
def delete(clothe_name: str):
name = clothe_name.upper()
if db.check_clothe(name) != None:
db.delete(name)
logging.critical(f"Item {name} was deleted")
return {"message": "Item was deleted"}
else:
logging.error(f"Item {name} not found")
raise HTTPException(status_code=404, detail=f"Item {name} not found")
api/health.py
+26
View File
@@ -0,0 +1,26 @@
from fastapi import APIRouter
import logging
#Logging
logging.basicConfig(
filename='api.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')
router = APIRouter(prefix="/health", tags=["health"])
@router.get("/")
def health_check():
logging.info("get health check request")
return {"status": "ok"}
api/query.py
+75
View File
@@ -0,0 +1,75 @@
from fastapi import APIRouter, HTTPException
import logging
import db
from pydantic import BaseModel
#Logging
logging.basicConfig(
filename='api.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')
router = APIRouter(prefix="/query", tags=["query"])
@router.get("/help", status_code=200)
def get_help():
return {"name of clothes": "example"}
@router.get("/{clothe_name}/weight", status_code=206)
def get_weight(clothe_name: str):
name = clothe_name.upper()
if db.check_clothe(name) != None:
weight = db.fetch_weight(name)
return {"weight": weight}
else:
logging.error(f"Item {name} not found")
raise HTTPException(status_code=404, detail=f"Item {name} not found")
@router.get("/weight", status_code=200)
def get_luggage_weight():
weight = db.fetch_luggage_weight()
return {"weight": weight}
@router.get("/{clothe_name}/pieces",status_code=206)
def get_pieces(clothe_name: str):
name = clothe_name.upper()
if db.check_clothe(name) != None:
pieces = db.fetch_piece(name)
return {"pieces": pieces}
else:
logging.error(f"Item {name} not found")
raise HTTPException(status_code=404, detail=f"Item {name} not found")
@router.get("/pieces", status_code=200)
def get_all_pieces():
pieces = db.fetch_all_pieces()
return {"pieces": pieces}
@router.get("/clothes")
def get_clothes():
clothes = db.fetch_clothes()
return {"clothes": clothes}
@router.get("/{clothe_name}/all", status_code=102)
def get_all(clothe_name: str):
name = clothe_name.upper()
if db.check_clothe(name) != None:
weight = db.fetch_weight(name)
pieces = db.fetch_piece(name)
return{
"clothe": name,
"weight": weight,
"pieces": pieces
}
else:
logging.error(f"Item {name} not found")
raise HTTPException(status_code=404, detail=f"Item {name} not found")
api_server.py
+7
View File
@@ -0,0 +1,7 @@
from fastapi import FastAPI
from api import health, db, query
app =FastAPI()
app.include_router(health.router)
app.include_router(db.router)
app.include_router(query.router)
cli.py
+191
View File
@@ -0,0 +1,191 @@
import db
import logging
import time
import os
import sys
print("loading")
#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 insert():
clothe = input("How is the clothe called? ").upper()
weight = float(input("The weight please? "))
pieces = int(input("How many pieces? ints only "))
db.new_part(clothe, weight, pieces)
print("it is inserted")
time.sleep(3)
os.system('cls' if os.name == 'nt' else 'clear')
menu()
def insert():
clothe = input("How is the clothe called? ").upper()
weight = float(input("The weight please? "))
pieces = int(input("How many pieces? ints only "))
db.new_part(clothe, weight, pieces)
print("it is inserted")
time.sleep(3)
os.system('cls' if os.name == 'nt' else 'clear')
menu()
def update():
clothe = input("How is the clothe called? ").upper()
weight = input("The weight please? ")
pieces = input("How many pieces? ints only ")
db.update_weight(clothe, weight)
db.update_pieces(clothe, pieces)
print("it is updated")
time.sleep(3)
os.system('cls' if os.name == 'nt' else 'clear')
menu()
def get_weight():
clothe = input("How is the clothe called? ").upper()
weight = db.fetch_weight(clothe)
print(weight)
time.sleep(3)
os.system('cls' if os.name == 'nt' else 'clear')
menu()
def get_luggage_weight():
weight = db.fetch_luggage_weight()
print(f"The luggage has a weight of {weight}g")
time.sleep(3)
os.system('cls' if os.name == 'nt' else 'clear')
menu()
def get_pieces():
clothe = input("How is the clothe called? ").upper()
pieces = db.fetch_piece(clothe)
print(f"You have {pieces} of {clothe}")
time.sleep(3)
os.system('cls' if os.name == 'nt' else 'clear')
menu()
def get_all_pieces():
pieces = db.fetch_all_pieces()
print(f"You have {pieces} in your luggage")
time.sleep(3)
os.system('cls' if os.name == 'nt' else 'clear')
menu()
def get_clothes():
clothes = db.fetch_clothes()
for clothe in clothes:
print(clothe)
time.sleep(3)
os.system('cls' if os.name == 'nt' else 'clear')
menu()
def get_weight_pieces():
clothe = input("How is the clothe called? ").upper()
weight = db.fetch_weight(clothe)
pieces = db.fetch_piece(clothe)
print(f"You have {pieces} of {clothe}. The weight is {weight}")
time.sleep(3)
os.system('cls' if os.name == 'nt' else 'clear')
menu()
def delete():
clothe = input("How is the clothe called? ").upper()
really = input("Do you really want to delete it? Y/N").upper()
if really == "Y":
db.delete(clothe)
print(f"{clothe} was deleted")
time.sleep(3)
else:
time.sleep(0.5)
logging.info("no item deleted")
os.system('cls' if os.name == 'nt' else 'clear')
menu()
def exit_programm():
time.sleep(0.5)
print(r"""
## ## ##########
### ### ## ##
#### #### ## ##
## #### ## ##########
## ## ## ## ##
## ## ## ##
""")
time.sleep(0.5)
print("")
print("")
print("Disconnect DB")
print("logout")
print("shutting down")
time.sleep(0.75)
os.system('cls' if os.name == 'nt' else 'clear')
sys.exit()
os._exit(1)
def menu():
print("Menu")
print("1: Insert Item")
print("2: Update Item")
print("3: Get weight of clothe")
print("4: Get weight of luggage")
print("5: Get pieces of clothe")
print("6: Get pieces of luggage")
print("7: Get clothes")
print("8: Get weight and pieces of clothes")
print("9: Delete Item")
print("0: EXIT")
choice = input("Type number here ")
os.system('cls' if os.name == 'nt' else 'clear')
logging.info(f"{choice} is choosen")
time.sleep(0.5)
if choice == "1":
insert()
if choice == "2":
update()
if choice == "3":
get_weight()
if choice == "4":
get_luggage_weight()
if choice == "5":
get_pieces()
if choice == "6":
get_all_pieces()
if choice == "7":
get_clothes()
if choice == "8":
get_weight_pieces()
if choice == "0":
exit_programm()
if choice == "9":
delete()
#The Terminal App Starts here
time.sleep(5)
print(r"""
## ## ##########
### ### ## ##
#### #### ## ##
## #### ## ##########
## ## ## ## ##
## ## ## ##
""")
logging.info("Menu is starting")
time.sleep(2.5)
menu()
db.py
+38 -18
View File
@@ -1,6 +1,7 @@
import mariadb
import logging
from dotenv import load_dotenv
import os
#Logging
@@ -33,14 +34,14 @@ def connect(db_user, db_psswd, db_host, db_port, db_name):
logging.error("Error connecting")
logging.debug(f"Error connecting to DB: {e}")
def new_part(cloth_type, weigth, pieces):
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, weigth, pieces) VALUES (?, ?, ?)"
executed_data = (cloth_type, weigth, pieces)
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")
@@ -52,13 +53,13 @@ def fetch_clothes():
logging.info("fetched clothes")
return clothes
def update_weigth(cloth_type, weigth):
def update_weight(cloth_type, weight):
cursor = connection.cursor()
executed = "UPDATE luggage SET weigth = ? WHERE clothes = ?"
executed_data = (weigth, cloth_type)
executed = "UPDATE luggage SET weight = ? WHERE clothes = ?"
executed_data = (weight, cloth_type)
cursor.execute(executed, executed_data)
connection.commit()
logging.info("weigth updated")
logging.info("weight updated")
def update_pieces(cloth_type, pieces):
if type(pieces) is not int:
@@ -71,19 +72,19 @@ def update_pieces(cloth_type, pieces):
connection.commit()
logging.info("pieces updated")
def fetch_weigth(cloth_type):
def fetch_weight(cloth_type):
cursor = connection.cursor()
cursor.execute("SELECT weigth FROM luggage WHERE clothes = ?", (cloth_type,))
weigth = cursor.fetchone()[0]
logging.info("fetched weigth")
return weigth
cursor.execute("SELECT weight FROM luggage WHERE clothes = ?", (cloth_type,))
weight = cursor.fetchone()[0]
logging.info("fetched weight")
return weight
def fetch_luggage_weigth():
def fetch_luggage_weight():
cursor = connection.cursor()
cursor.execute("SELECT sum(weigth) FROM luggage")
weigth = cursor.fetchone()[0]
logging.info("fetched luggage weigth")
return weigth
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()
@@ -104,3 +105,22 @@ def delete(cloth_type):
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)
main.py
+17 -169
View File
@@ -1,26 +1,13 @@
import db
import logging
import time
import os
import subprocess
import sys
from dotenv import load_dotenv
import os
#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)
print("loading")
open("api.log", "w").close()
open("app.log", "w").close()
db.connect(db_user, db_psswd, db_host, int(db_port), db_name)
open("api_server.log", "w").close()
#Logging
logging.basicConfig(
filename='app.log', # Log to a file
@@ -34,158 +21,19 @@ try:
except Exception as e:
logging.exception('An error occurred')
def insert():
clothe = input("How is the clothe called? ").upper()
weigth = float(input("The weigth please? "))
pieces = int(input("How many pieces? ints only "))
db.new_part(clothe, weigth, pieces)
print("it is inserted")
time.sleep(3)
os.system('cls' if os.name == 'nt' else 'clear')
menu()
def update():
clothe = input("How is the clothe called? ").upper()
weigth = input("The weigth please? ")
pieces = input("How many pieces? ints only ")
db.update_weigth(clothe, weigth)
db.update_pieces(clothe, pieces)
print("it is updated")
time.sleep(3)
os.system('cls' if os.name == 'nt' else 'clear')
menu()
env = os.environ.copy()
env["PYTHONIOENCODING"] = "utf-8"
def get_weigth():
clothe = input("How is the clothe called? ").upper()
weigth = db.fetch_weigth(clothe)
print(weigth)
time.sleep(3)
os.system('cls' if os.name == 'nt' else 'clear')
menu()
log_file = open("api_server.log", "w", encoding="utf-8")
api_process = subprocess.Popen(
["fastapi", "dev", "api_server.py"],
stdout=log_file,
stderr=log_file,
env=env
)
cli_process = subprocess.Popen([sys.executable, "cli.py"])
def get_luggage_weigth():
weigth = db.fetch_luggage_weigth()
print(f"The luggage has a weigth of {weigth}g")
time.sleep(3)
os.system('cls' if os.name == 'nt' else 'clear')
menu()
api_process.wait()
cli_process.wait()
def get_pieces():
clothe = input("How is the clothe called? ").upper()
pieces = db.fetch_piece(clothe)
print(f"You have {pieces} of {clothe}")
time.sleep(3)
os.system('cls' if os.name == 'nt' else 'clear')
menu()
def get_all_pieces():
pieces = db.fetch_all_pieces()
print(f"You have {pieces} in your luggage")
time.sleep(3)
os.system('cls' if os.name == 'nt' else 'clear')
menu()
def get_clothes():
clothes = db.fetch_clothes()
for clothe in clothes:
print(clothe)
time.sleep(3)
os.system('cls' if os.name == 'nt' else 'clear')
menu()
def get_weigth_pieces():
clothe = input("How is the clothe called? ").upper()
weigth = db.fetch_weigth(clothe)
pieces = db.fetch_piece(clothe)
print(f"You have {pieces} of {clothe}. The weigth is {weigth}")
time.sleep(3)
os.system('cls' if os.name == 'nt' else 'clear')
menu()
def delete():
clothe = input("How is the clothe called? ").upper()
really = input("Do you really want to delete it? Y/N").upper()
if really == "Y":
db.delete(clothe)
print(f"{clothe} was deleted")
time.sleep(3)
else:
time.sleep(0.5)
logging.info("no item deleted")
os.system('cls' if os.name == 'nt' else 'clear')
menu()
def exit_programm():
time.sleep(0.5)
print(r"""
## ## ##########
### ### ## ##
#### #### ## ##
## #### ## ##########
## ## ## ## ##
## ## ## ##
""")
time.sleep(0.5)
print("")
print("")
print("Disconnect DB")
print("logout")
print("shutting down")
time.sleep(0.75)
os.system('cls' if os.name == 'nt' else 'clear')
sys.exit()
def menu():
print("Menu")
print("1: Insert Item")
print("2: Update Item")
print("3: Get weigth of clothe")
print("4: Get weigth of luggage")
print("5: Get pieces of clothe")
print("6: Get pieces of luggage")
print("7: Get clothes")
print("8: Get weigth and pieces of clothes")
print("9: Delete Item")
print("0: EXIT")
choice = input("Type number here ")
os.system('cls' if os.name == 'nt' else 'clear')
logging.info(f"{choice} is choosen")
time.sleep(0.5)
if choice == "1":
insert()
if choice == "2":
update()
if choice == "3":
get_weigth()
if choice == "4":
get_luggage_weigth()
if choice == "5":
get_pieces()
if choice == "6":
get_all_pieces()
if choice == "7":
get_clothes()
if choice == "8":
get_weigth_pieces()
if choice == "0":
exit_programm()
if choice == "9":
delete()
#The Terminal App Starts here
time.sleep(5)
print(r"""
## ## ##########
### ### ## ##
#### #### ## ##
## #### ## ##########
## ## ## ## ##
## ## ## ##
""")
logging.info("Menu is starting")
time.sleep(2.5)
menu()
requirements.txt
+2 -1
View File
@@ -1,2 +1,3 @@
mariadb == 1.1.14
python-dotenv == 1.2.3
python-dotenv == 1.2.3
fastapi[standard] == 0.141.1
venv/pyvenv.cfg
+4 -4
View File
@@ -1,5 +1,5 @@
home = /usr/bin
home = C:\Users\bokaj\AppData\Local\Microsoft\WindowsApps\PythonSoftwareFoundation.Python.3.13_qbz5n2kfra8p0
include-system-site-packages = false
version = 3.12.3
executable = /usr/bin/python3.12
command = /usr/bin/python3 -m venv /home/jakob/Documents/py_luggage/venv
version = 3.13.14
executable = C:\Users\bokaj\AppData\Local\Microsoft\WindowsApps\PythonSoftwareFoundation.Python.3.13_qbz5n2kfra8p0\python.exe
command = C:\Users\bokaj\AppData\Local\Microsoft\WindowsApps\PythonSoftwareFoundation.Python.3.13_qbz5n2kfra8p0\python.exe -m venv J:\_Python\luggage\venv