Add scripts to seed data for testing + misc

data/arbeidsrom_smal_hylle_5.csv

@@ -12,14 +12,10 @@ isbn, note, bookcase, shelf
 9781292024967, abstract alg, arbeidsrom_smal, 5
 9780471728979, kreyzig, arbeidsrom_smal, 5
 9781847762399, calc 1, arbeidsrom_smal, 5
-9781847762399, calc 1 again, arbeidsrom_smal, 5
 9781787267763, calc 1 nome, arbeidsrom_smal, 5
 9781787267770, calc 2 nome, arbeidsrom_smal, 5
 9780199208258, non lin ode, arbeidsrom_smal, 5
 9788251915953, tabeller, arbeidsrom_smal, 5
-9788251915953, taeller 2, arbeidsrom_smal, 5
-9788251915953, tabeller 3, arbeidsrom_smal, 5
-9788251915953, tabeller 4, arbeidsrom_smal, 5
 9780750304009, fractals and chaos, arbeidsrom_smal, 5
 9788241902116, geometri, arbeidsrom_smal, 5
 9781620402788, simpsons, arbeidsrom_smal, 5

worblehat/devscripts/seed_content_for_deadline_daemon.py

@@ -0,0 +1,117 @@
+from datetime import datetime, timedelta
+
+from worblehat.models import (
+  BookcaseItem,
+  BookcaseItemBorrowing,
+  BookcaseItemBorrowingQueue,
+  DeadlineDaemonLastRunDatetime,
+)
+
+from worblehat.services.config import Config
+
+from .seed_test_data import main as seed_test_data_main
+
+
+def clear_db(sql_session):
+  sql_session.query(BookcaseItemBorrowingQueue).delete()
+  sql_session.query(BookcaseItemBorrowing).delete()
+  sql_session.query(DeadlineDaemonLastRunDatetime).delete()
+  sql_session.commit()
+
+# NOTE: feel free to change this function to suit your needs
+#       it's just a quick and dirty way to get some data into the database
+#       for testing the deadline daemon - oysteikt 2024
+def main(sql_session):
+  borrow_warning_days = [timedelta(days=int(d)) for d in Config['deadline_daemon.warn_days_before_borrowing_deadline']]
+  queue_warning_days = [timedelta(days=int(d)) for d in Config['deadline_daemon.warn_days_before_expiring_queue_position_deadline']]
+  queue_expire_days = int(Config['deadline_daemon.days_before_queue_position_expires'])
+
+  clear_db(sql_session)
+  seed_test_data_main(sql_session)
+
+  books = sql_session.query(BookcaseItem).all()
+
+  last_run_datetime = datetime.now() - timedelta(days=16)
+  last_run = DeadlineDaemonLastRunDatetime(last_run_datetime)
+  sql_session.add(last_run)
+
+  # Create at least one item that is borrowed and not supposed to be returned yet
+  borrowing = BookcaseItemBorrowing(
+    item=books[0],
+    username='test_borrower_still_borrowing',
+  )
+  borrowing.start_time = last_run_datetime - timedelta(days=1)
+  borrowing.end_time = datetime.now() - timedelta(days=6)
+  sql_session.add(borrowing)
+
+  # Create at least one item that is borrowed and is supposed to be returned soon
+  borrowing = BookcaseItemBorrowing(
+    item=books[1],
+    username='test_borrower_return_soon',
+  )
+  borrowing.start_time = last_run_datetime - timedelta(days=1)
+  borrowing.end_time = datetime.now() - timedelta(days=2)
+  sql_session.add(borrowing)
+
+  # Create at least one item that is borrowed and is overdue
+  borrowing = BookcaseItemBorrowing(
+    item=books[2],
+    username='test_borrower_overdue',
+  )
+  borrowing.start_time = datetime.now() - timedelta(days=1)
+  borrowing.end_time = datetime.now() + timedelta(days=1)
+  sql_session.add(borrowing)
+
+  # Create at least one item that is in the queue and is not supposed to be borrowed yet
+  queue_item = BookcaseItemBorrowingQueue(
+    item=books[3],
+    username='test_queue_user_still_waiting',
+  )
+  queue_item.entered_queue_time = last_run_datetime - timedelta(days=1)
+  borrowing = BookcaseItemBorrowing(
+    item=books[3],
+    username='test_borrower_return_soon',
+  )
+  borrowing.start_time = last_run_datetime - timedelta(days=1)
+  borrowing.end_time = datetime.now() - timedelta(days=2)
+  sql_session.add(queue_item)
+  sql_session.add(borrowing)
+
+  # Create at least three items that is in the queue and two items were just returned
+  for i in range(3):
+    queue_item = BookcaseItemBorrowingQueue(
+      item=books[4 + i],
+      username=f'test_queue_user_{i}',
+    )
+    sql_session.add(queue_item)
+
+  for i in range(3):
+    borrowing = BookcaseItemBorrowing(
+      item=books[4 + i],
+      username=f'test_borrower_returned_{i}',
+    )
+    borrowing.start_time = last_run_datetime - timedelta(days=2)
+    borrowing.end_time = datetime.now() + timedelta(days=1)
+
+    if i != 2:
+      borrowing.delivered = datetime.now() - timedelta(days=1)
+
+    sql_session.add(borrowing)
+
+  # Create at least one item that has been in the queue for so long that the queue position should expire
+  queue_item = BookcaseItemBorrowingQueue(
+    item=books[7],
+    username='test_queue_user_expired',
+  )
+  queue_item.entered_queue_time = datetime.now() - timedelta(days=15)
+
+  # Create at least one item that has been in the queue for so long that the queue position should expire,
+  # but the queue person has already been notified
+  queue_item = BookcaseItemBorrowingQueue(
+    item=books[8],
+    username='test_queue_user_expired_notified',
+  )
+  queue_item.entered_queue_time = datetime.now() - timedelta(days=15)
+
+  sql_session.commit()
+

worblehat/devscripts/seed_test_data.py

@@ -0,0 +1,87 @@
+import csv
+
+from pathlib import Path
+from datetime import datetime, timedelta
+
+from worblehat.models import (
+  Bookcase,
+  BookcaseItem,
+  BookcaseShelf,
+  MediaType,
+  Language,
+)
+from worblehat.services.config import Config
+
+
+CSV_FILE = Path(__file__).parent.parent.parent / 'data' / 'arbeidsrom_smal_hylle_5.csv'
+
+def clear_db(sql_session):
+    sql_session.query(BookcaseItem).delete()
+    sql_session.query(BookcaseShelf).delete()
+    sql_session.query(Bookcase).delete()
+    sql_session.query(MediaType).delete()
+    sql_session.query(Language).delete()
+    sql_session.commit()
+
+def main(sql_session):
+    clear_db(sql_session)
+
+    media_type = MediaType(
+        name='Book',
+        description='A book',
+    )
+    sql_session.add(media_type)
+
+    language = Language(
+        name='Norwegian',
+        iso639_1_code='no',
+    )
+    sql_session.add(language)
+
+    seed_case = Bookcase(
+        name='seed_case',
+        description='test bookcase with test data',
+    )
+    sql_session.add(seed_case)
+
+    seed_shelf_1 = BookcaseShelf(
+      row=1,
+      column=1,
+      bookcase=seed_case,
+      description='test shelf with test data 1',
+    )
+    seed_shelf_2 = BookcaseShelf(
+      row=2,
+      column=1,
+      bookcase=seed_case,
+      description='test shelf with test data 2',
+    )
+    sql_session.add(seed_shelf_1)
+    sql_session.add(seed_shelf_2)
+
+    bookcase_items = []
+    with open(CSV_FILE) as csv_file:
+        csv_reader = csv.reader(csv_file, delimiter=',')
+
+        next(csv_reader)
+        for row in csv_reader:
+            item = BookcaseItem(
+                isbn=row[0],
+                name=row[1],
+            )
+            item.media_type = media_type
+            item.language = language
+            bookcase_items.append(item)
+
+    half = len(bookcase_items) // 2
+    first_half = bookcase_items[:half]
+    second_half = bookcase_items[half:]
+
+    for item in first_half:
+        item.bookcase_shelf = seed_shelf_1
+
+    for item in second_half:
+        item.bookcase_shelf = seed_shelf_2
+
+    sql_session.add_all(bookcase_items)
+    sql_session.commit()

worblehat/main.py

@@ -60,6 +60,19 @@ def main():
         WorblehatCli.run_with_safe_exit_wrapper(sql_session)
         exit(0)
 
+    if args.command == 'devscripts':
+        sql_session = _connect_to_database(echo=Config['logging.debug_sql'])
+        if args.script == 'seed-content-for-deadline-daemon':
+            from .devscripts.seed_content_for_deadline_daemon import main
+            main(sql_session)
+        elif args.script == 'seed-test-data':
+            from .devscripts.seed_test_data import main
+            main(sql_session)
+        else:
+            print(f'Error: no such script: {args.script}')
+            exit(1)
+        exit(0)
+
     if args.command == 'flask-dev':
         flask_dev_main()
         exit(0)

worblehat/models/BookcaseItem.py

@@ -2,10 +2,11 @@ from __future__ import annotations
 from typing import TYPE_CHECKING
 
 from sqlalchemy import (
+  ForeignKey,
   Integer,
   SmallInteger,
   String,
-  ForeignKey,
+  Text,
 )
 from sqlalchemy.orm import (
     Mapped,
@@ -31,8 +32,11 @@ if TYPE_CHECKING:
     from .Language import Language
     from .MediaType import MediaType
 
-class BookcaseItem(Base, UidMixin, UniqueNameMixin):
+from worblehat.flaskapp.database import db
+
+class BookcaseItem(Base, UidMixin):
     isbn: Mapped[int] = mapped_column(String, unique=True, index=True)
+    name: Mapped[str] = mapped_column(Text, index=True)
     owner: Mapped[str] = mapped_column(String, default='PVV')
     amount: Mapped[int] = mapped_column(SmallInteger, default=1)
 
@@ -64,3 +68,12 @@ class BookcaseItem(Base, UidMixin, UniqueNameMixin):
         self.name = name
         self.isbn = isbn
         self.owner = owner
+
+    @classmethod
+    def get_by_isbn(cls, isbn: str, sql_session: Session = db.session) -> Self | None:
+        """
+        NOTE:
+        This method defaults to using the flask_sqlalchemy session.
+        It will not work outside of a request context, unless another session is provided.
+        """
+        return sql_session.query(cls).where(cls.isbn == isbn).one_or_none()

worblehat/models/DeadlineDaemonLastRunDatetime.py

@@ -21,3 +21,7 @@ class DeadlineDaemonLastRunDatetime(Base):
     )
     uid: Mapped[bool] = mapped_column(Boolean, primary_key=True, default=True)
     time: Mapped[datetime] = mapped_column(DateTime, default=datetime.now())
+
+    def __init__(self, time: datetime | None = None):
+        if time is not None:
+            self.time = time

worblehat/services/argument_parser.py

@@ -31,6 +31,14 @@ subparsers.add_parser(
     help = 'Start the web interface in production mode',
 )
 
+subparsers.add_parser(
+    'devscripts',
+    help = 'Run development scripts',
+).add_argument(
+    'script',
+    help = 'The development script to run',
+)
+
 arg_parser.add_argument(
     '-V',
     '--version',