Data management and updating

Tracker for updating and getting data
Datamanager to serialize/deserialize
Main for initialization
Config for config
This commit is contained in:
2026-06-02 11:53:52 +02:00
parent 0e9d343b1e
commit 232e74c8c9
5 changed files with 207 additions and 0 deletions
+2
View File
@@ -174,3 +174,5 @@ cython_debug/
# PyPI configuration file
.pypirc
pyrightconfig.json
data/
+10
View File
@@ -0,0 +1,10 @@
ADMIN_PASSWORD = "admin123"
# Use UUIDs or simple string IDs
REGISTERED_USERS = {"user_1": "Alice", "user_2": "Bob", "user_3": "Charlie"}
MENU_ITEMS = {
"item_1": {"name": "Coffee", "price": 2.50},
"item_2": {"name": "Sandwich", "price": 6.00},
"item_3": {"name": "Drink", "price": 1.50},
}
+86
View File
@@ -0,0 +1,86 @@
from uuid import uuid4
from core.transaction import Transaction
class Tracker:
def __init__(self, data_manager) -> None:
self.data_manager = data_manager
self.users = {}
self.items = {}
self.transactions = []
self.debts = {}
self._load_all_data()
def _load_all_data(self):
self.users = self.data_manager.load_users()
self.items = self.data_manager.load_items()
self.debts = self.data_manager.load_debts()
self.transactions = self.data_manager.load_transactions()
def add_purchase(self, user_id, item_id):
if user_id not in self.users:
return {"status": "error", "message": "User not found"}
if item_id not in self.items:
return {"status": "error", "message": "Item not found"}
item = self.items[item_id]
if user_id not in self.debts:
self.debts[user_id] = 0.0
self.debts[user_id] += item.price
transaction = Transaction(str(uuid4()), user_id, item_id, item.price)
self.transactions.append(transaction)
self.data_manager.save_debts(self.debts)
self.data_manager.save_transactions(self.transactions)
return {
"status": "success",
"message": f"Added {item.name} to {self.users[user_id].name}'s debt",
"new_debt": self.debts[user_id],
}
def get_user_debt(self, user_id):
return self.debts.get(user_id, 0.0)
def get_all_debts(self):
return {
user_id: {
"name": self.users[user_id].name,
"debt": self.debts.get(user_id, 0.0),
}
for user_id in self.users
}
def adjust_debt(self, user_id, amount, admin_password):
if not self._verify_admin(admin_password):
return {"status": "error", "message": "Invalid password"}
if user_id not in self.users:
return {"status": "error", "message": "User not found"}
self.debts[user_id] -= amount
self.data_manager.save_debts(self.debts)
return {
"status": "success",
"message": f"Reduced debt by {amount}",
"new_debt": self.debts[user_id],
}
def _verify_admin(self, pwd):
from config import ADMIN_PASSWORD
return pwd == ADMIN_PASSWORD
def get_user_by_id(self, user_id):
return self.users.get(user_id)
def get_all_users(self):
return self.users
def get_items(self):
return self.items
+30
View File
@@ -0,0 +1,30 @@
from core.tracker import Tracker
from storage.data_manager import DataManager
from config import REGISTERED_USERS, MENU_ITEMS
def main():
data_manager = DataManager("data")
tracker = Tracker(data_manager)
if not tracker.users:
from core.user import User
from core.item import Item
for user_id, name in REGISTERED_USERS.items():
tracker.users[user_id] = User(user_id, name)
data_manager.save_users(tracker.users)
for item_id, item_data in MENU_ITEMS.items():
tracker.items[item_id] = Item(
item_id, item_data["name"], item_data["price"]
)
data_manager.save_items(tracker.items)
print("DebtTracker initialized successfully!")
print(f"Users: {[u.name for u in tracker.users.values()]}")
print(f"Menu items: {[m.name for m in tracker.items.values()]}")
if __name__ == "__main__":
main()
+79
View File
@@ -0,0 +1,79 @@
import json
from pathlib import Path
from core.user import User
from core.item import Item
from core.transaction import Transaction
class DataManager:
def __init__(self, data_dir) -> None:
self.data_dir = Path(data_dir)
self.data_dir.mkdir(exist_ok=True)
self.users_file = self.data_dir / "users.json"
self.items_file = self.data_dir / "items.json"
self.debts_file = self.data_dir / "debts.json"
self.transactions_file = self.data_dir / "transactions.json"
self._init_files()
def _init_files(self):
if not self.users_file.exists():
self._save_json(self.users_file, {})
if not self.items_file.exists():
self._save_json(self.items_file, {})
if not self.debts_file.exists():
self._save_json(self.debts_file, {})
if not self.transactions_file.exists():
self._save_json(self.transactions_file, [])
def load_users(self):
data = self._load_json(self.users_file)
return {
user_id: User.from_json(user_data) for user_id, user_data in data.items()
}
def load_items(self):
data = self._load_json(self.items_file)
return {
item_id: Item.from_json(item_data) for item_id, item_data in data.items()
}
def load_debts(self):
return self._load_json(self.debts_file)
def load_transactions(self):
data = self._load_json(self.transactions_file)
return [Transaction.from_json(t) for t in data]
def save_users(self, users):
data = {user_id: user.to_json() for user_id, user in users.items()}
self._save_json(self.users_file, data)
def save_items(self, items):
data = {item_id: item.to_json() for item_id, item in items.items()}
self._save_json(self.items_file, data)
def save_debts(self, debts):
self._save_json(self.debts_file, debts)
def save_transactions(self, transactions):
data = [t.to_json() for t in transactions]
self._save_json(self.transactions_file, data)
def _load_json(self, filepath):
try:
with open(filepath, "r") as f:
return json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
return (
{}
if filepath.name.endswith(".json")
and "transaction" not in filepath.name
else []
)
def _save_json(self, filepath, data):
with open(filepath, "w") as f:
json.dump(data, f, indent=2)