Implement NoteColoursDTO

This commit is contained in:
Keith Edmunds 2025-04-12 11:15:54 +01:00
parent 38c49b32d7
commit ac685426d9
2 changed files with 68 additions and 3 deletions

View File

@ -95,6 +95,19 @@ class PlaydatesDTO(TrackDTO):
lastplayed: dt.datetime lastplayed: dt.datetime
@dataclass
class NoteColoursDTO:
notecolour_id: int
substring: str
colour: str
enabled: bool = True
foreground: str | None = None
is_regex: bool = False
is_casesensitive: bool = False
order: int | None = None
strip_substring: bool = True
class ApplicationError(Exception): class ApplicationError(Exception):
""" """
Custom exception Custom exception

View File

@ -5,6 +5,8 @@ import re
# PyQt imports # PyQt imports
# Third party imports # Third party imports
from dogpile.cache import make_region
from dogpile.cache.api import NO_VALUE
from sqlalchemy import ( from sqlalchemy import (
delete, delete,
func, func,
@ -19,6 +21,7 @@ from sqlalchemy.sql.elements import BinaryExpression, ColumnElement
from classes import ( from classes import (
ApplicationError, ApplicationError,
Filter, Filter,
NoteColoursDTO,
PlaydatesDTO, PlaydatesDTO,
PlaylistDTO, PlaylistDTO,
PlaylistRowDTO, PlaylistRowDTO,
@ -39,6 +42,13 @@ from models import (
) )
# Configure the dogpile cache region
cache_region = make_region().configure(
'dogpile.cache.memory', # Use in-memory caching for now (switch to Redis if needed)
expiration_time=600 # Cache expires after 10 minutes
)
# Helper functions # Helper functions
@log_call @log_call
def _remove_substring_case_insensitive(parent_string: str, substring: str) -> str: def _remove_substring_case_insensitive(parent_string: str, substring: str) -> str:
@ -68,13 +78,53 @@ def _remove_substring_case_insensitive(parent_string: str, substring: str) -> st
# Notecolour functions # Notecolour functions
def _get_colour_record(text: str) -> tuple[NoteColours | None, str]: def _all_notecolours(session: Session) -> list[NoteColoursDTO]:
"""
Return all notecolour records
"""
cache_key = "note_colours_all"
cached_result = cache_region.get(cache_key)
if cached_result is not NO_VALUE:
return cached_result
# Query the database
records = session.scalars(
select(NoteColours)
.where(
NoteColours.enabled.is_(True),
)
.order_by(NoteColours.order)
).all()
results: list[NoteColoursDTO] = []
for record in records:
result = NoteColoursDTO(
notecolour_id=record.id,
substring=record.substring,
colour=record.colour,
enabled=record.enabled,
foreground=record.foreground,
is_regex=record.is_regex,
is_casesensitive=record.is_casesensitive,
order=record.order,
strip_substring=record.strip_substring,
)
results.append(result)
cache_region.set(cache_key, results)
return results
def _get_colour_record(text: str) -> tuple[NoteColoursDTO | None, str]:
""" """
Parse text and return first matching colour record or None Parse text and return first matching colour record or None
""" """
with db.Session() as session: with db.Session() as session:
for rec in NoteColours.get_all(session): for rec in _all_notecolours(session):
if rec.is_regex: if rec.is_regex:
flags = re.UNICODE flags = re.UNICODE
if not rec.is_casesensitive: if not rec.is_casesensitive:
@ -751,7 +801,7 @@ def copy_playlist(src_id: int, dst_id: int) -> None:
with db.Session() as session: with db.Session() as session:
src_rows = session.scalars( src_rows = session.scalars(
select(PlaylistRows).filter(PlaylistRows.playlist_id == src_id) select(PlaylistRows).where(PlaylistRows.playlist_id == src_id)
).all() ).all()
for plr in src_rows: for plr in src_rows:
@ -763,6 +813,8 @@ def copy_playlist(src_id: int, dst_id: int) -> None:
track_id=plr.track_id, track_id=plr.track_id,
) )
session.commit()
def playlist_row_count(playlist_id: int) -> int: def playlist_row_count(playlist_id: int) -> int:
""" """