savedata.py

lunaengine/storage/savedata.py

A powerful, table‑based save‑data system for LunaEngine.
- Multiple tables with typed columns
- Insert, update, delete, select with filtering
- Primary key indexing for fast lookups
- Auto‑increment for integer primary keys
- Upsert (insert or update)
- Bulk update by primary key via `updates()` dict
- Advanced query support: ordering, searching (contains, startswith, endswith, exact, case‑sensitive)
- Custom binary file format with compression (zlib) and optional encryption
- Export/Import to JSON for machine migration
- Integrates with MachineEncryption for HWID‑based XOR encryption

Usage:
   data = Savedata("game.sav")
   users = data.create_table("users", ["id", "name", "score"],
                             primary_key="id", auto_increment=True)
   users.insert(name="Alice", score=100)       # id auto‑assigned
   users.insert(id=5, name="Bob", score=200)   # manual ID
   data.save()

   # Update by primary key
   users.update_by_primary_key(1, score=150)

   # Upsert
   users.upsert(id=2, name="Charlie", score=300)

   # Bulk update by primary keys
   users.updates({
       1: {"name": "Alicia", "score": 110},
       5: {"score": 250}
   })

   # Advanced select with query builder
   results = users.query() \
       .where(lambda r: r["score"] > 50) \
       .order_by("score", desc=True) \
       .search("al", columns=["name"], mode="contains") \
       .execute()

   # Export to JSON for migration
   data.export_to_json("backup.json")
   # On another machine:
   data.import_from_json("backup.json")

class SavedataError

Description

Base exception for Savedata errors.

Methods

No methods defined.

class Table

Description

A table in the savedata: stores rows as dictionaries with column names as keys.
Supports primary key indexing for O(1) lookups and optional auto‑increment.

Methods
def __init__(self: Any, name: str, columns: List[str], primary_key: Optional[str] = None, auto_increment: bool = False) -> Any
Args:
   name: Table name.
   columns: Ordered list of column names.
   primary_key: Column name to use as primary key (must be in columns).
   auto_increment: If True and primary_key is an integer column, automatically
                   assign the next ID on insert when the key is omitted.
def insert(self: Any, **kwargs: dict) -> int
Insert a row. Keyword arguments must match the table's columns.
If auto_increment is enabled and the primary key is not provided,
the next ID is assigned automatically.
Returns the row index.
def update(self: Any, where: Callable[[Dict], bool], **kwargs: dict) -> int
Update rows that match the where predicate with the given kwargs.
Returns the number of updated rows.
def update_where(self: Any, where: Callable[[Dict], bool], updates: Dict[str, Any]) -> int
Update rows that match the where predicate with the given updates dict.
Returns the number of updated rows.
def update_by_primary_key(self: Any, key: Any, **kwargs: dict) -> int
Update the row with the given primary key using the provided kwargs.
Returns 1 if updated, 0 if not found.
def updates(self: Any, updates_dict: Dict[Any, Union[Any, Dict[str, Any]]]) -> int
Update multiple rows by primary key.

Args:
   updates_dict: A dictionary where keys are primary key values,
                 and values are either:
                   - a dict mapping column names to new values
                   - a scalar value (shorthand):
                       - If the table has exactly one non‑primary‑key column,
                         that column is updated with this value.
                       - Otherwise, if a column named 'value' exists,
                         that column is updated.
                       - Else, raises ValueError.

Returns:
   The number of rows that were updated (i.e., keys that existed).
def upsert(self: Any, **kwargs: dict) -> int
Insert a new row, or update the existing one if the primary key already exists.
Returns the row index (for insert) or 1 (for update).
Raises ValueError if primary key is missing when auto_increment is off.
def delete(self: Any, where: Callable[[Dict], bool]) -> int
Delete rows matching the predicate.
Returns the number of deleted rows.
def delete_by_primary_key(self: Any, key: Any) -> int
Delete the row with the given primary key.
Returns 1 if deleted, 0 if not found.
def select(self: Any, where: Optional[Callable[[Dict], bool]] = None, columns: Optional[List[str]] = None, order_by: Optional[Union[str, List[str]]] = None, order_desc: Union[bool, List[bool]] = False, search: Optional[str] = None, search_columns: Optional[List[str]] = None, search_mode: str = 'contains', case_sensitive: bool = False) -> List[Dict]
Select rows with optional filtering, ordering, and searching.

Args:
   where: Predicate function that takes a row dict and returns True to include.
   columns: List of column names to return. If None, returns all columns.
   order_by: Column name(s) to sort by. Can be a string or list of strings.
   order_desc: If True, sort descending. If a list, must match length of order_by.
   search: Search term to filter rows.
   search_columns: Columns to search in. If None, searches all columns.
   search_mode: 'contains', 'startswith', 'endswith', or 'exact'.
   case_sensitive: If False, performs case‑insensitive search.

Returns:
   List of row dictionaries (or partial dicts if columns specified).
def find_one(self: Any, where: Optional[Callable[[Dict], bool]] = None) -> Optional[Dict]
Return the first row that matches the predicate, or None if not found.
def exists(self: Any, where: Callable[[Dict], bool]) -> bool
Return True if any row matches the predicate.
def count(self: Any, where: Optional[Callable[[Dict], bool]] = None) -> int
Return the number of rows (optionally filtered by predicate).
def get_next_id(self: Any) -> int
Return the next auto‑increment ID (only valid if auto_increment is True
and the primary key is an integer column).
def _apply_search(self: Any, rows: List[Dict], term: str, search_columns: Optional[List[str]], mode: str, case_sensitive: bool) -> List[Dict]
Filter rows based on search term.
def _sort_rows(self: Any, rows: List[Dict], order_by: Union[str, List[str]], order_desc: Union[bool, List[bool]]) -> List[Dict]
Sort rows by one or more columns with specified directions.
def get_by_primary_key(self: Any, key: Any) -> Optional[Dict]
Retrieve a row by its primary key value, or None if not found.
def clear(self: Any) -> None
Remove all rows.
def query(self: Any) -> 'Query'
Return a new Query builder for this table.
Example:
   rows = table.query() \
       .where(lambda r: r["score"] > 100) \
       .order_by("score", desc=True) \
       .search("al", columns=["name"], mode="contains") \
       .execute()
def _find_primary_key_for_idx(self: Any, idx: int) -> Any
No documentation
def _rebuild_index(self: Any) -> None
No documentation
def _update_next_id(self: Any) -> None
Recalculate the next auto‑increment ID from existing rows.
def __len__(self: Any) -> int
No documentation
def __repr__(self: Any) -> str
No documentation

class Query

Description

Fluent query builder for Table.

Methods
def __init__(self: Any, table: Table) -> Any
No documentation
def where(self: Any, predicate: Callable[[Dict], bool]) -> 'Query'
No documentation
def order_by(self: Any, *columns: tuple) -> 'Query'
No documentation
def execute(self: Any, columns: Optional[List[str]] = None) -> List[Dict]
No documentation

class Savedata

Description

Main savedata manager. Holds multiple tables and provides load/save functionality.
Supports compression (zlib) and optional encryption via MachineEncryption.
File format:
   - Magic bytes: "LUNASD" (7 bytes)
   - Version: uint16
   - Flags: uint8 (bit 0: compressed, bit 1: encrypted)
   - Table count: uint32
   - For each table:
       - Name length (uint16) + name string (UTF-8)
       - Column count (uint16)
       - For each column: name length (uint16) + name string
       - Primary key index (int16, -1 if none)
       - Auto‑increment flag (uint8)
       - Row count (uint32)
       - For each row:
           - For each column:
               - Type tag (uint8): 0=None, 1=int, 2=float, 3=str, 4=bool, 5=bytes
               - Data (variable)

Attributes
MAGIC: Any = b'LUNASD'
VERSION: Any = 2
Methods
def __init__(self: Any, filepath: Optional[Union[str, Path]] = None, encryption_key: Optional[str] = None) -> Any
Args:
   filepath: Path to the savedata file. If provided and exists, loads automatically.
   encryption_key: Key for XOR encryption (if None, uses default MachineEncryption HWID).
def create_table(self: Any, name: str, columns: List[str], primary_key: Optional[str] = None, auto_increment: bool = False) -> Table
No documentation
def drop_table(self: Any, name: str) -> None
No documentation
def table(self: Any, name: str) -> Optional[Table]
No documentation
def clear(self: Any) -> None
No documentation
def save(self: Any, filepath: Optional[Union[str, Path]] = None, encryption_key: Optional[str] = None, compress: bool = True) -> None
No documentation
def load(self: Any, filepath: Optional[Union[str, Path]] = None, encryption_key: Optional[str] = None) -> None
No documentation
def _serialize(self: Any) -> bytes
No documentation
def _deserialize(self: Any, data: bytes) -> None
No documentation
def _append_value(self: Any, parts: List[bytes], value: Any) -> None
No documentation
def _read_value(self: Any, data: bytes, pos: int) -> Tuple[Any, int]
No documentation
def _xor_encrypt(self: Any, data: bytes, key: str) -> bytes
No documentation
def export_to_dict(self: Any) -> Dict[str, List[Dict]]
Export all tables to a dictionary: {table_name: [row_dict, ...]}.
Useful for serialization to JSON or migration.
def import_from_dict(self: Any, data: Dict[str, List[Dict]]) -> None
Clear all tables and import data from a dictionary.
Expects the same structure as export_to_dict.
Columns and primary keys are recreated from the first row.
Auto‑increment is disabled on imported tables (you can re‑enable manually).
def export_to_json(self: Any, filepath: Union[str, Path], indent: int = 2) -> None
Export all tables to a JSON file (unencrypted, no compression).
def import_from_json(self: Any, filepath: Union[str, Path]) -> None
Import tables from a JSON file (clears existing data).
def migrate_to_plain(self: Any, output_path: Union[str, Path]) -> None
Save the current data without encryption and without compression.
Useful for creating a portable backup.
def migrate_from_plain(self: Any, input_path: Union[str, Path], new_key: Optional[str] = None) -> None
Load a plain (unencrypted) file and re-save it with the current encryption key.
If new_key is provided, use that key for subsequent saves.
def __repr__(self: Any) -> str
No documentation

Global Functions

def load_savedata(filepath: Union[str, Path], encryption_key: Optional[str] = None) -> Savedata

No documentation

def save_savedata(data: Savedata, filepath: Union[str, Path], encryption_key: Optional[str] = None, compress: bool = True) -> None

No documentation

Back to Storage Module