Projects/worblehat
4
2
Fork 0
Code Issues 19 Pull Requests Actions Activity

treewide: extract all read-only queries into their own files

Browse Source
This commit is contained in:
Oystein Kristoffer Tveit
2026-07-26 21:52:02 +09:00
parent 931bd2d63e
commit 20c48b3989
15 changed files with 563 additions and 237 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/worblehat/cli/main.py
+16 -35
View File
@@ -1,4 +1,3 @@
from datetime import datetime
from textwrap import dedent
from libdib.repl import (
@@ -6,13 +5,17 @@ from libdib.repl import (
NumberedCmd,
prompt_yes_no,
)
from sqlalchemy import (
event,
select,
)
from sqlalchemy import event
from sqlalchemy.orm import Session
from worblehat.models import *
from worblehat.queries import (
find_bookcase_item_by_isbn,
find_media_type_by_name,
list_active_borrowings,
list_all_queue_items,
list_overdue_borrowings,
)
from worblehat.services import (
create_bookcase_item_from_isbn,
is_valid_isbn,
@@ -72,11 +75,7 @@ class WorblehatCli(NumberedCmd):
print(f" {item.name} - {item.amount} copies")
def do_show_borrowed_queued(self, _: str) -> None:
borrowed_items = self.sql_session.scalars(
select(BookcaseItemBorrowing)
.where(BookcaseItemBorrowing.delivered.is_(None))
.order_by(BookcaseItemBorrowing.end_time),
).all()
borrowed_items = list_active_borrowings(self.sql_session)
if len(borrowed_items) == 0:
print("No borrowed items found.")
@@ -89,11 +88,7 @@ class WorblehatCli(NumberedCmd):
print()
queued_items = self.sql_session.scalars(
select(BookcaseItemBorrowingQueue).order_by(
BookcaseItemBorrowingQueue.entered_queue_time,
),
).all()
queued_items = list_all_queue_items(self.sql_session)
if len(queued_items) == 0:
print("No queued items found.")
@@ -137,9 +132,7 @@ class WorblehatCli(NumberedCmd):
media_type_selector = InteractiveItemSelector(
cls=MediaType,
sql_session=self.sql_session,
default=self.sql_session.scalars(
select(MediaType).where(MediaType.name.ilike("book")),
).one(),
default=find_media_type_by_name(self.sql_session, "book"),
)
media_type_selector.cmdloop()
@@ -161,12 +154,10 @@ class WorblehatCli(NumberedCmd):
return
if (
existing_item := self.sql_session.scalars(
select(BookcaseItem)
.where(BookcaseItem.isbn == isbn)
.join(BookcaseItemBorrowing)
.join(BookcaseItemBorrowingQueue),
).one_or_none()
existing_item := find_bookcase_item_by_isbn(
self.sql_session,
isbn,
)
) is not None:
print(f'\nFound existing item for isbn "{isbn}"')
BookcaseItemCli(
@@ -191,17 +182,7 @@ class WorblehatCli(NumberedCmd):
).cmdloop()
def do_show_slabbedasker(self, _: str) -> None:
slubberter = self.sql_session.scalars(
select(BookcaseItemBorrowing)
.join(BookcaseItem)
.where(
BookcaseItemBorrowing.end_time < datetime.now(),
BookcaseItemBorrowing.delivered.is_(None),
)
.order_by(
BookcaseItemBorrowing.end_time,
),
).all()
slubberter = list_overdue_borrowings(self.sql_session)
if len(slubberter) == 0:
print("No slubberts found. Life is good.")
src/worblehat/cli/subclis/advanced_options.py
+8 -23
View File
@@ -2,10 +2,14 @@ from libdib.repl import (
InteractiveItemSelector,
NumberedCmd,
)
from sqlalchemy import select
from sqlalchemy.orm import Session
from worblehat.models import Bookcase, BookcaseShelf
from worblehat.queries import (
find_bookcase_by_name,
find_bookcase_shelf,
list_bookcase_shelfs_ordered,
)
class AdvancedOptionsCli(NumberedCmd):
@@ -20,12 +24,7 @@ class AdvancedOptionsCli(NumberedCmd):
print("Error: name cannot be empty")
continue
if (
self.sql_session.scalars(
select(Bookcase).where(Bookcase.name == name),
).one_or_none()
is not None
):
if find_bookcase_by_name(self.sql_session, name) is not None:
print(f"Error: a bookcase with name {name} already exists")
continue
@@ -67,13 +66,7 @@ class AdvancedOptionsCli(NumberedCmd):
break
if (
self.sql_session.scalars(
select(BookcaseShelf).where(
BookcaseShelf.bookcase == bookcase,
BookcaseShelf.column == column,
BookcaseShelf.row == row,
),
).one_or_none()
find_bookcase_shelf(self.sql_session, bookcase, column, row)
is not None
):
print(
@@ -95,15 +88,7 @@ class AdvancedOptionsCli(NumberedCmd):
self.sql_session.flush()
def do_list_bookcases(self, _: str) -> None:
bookcase_shelfs = self.sql_session.scalars(
select(BookcaseShelf)
.join(Bookcase)
.order_by(
Bookcase.name,
BookcaseShelf.column,
BookcaseShelf.row,
),
).all()
bookcase_shelfs = list_bookcase_shelfs_ordered(self.sql_session)
bookcase_uid = None
for shelf in bookcase_shelfs:
src/worblehat/cli/subclis/bookcase_item.py
+23 -66
View File
@@ -8,7 +8,6 @@ from libdib.repl import (
format_date,
prompt_yes_no,
)
from sqlalchemy import select
from sqlalchemy.orm import Session
from worblehat.models import (
@@ -19,6 +18,15 @@ from worblehat.models import (
Language,
MediaType,
)
from worblehat.queries import (
find_bookcase_item_by_isbn,
find_bookcase_item_by_name,
has_active_borrowing,
is_in_borrowing_queue,
list_active_borrowings_for_item,
list_borrowings_for_isbn,
list_pending_queue_items_for_item,
)
from worblehat.services.bookcase_item import (
create_bookcase_item_from_isbn,
is_valid_isbn,
@@ -77,37 +85,16 @@ class BookcaseItemCli(NumberedCmd):
return username
def _has_active_borrowing(self, username: str) -> bool:
return (
self.sql_session.scalars(
select(BookcaseItemBorrowing).where(
BookcaseItemBorrowing.username == username,
BookcaseItemBorrowing.item == self.bookcase_item,
BookcaseItemBorrowing.delivered.is_(None),
),
).one_or_none()
is not None
)
return has_active_borrowing(self.sql_session, username, self.bookcase_item)
def _has_borrowing_queue_item(self, username: str) -> bool:
return (
self.sql_session.scalars(
select(BookcaseItemBorrowingQueue).where(
BookcaseItemBorrowingQueue.username == username,
BookcaseItemBorrowingQueue.item == self.bookcase_item,
),
).one_or_none()
is not None
)
return is_in_borrowing_queue(self.sql_session, username, self.bookcase_item)
def do_borrow(self, _: str) -> None:
active_borrowings = self.sql_session.scalars(
select(BookcaseItemBorrowing)
.where(
BookcaseItemBorrowing.item == self.bookcase_item,
BookcaseItemBorrowing.delivered.is_(None),
)
.order_by(BookcaseItemBorrowing.end_time),
).all()
active_borrowings = list_active_borrowings_for_item(
self.sql_session,
self.bookcase_item,
)
if len(active_borrowings) >= self.bookcase_item.amount:
print("This item is currently not available")
@@ -157,15 +144,7 @@ class BookcaseItemCli(NumberedCmd):
)
def do_deliver(self, _: str) -> None:
borrowings = self.sql_session.scalars(
select(BookcaseItemBorrowing)
.join(
BookcaseItem,
BookcaseItem.uid == BookcaseItemBorrowing.fk_bookcase_item_uid,
)
.where(BookcaseItem.isbn == self.bookcase_item.isbn)
.order_by(BookcaseItemBorrowing.username),
).all()
borrowings = list_borrowings_for_isbn(self.sql_session, self.bookcase_item.isbn)
if len(borrowings) == 0:
print("No one seems to have borrowed this item")
@@ -194,28 +173,16 @@ class BookcaseItemCli(NumberedCmd):
print(f"Successfully delivered the item for {borrowing.username}")
def do_extend_borrowing(self, _: str) -> None:
borrowings = self.sql_session.scalars(
select(BookcaseItemBorrowing)
.join(
BookcaseItem,
BookcaseItem.uid == BookcaseItemBorrowing.fk_bookcase_item_uid,
)
.where(BookcaseItem.isbn == self.bookcase_item.isbn)
.order_by(BookcaseItemBorrowing.username),
).all()
borrowings = list_borrowings_for_isbn(self.sql_session, self.bookcase_item.isbn)
if len(borrowings) == 0:
print("No one seems to have borrowed this item")
return
borrowing_queue = self.sql_session.scalars(
select(BookcaseItemBorrowingQueue)
.where(
BookcaseItemBorrowingQueue.item == self.bookcase_item,
BookcaseItemBorrowingQueue.item_became_available_time is None,
)
.order_by(BookcaseItemBorrowingQueue.entered_queue_time),
).all()
borrowing_queue = list_pending_queue_items_for_item(
self.sql_session,
self.bookcase_item,
)
if len(borrowing_queue) != 0:
print(
@@ -302,12 +269,7 @@ class EditBookcaseCli(NumberedCmd):
print("Error: name cannot be empty")
continue
if (
self.sql_session.scalars(
select(BookcaseItem).where(BookcaseItem.name == name),
).one_or_none()
is not None
):
if find_bookcase_item_by_name(self.sql_session, name) is not None:
print(f"Error: an item with name {name} already exists")
continue
@@ -326,12 +288,7 @@ class EditBookcaseCli(NumberedCmd):
print("Error: ISBN is not valid")
continue
if (
self.sql_session.scalars(
select(BookcaseItem).where(BookcaseItem.isbn == isbn),
).one_or_none()
is not None
):
if find_bookcase_item_by_isbn(self.sql_session, isbn) is not None:
print(f"Error: an item with ISBN {isbn} already exists")
continue
src/worblehat/cli/subclis/bookcase_shelf_selector.py
+19 -12
View File
@@ -1,11 +1,14 @@
from libdib.repl import InteractiveItemSelector
from sqlalchemy import select
from sqlalchemy.orm import Session
from worblehat.models import (
Bookcase,
BookcaseShelf,
)
from worblehat.queries import (
find_bookcase_shelf,
list_bookcase_shelf_positions,
)
def select_bookcase_shelf(
@@ -15,29 +18,33 @@ def select_bookcase_shelf(
) -> BookcaseShelf:
def __complete_bookshelf_selection(session: Session, cls: type, arg: str):
args = arg.split("-")
query = select(cls.row, cls.column).where(cls.bookcase == bookcase)
column = None
row = None
try:
if arg != "" and len(args) > 0:
query = query.where(cls.column == int(args[0]))
column = int(args[0])
if len(args) > 1:
query = query.where(cls.row == int(args[1]))
row = int(args[1])
except ValueError:
return []
result = session.execute(query).all()
result = list_bookcase_shelf_positions(session, bookcase, column, row)
return [f"{c}-{r}" for r, c in result]
def __execute_bookshelf_selection(session: Session, cls: type, arg: str):
shelf = find_bookcase_shelf(
session,
bookcase,
int(arg.split("-")[0]),
int(arg.split("-")[1]),
)
return [shelf] if shelf is not None else []
print(prompt)
bookcase_shelf_selector = InteractiveItemSelector(
cls=BookcaseShelf,
sql_session=sql_session,
execute_selection=lambda session, cls, arg: session.scalars(
select(cls).where(
cls.bookcase == bookcase,
cls.column == int(arg.split("-")[0]),
cls.row == int(arg.split("-")[1]),
),
).all(),
execute_selection=__execute_bookshelf_selection,
complete_selection=__complete_bookshelf_selection,
)
src/worblehat/cli/subclis/search.py
+10 -16
View File
@@ -2,10 +2,14 @@ from libdib.repl import (
NumberedCmd,
NumberedItemSelector,
)
from sqlalchemy import select
from sqlalchemy.orm import Session
from worblehat.models import Author, BookcaseItem
from worblehat.queries import (
list_bookcase_items_by_owner,
search_authors_by_name,
search_bookcase_item_owners,
search_bookcase_items_by_title,
)
class SearchCli(NumberedCmd):
@@ -21,9 +25,7 @@ class SearchCli(NumberedCmd):
while (input_text := input("Enter title: ")) == "":
pass
items = self.sql_session.scalars(
select(BookcaseItem).where(BookcaseItem.name.ilike(f"%{input_text}%")),
).all()
items = search_bookcase_items_by_title(self.sql_session, input_text)
if len(items) == 0:
print("No items found.")
@@ -43,9 +45,7 @@ class SearchCli(NumberedCmd):
while (input_text := input("Enter author name: ")) == "":
pass
author = self.sql_session.scalars(
select(Author).where(Author.name.ilike(f"%{input_text}%")),
).all()
author = search_authors_by_name(self.sql_session, input_text)
if len(author) == 0:
print("No authors found.")
@@ -80,11 +80,7 @@ class SearchCli(NumberedCmd):
while (input_text := input("Enter username: ")) == "":
pass
users = self.sql_session.scalars(
select(BookcaseItem.owner)
.where(BookcaseItem.owner.ilike(f"%{input_text}%"))
.distinct(),
).all()
users = search_bookcase_item_owners(self.sql_session, input_text)
if len(users) == 0:
print("No users found.")
@@ -100,9 +96,7 @@ class SearchCli(NumberedCmd):
return None
selected_user = selector.result
items = self.sql_session.scalars(
select(BookcaseItem).where(BookcaseItem.owner == selected_user),
).all()
items = list_bookcase_items_by_owner(self.sql_session, selected_user)
selector = NumberedItemSelector(
items=items,
src/worblehat/deadline_daemon/main.py
+39 -85
View File
@@ -2,7 +2,6 @@ import logging
from datetime import datetime, timedelta
from textwrap import dedent
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from worblehat.models import (
@@ -10,6 +9,15 @@ from worblehat.models import (
BookcaseItemBorrowingQueue,
DeadlineDaemonLastRunDatetime,
)
from worblehat.queries import (
find_last_run,
find_next_queue_position,
list_close_deadline_borrowings,
list_expiring_queue_positions,
list_newly_available_queue_items,
list_overdue_queue_positions,
list_undelivered_overdue_borrowings,
)
from worblehat.services.config import Config
from worblehat.services.email import send_email
@@ -21,9 +29,7 @@ class DeadlineDaemon:
self.sql_session = sql_session
self.last_run = self.sql_session.scalars(
select(DeadlineDaemonLastRunDatetime),
).one_or_none()
self.last_run = find_last_run(self.sql_session)
if self.last_run is None:
logging.info("No previous run found, assuming this is the first run")
@@ -158,16 +164,6 @@ class DeadlineDaemon:
# EMAIL ROUTINES #
##################
def _sql_subtract_date(self, x: datetime, y: timedelta):
if self.sql_session.bind.dialect.name == "sqlite":
# SQLite does not support timedelta in queries
return func.datetime(x, f"-{y.days} days")
if self.sql_session.bind.dialect.name == "postgresql":
return x - y
raise NotImplementedError(
f"Unsupported dialect: {self.sql_session.bind.dialect.name}",
)
def send_close_deadline_reminder_mails(self) -> None:
logging.info("Sending mails for items with a closing deadline")
@@ -175,30 +171,22 @@ class DeadlineDaemon:
days = [int(d) for d in Config["deadline_daemon.warn_days_before_borrowing_deadline"]]
for day in days:
borrowings_to_remind = self.sql_session.scalars(
select(BookcaseItemBorrowing).where(
self._sql_subtract_date(
BookcaseItemBorrowing.end_time,
timedelta(days=day),
).between(
self.last_run_datetime,
self.current_run_datetime,
),
BookcaseItemBorrowing.delivered.is_(None),
),
).all()
borrowings_to_remind = list_close_deadline_borrowings(
self.sql_session,
day,
self.last_run_datetime,
self.current_run_datetime,
)
for borrowing in borrowings_to_remind:
self._send_close_deadline_mail(borrowing)
def send_overdue_mails(self) -> None:
logging.info("Sending mails for overdue items")
to_remind = self.sql_session.scalars(
select(BookcaseItemBorrowing).where(
BookcaseItemBorrowing.end_time < self.current_run_datetime,
BookcaseItemBorrowing.delivered.is_(None),
),
).all()
to_remind = list_undelivered_overdue_borrowings(
self.sql_session,
self.current_run_datetime,
)
for borrowing in to_remind:
self._send_overdue_mail(borrowing)
@@ -206,24 +194,11 @@ class DeadlineDaemon:
def send_newly_available_mails(self) -> None:
logging.info("Sending mails about newly available items")
newly_available = self.sql_session.scalars(
select(BookcaseItemBorrowingQueue)
.join(
BookcaseItemBorrowing,
BookcaseItemBorrowing.fk_bookcase_item_uid
== BookcaseItemBorrowingQueue.fk_bookcase_item_uid,
)
.where(
BookcaseItemBorrowingQueue.expired.is_(False),
BookcaseItemBorrowing.delivered.is_not(None),
BookcaseItemBorrowing.delivered.between(
self.last_run_datetime,
self.current_run_datetime,
),
)
.order_by(BookcaseItemBorrowingQueue.entered_queue_time)
.group_by(BookcaseItemBorrowingQueue.fk_bookcase_item_uid),
).all()
newly_available = list_newly_available_queue_items(
self.sql_session,
self.last_run_datetime,
self.current_run_datetime,
)
for queue_item in newly_available:
logging.info(
@@ -243,23 +218,11 @@ class DeadlineDaemon:
for d in Config["deadline_daemon.warn_days_before_expiring_queue_position_deadline"]
]
for day in days:
queue_positions_to_remind = self.sql_session.scalars(
select(BookcaseItemBorrowingQueue)
.join(
BookcaseItemBorrowing,
BookcaseItemBorrowing.fk_bookcase_item_uid
== BookcaseItemBorrowingQueue.fk_bookcase_item_uid,
)
.where(
self._sql_subtract_date(
BookcaseItemBorrowingQueue.item_became_available_time + timedelta(days=day),
timedelta(days=day),
).between(
self.last_run_datetime,
self.current_run_datetime,
),
),
).all()
queue_positions_to_remind = list_expiring_queue_positions(
self.sql_session,
self.last_run_datetime,
self.current_run_datetime,
)
for queue_position in queue_positions_to_remind:
self._send_expiring_queue_position_mail(queue_position, day)
@@ -271,14 +234,11 @@ class DeadlineDaemon:
Config["deadline_daemon.days_before_queue_position_expires"],
)
overdue_queue_positions = self.sql_session.scalars(
select(BookcaseItemBorrowingQueue).where(
BookcaseItemBorrowingQueue.item_became_available_time
+ timedelta(days=queue_position_expiry_days)
< self.current_run_datetime,
BookcaseItemBorrowingQueue.expired.is_(False),
),
).all()
overdue_queue_positions = list_overdue_queue_positions(
self.sql_session,
queue_position_expiry_days,
self.current_run_datetime,
)
for queue_position in overdue_queue_positions:
logging.info(
@@ -287,16 +247,10 @@ class DeadlineDaemon:
queue_position.expired = True
next_queue_position = self.sql_session.scalars(
select(BookcaseItemBorrowingQueue)
.where(
BookcaseItemBorrowingQueue.fk_bookcase_item_uid
== queue_position.fk_bookcase_item_uid,
BookcaseItemBorrowingQueue.item_became_available_time.is_(None),
)
.order_by(BookcaseItemBorrowingQueue.entered_queue_time)
.limit(1),
).one_or_none()
next_queue_position = find_next_queue_position(
self.sql_session,
queue_position.fk_bookcase_item_uid,
)
self._send_queue_position_expired_mail(queue_position)
src/worblehat/queries/__init__.py
+65
View File
@@ -0,0 +1,65 @@
from .author import search_authors_by_name
from .bookcase import find_bookcase_by_name
from .bookcase_item import (
find_bookcase_item_by_isbn,
find_bookcase_item_by_name,
list_bookcase_items_by_owner,
search_bookcase_item_owners,
search_bookcase_items_by_title,
)
from .bookcase_shelf import (
find_bookcase_shelf,
list_bookcase_shelf_positions,
list_bookcase_shelfs_ordered,
)
from .borrowing import (
has_active_borrowing,
list_active_borrowings,
list_active_borrowings_for_item,
list_borrowings_for_isbn,
list_overdue_borrowings,
)
from .borrowing_queue import (
is_in_borrowing_queue,
list_all_queue_items,
list_pending_queue_items_for_item,
)
from .deadline_daemon import (
find_last_run,
find_next_queue_position,
list_close_deadline_borrowings,
list_expiring_queue_positions,
list_newly_available_queue_items,
list_overdue_queue_positions,
list_undelivered_overdue_borrowings,
)
from .media_type import find_media_type_by_name
__all__ = [
"find_bookcase_by_name",
"find_bookcase_item_by_isbn",
"find_bookcase_item_by_name",
"find_bookcase_shelf",
"find_last_run",
"find_media_type_by_name",
"find_next_queue_position",
"has_active_borrowing",
"is_in_borrowing_queue",
"list_active_borrowings",
"list_active_borrowings_for_item",
"list_all_queue_items",
"list_bookcase_items_by_owner",
"list_bookcase_shelf_positions",
"list_bookcase_shelfs_ordered",
"list_borrowings_for_isbn",
"list_close_deadline_borrowings",
"list_expiring_queue_positions",
"list_newly_available_queue_items",
"list_overdue_borrowings",
"list_overdue_queue_positions",
"list_pending_queue_items_for_item",
"list_undelivered_overdue_borrowings",
"search_authors_by_name",
"search_bookcase_item_owners",
"search_bookcase_items_by_title",
]
src/worblehat/queries/author.py
+12
View File
@@ -0,0 +1,12 @@
from sqlalchemy import select
from sqlalchemy.orm import Session
from worblehat.models import Author
def search_authors_by_name(sql_session: Session, text: str) -> list[Author]:
return list(
sql_session.scalars(
select(Author).where(Author.name.ilike(f"%{text}%")),
).all(),
)
src/worblehat/queries/bookcase.py
+10
View File
@@ -0,0 +1,10 @@
from sqlalchemy import select
from sqlalchemy.orm import Session
from worblehat.models import Bookcase
def find_bookcase_by_name(sql_session: Session, name: str) -> Bookcase | None:
return sql_session.scalars(
select(Bookcase).where(Bookcase.name == name),
).one_or_none()
src/worblehat/queries/bookcase_item.py
+42
View File
@@ -0,0 +1,42 @@
from sqlalchemy import select
from sqlalchemy.orm import Session
from worblehat.models import BookcaseItem
def find_bookcase_item_by_isbn(sql_session: Session, isbn: str) -> BookcaseItem | None:
return sql_session.scalars(
select(BookcaseItem).where(BookcaseItem.isbn == isbn),
).one_or_none()
def find_bookcase_item_by_name(sql_session: Session, name: str) -> BookcaseItem | None:
return sql_session.scalars(
select(BookcaseItem).where(BookcaseItem.name == name),
).one_or_none()
def search_bookcase_items_by_title(sql_session: Session, text: str) -> list[BookcaseItem]:
return list(
sql_session.scalars(
select(BookcaseItem).where(BookcaseItem.name.ilike(f"%{text}%")),
).all(),
)
def search_bookcase_item_owners(sql_session: Session, text: str) -> list[str]:
return list(
sql_session.scalars(
select(BookcaseItem.owner)
.where(BookcaseItem.owner.ilike(f"%{text}%"))
.distinct(),
).all(),
)
def list_bookcase_items_by_owner(sql_session: Session, owner: str) -> list[BookcaseItem]:
return list(
sql_session.scalars(
select(BookcaseItem).where(BookcaseItem.owner == owner),
).all(),
)
src/worblehat/queries/bookcase_shelf.py
+49
View File
@@ -0,0 +1,49 @@
from sqlalchemy import select
from sqlalchemy.orm import Session
from worblehat.models import Bookcase, BookcaseShelf
def find_bookcase_shelf(
sql_session: Session,
bookcase: Bookcase,
column: int,
row: int,
) -> BookcaseShelf | None:
return sql_session.scalars(
select(BookcaseShelf).where(
BookcaseShelf.bookcase == bookcase,
BookcaseShelf.column == column,
BookcaseShelf.row == row,
),
).one_or_none()
def list_bookcase_shelfs_ordered(sql_session: Session) -> list[BookcaseShelf]:
return list(
sql_session.scalars(
select(BookcaseShelf)
.join(Bookcase)
.order_by(
Bookcase.name,
BookcaseShelf.column,
BookcaseShelf.row,
),
).all(),
)
def list_bookcase_shelf_positions(
sql_session: Session,
bookcase: Bookcase,
column: int | None = None,
row: int | None = None,
) -> list[tuple[int, int]]:
query = select(BookcaseShelf.row, BookcaseShelf.column).where(
BookcaseShelf.bookcase == bookcase,
)
if column is not None:
query = query.where(BookcaseShelf.column == column)
if row is not None:
query = query.where(BookcaseShelf.row == row)
return [tuple(r) for r in sql_session.execute(query)]
src/worblehat/queries/borrowing.py
+73
View File
@@ -0,0 +1,73 @@
from datetime import datetime
from sqlalchemy import select
from sqlalchemy.orm import Session
from worblehat.models import BookcaseItem, BookcaseItemBorrowing
def list_active_borrowings(sql_session: Session) -> list[BookcaseItemBorrowing]:
return list(
sql_session.scalars(
select(BookcaseItemBorrowing)
.where(BookcaseItemBorrowing.delivered.is_(None))
.order_by(BookcaseItemBorrowing.end_time),
).all(),
)
def list_active_borrowings_for_item(
sql_session: Session,
item: BookcaseItem,
) -> list[BookcaseItemBorrowing]:
return list(
sql_session.scalars(
select(BookcaseItemBorrowing)
.where(
BookcaseItemBorrowing.item == item,
BookcaseItemBorrowing.delivered.is_(None),
)
.order_by(BookcaseItemBorrowing.end_time),
).all(),
)
def has_active_borrowing(sql_session: Session, username: str, item: BookcaseItem) -> bool:
return (
sql_session.scalars(
select(BookcaseItemBorrowing).where(
BookcaseItemBorrowing.username == username,
BookcaseItemBorrowing.item == item,
BookcaseItemBorrowing.delivered.is_(None),
),
).one_or_none()
is not None
)
def list_borrowings_for_isbn(sql_session: Session, isbn: str) -> list[BookcaseItemBorrowing]:
return list(
sql_session.scalars(
select(BookcaseItemBorrowing)
.join(
BookcaseItem,
BookcaseItem.uid == BookcaseItemBorrowing.fk_bookcase_item_uid,
)
.where(BookcaseItem.isbn == isbn)
.order_by(BookcaseItemBorrowing.username),
).all(),
)
def list_overdue_borrowings(sql_session: Session) -> list[BookcaseItemBorrowing]:
return list(
sql_session.scalars(
select(BookcaseItemBorrowing)
.join(BookcaseItem)
.where(
BookcaseItemBorrowing.end_time < datetime.now(),
BookcaseItemBorrowing.delivered.is_(None),
)
.order_by(BookcaseItemBorrowing.end_time),
).all(),
)
src/worblehat/queries/borrowing_queue.py
+42
View File
@@ -0,0 +1,42 @@
from sqlalchemy import select
from sqlalchemy.orm import Session
from worblehat.models import BookcaseItem, BookcaseItemBorrowingQueue
def list_all_queue_items(sql_session: Session) -> list[BookcaseItemBorrowingQueue]:
return list(
sql_session.scalars(
select(BookcaseItemBorrowingQueue).order_by(
BookcaseItemBorrowingQueue.entered_queue_time,
),
).all(),
)
def is_in_borrowing_queue(sql_session: Session, username: str, item: BookcaseItem) -> bool:
return (
sql_session.scalars(
select(BookcaseItemBorrowingQueue).where(
BookcaseItemBorrowingQueue.username == username,
BookcaseItemBorrowingQueue.item == item,
),
).one_or_none()
is not None
)
def list_pending_queue_items_for_item(
sql_session: Session,
item: BookcaseItem,
) -> list[BookcaseItemBorrowingQueue]:
return list(
sql_session.scalars(
select(BookcaseItemBorrowingQueue)
.where(
BookcaseItemBorrowingQueue.item == item,
BookcaseItemBorrowingQueue.item_became_available_time.is_(None),
)
.order_by(BookcaseItemBorrowingQueue.entered_queue_time),
).all(),
)
src/worblehat/queries/deadline_daemon.py
+145
View File
@@ -0,0 +1,145 @@
from datetime import datetime, timedelta
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from worblehat.models import (
BookcaseItemBorrowing,
BookcaseItemBorrowingQueue,
DeadlineDaemonLastRunDatetime,
)
def _sql_subtract_date(sql_session: Session, x: datetime, y: timedelta):
if sql_session.bind.dialect.name == "sqlite":
# SQLite does not support timedelta in queries
return func.datetime(x, f"-{y.days} days")
if sql_session.bind.dialect.name == "postgresql":
return x - y
raise NotImplementedError(
f"Unsupported dialect: {sql_session.bind.dialect.name}",
)
def find_last_run(sql_session: Session) -> DeadlineDaemonLastRunDatetime | None:
return sql_session.scalars(
select(DeadlineDaemonLastRunDatetime),
).one_or_none()
def list_close_deadline_borrowings(
sql_session: Session,
day: int,
last_run_datetime: datetime,
current_run_datetime: datetime,
) -> list[BookcaseItemBorrowing]:
return list(
sql_session.scalars(
select(BookcaseItemBorrowing).where(
_sql_subtract_date(
sql_session,
BookcaseItemBorrowing.end_time,
timedelta(days=day),
).between(
last_run_datetime,
current_run_datetime,
),
BookcaseItemBorrowing.delivered.is_(None),
),
).all(),
)
def list_undelivered_overdue_borrowings(
sql_session: Session,
current_run_datetime: datetime,
) -> list[BookcaseItemBorrowing]:
return list(
sql_session.scalars(
select(BookcaseItemBorrowing).where(
BookcaseItemBorrowing.end_time < current_run_datetime,
BookcaseItemBorrowing.delivered.is_(None),
),
).all(),
)
def list_newly_available_queue_items(
sql_session: Session,
last_run_datetime: datetime,
current_run_datetime: datetime,
) -> list[BookcaseItemBorrowingQueue]:
return list(
sql_session.scalars(
select(BookcaseItemBorrowingQueue)
.join(
BookcaseItemBorrowing,
BookcaseItemBorrowing.fk_bookcase_item_uid
== BookcaseItemBorrowingQueue.fk_bookcase_item_uid,
)
.where(
BookcaseItemBorrowingQueue.expired.is_(False),
BookcaseItemBorrowing.delivered.is_not(None),
BookcaseItemBorrowing.delivered.between(
last_run_datetime,
current_run_datetime,
),
)
.order_by(BookcaseItemBorrowingQueue.entered_queue_time)
.group_by(BookcaseItemBorrowingQueue.fk_bookcase_item_uid),
).all(),
)
def list_expiring_queue_positions(
sql_session: Session,
last_run_datetime: datetime,
current_run_datetime: datetime,
) -> list[BookcaseItemBorrowingQueue]:
return list(
sql_session.scalars(
select(BookcaseItemBorrowingQueue)
.join(
BookcaseItemBorrowing,
BookcaseItemBorrowing.fk_bookcase_item_uid
== BookcaseItemBorrowingQueue.fk_bookcase_item_uid,
)
.where(
BookcaseItemBorrowingQueue.item_became_available_time.between(
last_run_datetime,
current_run_datetime,
),
),
).all(),
)
def list_overdue_queue_positions(
sql_session: Session,
queue_position_expiry_days: int,
current_run_datetime: datetime,
) -> list[BookcaseItemBorrowingQueue]:
expiry_cutoff = current_run_datetime - timedelta(days=queue_position_expiry_days)
return list(
sql_session.scalars(
select(BookcaseItemBorrowingQueue).where(
BookcaseItemBorrowingQueue.item_became_available_time < expiry_cutoff,
BookcaseItemBorrowingQueue.expired.is_(False),
),
).all(),
)
def find_next_queue_position(
sql_session: Session,
item_uid: int,
) -> BookcaseItemBorrowingQueue | None:
return sql_session.scalars(
select(BookcaseItemBorrowingQueue)
.where(
BookcaseItemBorrowingQueue.fk_bookcase_item_uid == item_uid,
BookcaseItemBorrowingQueue.item_became_available_time.is_(None),
)
.order_by(BookcaseItemBorrowingQueue.entered_queue_time)
.limit(1),
).one_or_none()
src/worblehat/queries/media_type.py
+10
View File
@@ -0,0 +1,10 @@
from sqlalchemy import select
from sqlalchemy.orm import Session
from worblehat.models import MediaType
def find_media_type_by_name(sql_session: Session, name: str) -> MediaType:
return sql_session.scalars(
select(MediaType).where(MediaType.name.ilike(name)),
).one()