2 Commits

Author SHA1 Message Date
e258358402 WIP 2025-06-07 15:02:36 +02:00
2c77ae6357 flake.nix: add libdib 2025-06-07 15:02:36 +02:00
12 changed files with 335 additions and 119 deletions

@ -1,4 +1,8 @@
import argparse
from pathlib import Path
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from dibbler.conf import config
@ -23,29 +27,62 @@ subparsers.add_parser("slabbedasker", help="Find out who is slabbedasker")
subparsers.add_parser("seed-data", help="Fill with mock data")
def _get_database_url_from_config() -> str:
"""Get the database URL from the configuration."""
url = config.get("database", "url")
if url is not None:
return url
url_file = config.get("database", "url_file")
if url_file is not None:
with Path(url_file).open() as file:
return file.read().strip()
raise ValueError("No database URL found in configuration.")
def _connect_to_database(url: str, **engine_args) -> Session:
try:
engine = create_engine(url, **engine_args)
sql_session = Session(engine)
except Exception as err:
print("Error: could not connect to database.")
print(err)
exit(1)
print(f"Debug: Connected to database at '{url}'")
return sql_session
def main():
args = parser.parse_args()
config.read(args.config)
database_url = _get_database_url_from_config()
sql_session = _connect_to_database(
database_url,
echo=config.getboolean("database", "echo_sql", fallback=False),
)
if args.subcommand == "loop":
import dibbler.subcommands.loop as loop
loop.main()
loop.main(sql_session)
elif args.subcommand == "create-db":
import dibbler.subcommands.makedb as makedb
makedb.main()
makedb.main(sql_session)
elif args.subcommand == "slabbedasker":
import dibbler.subcommands.slabbedasker as slabbedasker
slabbedasker.main()
slabbedasker.main(sql_session)
elif args.subcommand == "seed-data":
import dibbler.subcommands.seed_test_data as seed_test_data
seed_test_data.main()
seed_test_data.main(sql_session)
if __name__ == "__main__":

176
dibbler/menus/main.py Normal file

@ -0,0 +1,176 @@
import sys
import signal
import traceback
from sqlalchemy import (
event,
)
from sqlalchemy.orm import Session
from libdib.repl import (
NumberedCmd,
InteractiveItemSelector,
prompt_yes_no,
)
from dibbler.conf import config
class DibblerCli(NumberedCmd):
def __init__(self, sql_session: Session):
super().__init__()
self.sql_session = sql_session
self.sql_session_dirty = False
@event.listens_for(self.sql_session, "after_flush")
def mark_session_as_dirty(*_):
self.sql_session_dirty = True
self.prompt_header = "(unsaved changes)"
@event.listens_for(self.sql_session, "after_commit")
@event.listens_for(self.sql_session, "after_rollback")
def mark_session_as_clean(*_):
self.sql_session_dirty = False
self.prompt_header = None
# TODO: move to libdib.repl
@classmethod
def run_with_safe_exit_wrapper(cls, sql_session: Session):
tool = cls(sql_session)
if not config.getboolean("general", "stop_allowed"):
signal.signal(signal.SIGQUIT, signal.SIG_IGN)
if not config.getboolean("general", "stop_allowed"):
signal.signal(signal.SIGTSTP, signal.SIG_IGN)
while True:
try:
tool.cmdloop()
except KeyboardInterrupt:
if not tool.sql_session_dirty:
exit(0)
try:
print()
if prompt_yes_no(
"Are you sure you want to exit without saving?", default=False
):
raise KeyboardInterrupt
except KeyboardInterrupt:
if tool.sql_session is not None:
tool.sql_session.rollback()
exit(0)
except Exception:
print("Something went wrong.")
print(f"{sys.exc_info()[0]}: {sys.exc_info()[1]}")
if config.getboolean("general", "show_tracebacks"):
traceback.print_tb(sys.exc_info()[2])
def default(self, maybe_barcode: str):
raise NotImplementedError(
"This command is not implemented yet. Please use the numbered commands instead."
)
def do_buy(self, arg: str):
...
def do_product_list(self, arg: str):
...
def do_show_user(self, arg: str):
...
def do_user_list(self, arg: str):
...
def do_adjust_credit(self, arg: str):
...
def do_transfer(self, arg: str):
...
def do_add_stock(self, arg: str):
...
def do_add_edit(self, arg: str):
...
# AddEditMenu(self.sql_session).cmdloop()
def do_product_search(self, arg: str):
...
def do_statistics(self, arg: str):
...
def do_faq(self, arg: str):
...
def do_print_label(self, arg: str):
...
def do_exit(self, _: str):
if self.sql_session_dirty:
if prompt_yes_no("Would you like to save your changes?"):
self.sql_session.commit()
else:
self.sql_session.rollback()
exit(0)
funcs = {
0: {
"f": default,
"doc": "Choose / Add item with its ISBN",
},
1: {
"f": do_buy,
"doc": "Buy",
},
2: {
"f": do_product_list,
"doc": "Product List",
},
3: {
"f": do_show_user,
"doc": "Show User",
},
4: {
"f": do_user_list,
"doc": "User List",
},
5: {
"f": do_adjust_credit,
"doc": "Adjust Credit",
},
6: {
"f": do_transfer,
"doc": "Transfer",
},
7: {
"f": do_add_stock,
"doc": "Add Stock",
},
8: {
"f": do_add_edit,
"doc": "Add/Edit",
},
9: {
"f": do_product_search,
"doc": "Product Search",
},
10: {
"f": do_statistics,
"doc": "Statistics",
},
11: {
"f": do_faq,
"doc": "FAQ",
},
12: {
"f": do_print_label,
"doc": "Print Label",
},
13: {
"f": do_exit,
"doc": "Exit",
},
}

@ -1,79 +1,12 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
import random
import sys
import traceback
from ..conf import config
from ..lib.helpers import *
from ..menus import *
from sqlalchemy.orm import Session
random.seed()
from ..menus.main import DibblerCli
def main(sql_session: Session):
random.seed()
def main():
if not config.getboolean("general", "stop_allowed"):
signal.signal(signal.SIGQUIT, signal.SIG_IGN)
DibblerCli.run_with_safe_exit_wrapper(sql_session)
if not config.getboolean("general", "stop_allowed"):
signal.signal(signal.SIGTSTP, signal.SIG_IGN)
main = MainMenu(
"Dibbler main menu",
items=[
BuyMenu(),
ProductListMenu(),
ShowUserMenu(),
UserListMenu(),
AdjustCreditMenu(),
TransferMenu(),
AddStockMenu(),
Menu(
"Add/edit",
items=[
AddUserMenu(),
EditUserMenu(),
AddProductMenu(),
EditProductMenu(),
AdjustStockMenu(),
CleanupStockMenu(),
],
),
ProductSearchMenu(),
Menu(
"Statistics",
items=[
ProductPopularityMenu(),
ProductRevenueMenu(),
BalanceMenu(),
LoggedStatisticsMenu(),
],
),
FAQMenu(),
PrintLabelMenu(),
],
exit_msg="happy happy joy joy",
exit_confirm_msg="Really quit Dibbler?",
)
if not config.getboolean("general", "quit_allowed"):
main.exit_disallowed_msg = "You can check out any time you like, but you can never leave."
while True:
# noinspection PyBroadException
try:
main.execute()
except KeyboardInterrupt:
print("")
print("Interrupted.")
except:
print("Something went wrong.")
print(f"{sys.exc_info()[0]}: {sys.exc_info()[1]}")
if config.getboolean("general", "show_tracebacks"):
traceback.print_tb(sys.exc_info()[2])
else:
break
print("Restarting main menu.")
if __name__ == "__main__":
main()
exit(0)

@ -1,11 +1,8 @@
#!/usr/bin/python
from sqlalchemy.orm import Session
from dibbler.models import Base
from dibbler.db import engine
def main():
Base.metadata.create_all(engine)
if __name__ == "__main__":
main()
def main(sql_session: Session):
if not sql_session.bind:
raise RuntimeError("SQLAlchemy session is not bound to a database engine.")
Base.metadata.create_all(sql_session.bind)

@ -1,24 +1,21 @@
import json
from dibbler.db import Session
from pathlib import Path
from dibbler.models.Product import Product
from sqlalchemy.orm import Session
from dibbler.models.User import User
from dibbler.models import Product, User
JSON_FILE = Path(__file__).parent.parent.parent / "mock_data.json"
def clear_db(session):
session.query(Product).delete()
session.query(User).delete()
session.commit()
def clear_db(sql_session: Session):
sql_session.query(Product).delete()
sql_session.query(User).delete()
sql_session.commit()
def main():
session = Session()
clear_db(session)
def main(sql_session: Session):
clear_db(sql_session)
product_items = []
user_items = []
@ -43,6 +40,6 @@ def main():
)
user_items.append(user_item)
session.add_all(product_items)
session.add_all(user_items)
session.commit()
sql_session.add_all(product_items)
sql_session.add_all(user_items)
sql_session.commit()

@ -1,18 +1,12 @@
#!/usr/bin/python
from sqlalchemy.orm import Session
from dibbler.db import Session
# from dibbler.db import Session
from dibbler.models import User
def main():
# Start an SQL session
session = Session()
# Let's find all users with a negative credit
slabbedasker = session.query(User).filter(User.credit < 0).all()
def main(sql_session: Session):
# Let's find all users with a negative credit
slabbedasker = sql_session.query(User).filter(User.credit < 0).all()
for slubbert in slabbedasker:
print(f"{slubbert.name}, {slubbert.credit}")
if __name__ == "__main__":
main()

@ -6,7 +6,7 @@ input_encoding = 'utf8'
[database]
# url = "postgresql://robertem@127.0.0.1/pvvvv"
url = "sqlite:///test.db"
url = sqlite:///test.db
[limits]
low_credit_warning_limit = -100

76
flake.lock generated

@ -17,13 +17,65 @@
"type": "indirect"
}
},
"flake-utils_2": {
"inputs": {
"systems": "systems_2"
},
"locked": {
"lastModified": 1731533236,
"narHash": "sha256-l0KFg5HjrsfsO/JpG+r7fRrqm12kzFHyUHqHCVpMMbI=",
"owner": "numtide",
"repo": "flake-utils",
"rev": "11707dc2f618dd54ca8739b309ec4fc024de578b",
"type": "github"
},
"original": {
"id": "flake-utils",
"type": "indirect"
}
},
"libdib": {
"inputs": {
"flake-utils": "flake-utils_2",
"nixpkgs": "nixpkgs"
},
"locked": {
"lastModified": 1749301134,
"narHash": "sha256-JHLVV4ug8AgG71xhXEdmXozQfesXut6NUdXbBZNkD3c=",
"ref": "refs/heads/main",
"rev": "ca26131c22bb2833c81254dbabab6d785b9f37f0",
"revCount": 8,
"type": "git",
"url": "https://git.pvv.ntnu.no/Projects/libdib.git"
},
"original": {
"type": "git",
"url": "https://git.pvv.ntnu.no/Projects/libdib.git"
}
},
"nixpkgs": {
"locked": {
"lastModified": 1742288794,
"narHash": "sha256-Txwa5uO+qpQXrNG4eumPSD+hHzzYi/CdaM80M9XRLCo=",
"lastModified": 1749143949,
"narHash": "sha256-QuUtALJpVrPnPeozlUG/y+oIMSLdptHxb3GK6cpSVhA=",
"owner": "NixOS",
"repo": "nixpkgs",
"rev": "b6eaf97c6960d97350c584de1b6dcff03c9daf42",
"rev": "d3d2d80a2191a73d1e86456a751b83aa13085d7d",
"type": "github"
},
"original": {
"owner": "NixOS",
"ref": "nixos-unstable",
"repo": "nixpkgs",
"type": "github"
}
},
"nixpkgs_2": {
"locked": {
"lastModified": 1749143949,
"narHash": "sha256-QuUtALJpVrPnPeozlUG/y+oIMSLdptHxb3GK6cpSVhA=",
"owner": "NixOS",
"repo": "nixpkgs",
"rev": "d3d2d80a2191a73d1e86456a751b83aa13085d7d",
"type": "github"
},
"original": {
@ -36,7 +88,8 @@
"root": {
"inputs": {
"flake-utils": "flake-utils",
"nixpkgs": "nixpkgs"
"libdib": "libdib",
"nixpkgs": "nixpkgs_2"
}
},
"systems": {
@ -53,6 +106,21 @@
"repo": "default",
"type": "github"
}
},
"systems_2": {
"locked": {
"lastModified": 1681028828,
"narHash": "sha256-Vy1rq5AaRuLzOxct8nz4T6wlgyUR7zLU309k9mBC768=",
"owner": "nix-systems",
"repo": "default",
"rev": "da67096a3b9bf56a91d16901293e51ba5b49a27e",
"type": "github"
},
"original": {
"owner": "nix-systems",
"repo": "default",
"type": "github"
}
}
},
"root": "root",

@ -1,9 +1,13 @@
{
description = "Dibbler samspleisebod";
inputs.nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable";
inputs = {
nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable";
outputs = { self, nixpkgs, flake-utils }: let
libdib.url = "git+https://git.pvv.ntnu.no/Projects/libdib.git";
};
outputs = { self, nixpkgs, flake-utils, libdib }: let
inherit (nixpkgs) lib;
systems = [
@ -14,7 +18,12 @@
];
forAllSystems = f: lib.genAttrs systems (system: let
pkgs = nixpkgs.legacyPackages.${system};
pkgs = import nixpkgs {
inherit system;
overlays = [
libdib.overlays.default
];
};
in f system pkgs);
in {
packages = forAllSystems (system: pkgs: {

@ -1,6 +1,5 @@
{ lib
, python3Packages
, fetchFromGitHub
}:
python3Packages.buildPythonApplication {
pname = "dibbler";
@ -16,6 +15,7 @@ python3Packages.buildPythonApplication {
nativeBuildInputs = with python3Packages; [ setuptools ];
propagatedBuildInputs = with python3Packages; [
brother-ql
libdib
matplotlib
psycopg2
python-barcode

@ -12,6 +12,7 @@ mkShell {
(python.withPackages (ps: with ps; [
brother-ql
matplotlib
libdib
psycopg2
python-barcode
sqlalchemy

@ -15,11 +15,15 @@ dependencies = [
"SQLAlchemy >= 2.0, <2.1",
"brother-ql",
"matplotlib",
"libdib",
"psycopg2 >= 2.8, <2.10",
"python-barcode",
]
dynamic = ["version"]
[tool.uv.sources]
libdib = { git = "https://git.pvv.ntnu.no/Projects/libdib.git" }
[tool.setuptools.packages.find]
include = ["dibbler*"]