db_handler Module

Database layer for ttmp32gme with thread-safe SQLite access.

DBHandler Class

CRITICAL: All database operations MUST go through DBHandler methods.

result = db.fetchone("SELECT ...")  # ✓
cursor = db.cursor()  # ✗ Never use raw cursors

Core Methods

connect() - Establish database connection
execute(query, params) - Execute SQL
fetchone(query, params) - Fetch one result
fetchall(query, params) - Fetch all results
commit() - Commit transaction

Album Operations

get_album(oid) - Get album by OID
get_all_albums() - List all albums
create_album(data) - Create new album
update_album(oid, updates) - Update album
delete_album(oid) - Delete album

Track Operations

get_tracks(album_oid) - Get tracks for album
create_track(data) - Create track
update_track(track_id, updates) - Update track

Config Operations

get_config() - Get all config
get_config_value(key) - Get specific value
set_config_value(key, value) - Set value

Pydantic Models

AlbumUpdateModel: Validates album updates (oid, title, artist, tracks, mode)

ConfigUpdateModel: Validates config (host, port, audio_format, library_path)

LibraryActionModel: Validates actions (uid, tiptoi_dir)

AlbumMetadataModel: Validates metadata from audio files

Usage

from ttmp32gme.db_handler import DBHandler, AlbumUpdateModel
from pydantic import ValidationError

db = DBHandler("/path/to/db.sqlite")

# Validate and update
try:
    validated = AlbumUpdateModel(**data)
    db.update_album(validated.oid, validated.model_dump(exclude_none=True))
except ValidationError as e:
    print(f"Error: {e}")

ttmp32gme.db_handler.convert_str_to_int(v)[source]

Convert string to integer if needed.

Parameters:: v (Any) – Value to convert (can be str, int, or None)
Return type:: Optional[int]
Returns:: Integer value or None
Raises:: ValueError – If conversion fails

ttmp32gme.db_handler.trim_optional_str(v)[source]

Trim optional string fields.

Parameters:: v (Any) – String value or None or any other type
Return type:: Optional[str]
Returns:: Trimmed string or None

ttmp32gme.db_handler.validate_non_empty_str(v, field_name='field')[source]

Validate and trim non-empty string fields.

Parameters:

v (Any) – String value to validate
field_name (str) – Name of the field (for error messages)

Return type:

str

Returns:

Trimmed string

Raises:

ValueError – If string is empty or whitespace only
TypeError – If value is not a string

class ttmp32gme.db_handler.AlbumUpdateModel(**data)[source]

Bases: BaseModel

Validates album update data from frontend.

oid: Optional[int]

uid: Optional[int]

old_oid: Optional[int]

album_title: Optional[str]

album_artist: Optional[str]

num_tracks: Optional[int]

player_mode: Optional[str]

cover: Optional[str]

model_config: ClassVar[ConfigDict] = {'extra': 'allow'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod convert_to_int(v)[source]

Convert string OIDs to integers.

Return type:: Optional[int]

class ttmp32gme.db_handler.ConfigUpdateModel(**data)[source]

Bases: BaseModel

Validates configuration update data from frontend.

host: Optional[str]

port: Optional[int]

open_browser: Optional[bool]

audio_format: Optional[str]

pen_language: Optional[str]

library_path: Optional[str]

log_level: Optional[str]

model_config: ClassVar[ConfigDict] = {'extra': 'allow'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class ttmp32gme.db_handler.LibraryActionModel(**data)[source]

Bases: BaseModel

Validates library action data (delete, cleanup, make_gme, etc.).

uid: int

tiptoi_dir: Optional[str]

model_config: ClassVar[ConfigDict] = {'extra': 'allow'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod convert_uid_to_int(v)[source]

Convert string UID to integer.

Return type:: Optional[int]

class ttmp32gme.db_handler.AlbumMetadataModel(**data)[source]

Bases: BaseModel

Validates album-level metadata extracted from audio files.

oid: int

album_title: str

album_artist: Optional[str]

album_year: Optional[str]

num_tracks: int

picture_filename: Optional[str]

path: str

classmethod validate_album_title(v)[source]

Ensure album title is not empty.

Return type:: str

classmethod validate_album_artist(v)[source]

Trim album artist.

Return type:: Optional[str]

classmethod validate_year(v)[source]

Validate year format.

Return type:: Optional[str]

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class ttmp32gme.db_handler.TrackMetadataModel(**data)[source]

Bases: BaseModel

Validates track-level metadata extracted from audio files.

parent_oid: int

album: Optional[str]

artist: Optional[str]

disc: Optional[str]

duration: int

genre: Optional[str]

lyrics: Optional[str]

title: str

track: int

filename: str

classmethod validate_title(v)[source]

Ensure track title is not empty.

Return type:: str

classmethod trim_string_fields(v)[source]

Trim string fields.

Return type:: Optional[str]

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class ttmp32gme.db_handler.DBHandler(db_path)[source]

Bases: object

VALID_TABLES = {'config', 'gme_library', 'script_codes', 'tracks'}

__init__(db_path)[source]

property gme_library_columns: List[str]

property conn

close()[source]

initialize()[source]

execute_context(query, params=())[source]

Execute a database query with automatic locking and cursor cleanup.

Use as a context manager to automatically handle thread-safe locking and cursor cleanup.

Parameters:

query (str) – SQL query to execute
params (Tuple[Any, ...]) – Query parameters tuple

Yields:

Database cursor with query results

Return type:

Generator[Cursor, None, None]

Example

with db.execute_context(query, params) as cursor:: result = cursor.fetchone()

execute_and_commit(query, params=())[source]

Execute a single database query and commit the changes.

Parameters:

query (str) – SQL query to execute
params (Tuple[Any, ...]) – Query parameters tuple

Return type:

None

Returns:

None

fetchall(query, params=())[source]

Execute query and fetch all results with thread-safe locking.

Parameters:

query (str) – SQL query to execute
params (Tuple[Any, ...]) – Query parameters tuple

Return type:

List[Row]

Returns:

List of result rows

fetchone(query, params=())[source]

Execute query and fetch one result with thread-safe locking.

Parameters:

query (str) – SQL query to execute
params (Tuple[Any, ...]) – Query parameters tuple

Return type:

Optional[Row]

Returns:

Single result row or None

commit()[source]: Commit database changes with thread-safe locking.

get_config()[source]

Get configuration parameters.

Return type:: Dict[str, str]
Returns:: Configuration dictionary

write_to_database(table, data)[source]

Write data to database table.

Parameters:

table (str) – Table name
data (Dict[str, Any]) – Data dictionary
connection – Database connection

Raises:

ValueError – If table name or field names are invalid

get_config_value(param)[source]

Get a specific configuration value.

Parameters:: param (str) – Configuration parameter name
Return type:: Optional[str]
Returns:: Configuration value or None

set_config_value(param, value)[source]

Set a specific configuration value.

Parameters:

param (str) – Configuration parameter name
value (str) – Configuration value

Return type:

None

insert_or_replace_config(param, value)[source]

Insert or replace a configuration parameter.

Parameters:

param (str) – Configuration parameter name
value (str) – Configuration value

Return type:

None

oid_exist(oid)[source]

Check if an OID exists in the database.

Parameters:: oid (int) – OID to check
Return type:: bool
Returns:: True if OID exists

new_oid()[source]

Generate a new unique OID.

Args:

Return type:: int
Returns:: New OID

get_tracks(album)[source]

Get all tracks for an album.

Parameters:: album (Dict[str, Any]) – Album dictionary
Return type:: Dict[int, Dict[str, Any]]
Returns:: Dictionary of tracks indexed by track number

update_table_entry(table, keyname, search_keys, data)[source]

Update a table entry.

Parameters:

table (str) – Table name
keyname (str) – Key column name with condition (e.g., ‘oid=?’)
search_keys (List[Any]) – Values for the key condition
data (Dict[str, Any]) – Data to update

Return type:

bool

Returns:

True if successful

Raises:

ValueError – If table name or field names are invalid

create_library_entry(album_list, library_path)[source]

Create a new library entry from uploaded files.

Parameters:

album_list (List[Dict[str, Any]]) – List of albums with file paths
library_path (Path) – Library path

Return type:

bool

Returns:

True if successful

db_row_to_album(row)[source]

Convert a database row to a complete album dictionary including track info.

Parameters:

columns – list of column names
row (Row) – Database row

Return type:

Dict[str, Any]

Returns:

Album including tracks as dictionary

get_album(oid)[source]

Get album by OID.

Parameters:: oid (int) – Album OID
Return type:: Optional[Dict[str, Any]]
Returns:: Album dictionary or None

get_album_list()[source]

Get list of all albums.

Args:

Return type:: List[Dict[str, Any]]
Returns:: List of album dictionaries

get_gme_file_info(oid)[source]

Get GME file path and filename for an album.

Parameters:: oid (int) – Album OID
Return type:: Optional[Tuple[str, str]]
Returns:: Tuple of (album_path, gme_filename) or None if album not found

update_tracks(tracks, parent_oid, new_parent_oid)[source]

Update tracks in the database using UPDATE statements.

Parameters:

tracks (List[Dict[str, Any]]) – List of track dictionaries from frontend (with id, track, title)
parent_oid (int) – Original parent OID
new_parent_oid (int) – New parent OID

Return type:

bool

Returns:

True if successful

update_album(album_data)[source]

Update an existing album.

Parameters:: album_data (Dict[str, Any]) – Album data to update
Return type:: int
Returns:: Album OID

delete_album(uid)[source]

Delete an album.

Parameters:: uid (int) – Album OID
Return type:: int
Returns:: Deleted album OID

delete_album_tracks(oid)[source]

Delete all tracks of an album.

Parameters:: oid (int) – Album OID
Return type:: int
Returns:: Album OID

cleanup_album(uid)[source]

Clean up an album directory.

Parameters:: uid (int) – Album OID
Return type:: int
Returns:: Album OID

replace_cover(uid, filename, file_data)[source]

Replace album cover image.

Parameters:

uid (int) – Album OID
filename (str) – New cover filename
file_data (bytes) – Image data

Return type:

int

Returns:

Album OID

change_library_path(old_path, new_path)[source]

Change the library path in the configuration.

Parameters:

old_path (str) – Current library path
new_path (Path) – New library path

Return type:

bool

Returns:

True if successful

update_db()[source]

Update database schema to latest version.

Args:

Return type:: bool
Returns:: True if update successful

get_oid_cache()[source]

Get the OID cache directory from library path in database.

Return type:: Path
Returns:: Path to OID cache directory

create_oid_images_zip()[source]

Create a ZIP file containing all OID images from the cache.

Return type:: Optional[BytesIO]
Returns:: BytesIO object containing the ZIP file, or None if no images found

ttmp32gme.db_handler.extract_tracks_from_album(album)[source]

Extract track dictionaries from an album dictionary.

Parameters:: album (Dict[str, Any]) – Album dictionary
Return type:: Tuple[List[Dict[str, Any]], Dict[str, Any]]
Returns:: Tuple of (list of track dictionaries, modified album dictionary)

ttmp32gme.db_handler.get_cover_filename(mimetype, picture_data)[source]

Generate a filename for cover image.

Parameters:

mimetype (Optional[str]) – Image MIME type
picture_data (bytes) – Image data

Return type:

Optional[str]

Returns:

Cover filename or None