Live

This module handles live events and real-time changes in synchronous contexts.

LiveManager

LiveManager is the primary class for monitoring database changes and events in real-time.

class DbUnify.SQLite3.sync.Live.LiveManager(db_name: str, event_ttl: float, cache_ttl: int = 300)

Bases: object

register_callback(event: str, callback: Callable)

Register a callback for a specific event.

Parameters:

• event (str) – The event to register the callback for.

• callback (Callable) – The callback function to register.

restart()

Restart monitoring the database.

run()

Continuously monitor the database for changes.

start()

Start monitoring the database.

stop()

Stop monitoring the database.

register_callback

Register a callback for a specific event.

from DbUnify.SQLite3.sync.Live import LiveManager

def on_changes(manager: LiveManager, change: Attribute):
    print("Change detected:")
    print(f"Timestamp: {change.timestamp}")
    print(f"Database: {change.db_name}")
    print(f"Table: {change.table_name}")
    print(f"Operation: {change.operation}")
    print(f"Command: {change.command}")
    print(f"Details: {change.details}")

databaseLive = LiveManager("test.db", 0.1)
databaseLive.register_callback('change_detected', on_changes)

restart

Restart monitoring the database.

databaseLive.restart()

run

Continuously monitor the database for changes.

databaseLive.run()

start

Start monitoring the database.

databaseLive.start()

stop

Stop monitoring the database.

databaseLive.stop()

LiveException

Base exception class for LiveManager and its derived exceptions.

DbUnify.SQLite3.sync.Live.LiveException

alias of <module ‘DbUnify.SQLite3.sync.Live.LiveException’ from ‘/home/docs/checkouts/readthedocs.org/user_builds/dbunify/checkouts/latest/DbUnify/SQLite3/sync/Live/LiveException.py’>

DatabaseAccessException

Exception raised for errors accessing the database.

EventHandlingException

Exception raised for errors in event handling.

LogWriteException

Exception raised for errors writing to the log file.

SchemaChangeException

Exception raised for errors detecting schema changes.

Attribute

Attribute represents details of a change event.

class DbUnify.SQLite3.sync.Live.obj.Attribute(timestamp: str, db_name: str, table_name: str, operation: str, command: str, details: dict)

to_dict() → Dict

Convert the Attribute object to a dictionary.

to_dict

Convert the Attribute object to a dictionary.

from DbUnify.SQLite3.sync.Live import Attribute

attr = Attribute(timestamp='2024-08-07T12:34:56', db_name='test.db', table_name='users', operation='INSERT', command='INSERT INTO users (name, email) VALUES (?, ?)', details={'name': 'John Doe', 'email': 'john.doe@example.com'})
attr_dict = attr.to_dict()
print(attr_dict)

Examples

Live Event Listener Example

This example demonstrates how to set up a LiveManager instance to listen for database changes and schema changes.

from DbUnify.SQLite3.sync.Live import LiveManager, Attribute

def on_changes(manager: LiveManager, change: Attribute):
    print("Change detected:")
    print(f"Timestamp: {change.timestamp}")
    print(f"Database: {change.db_name}")
    print(f"Table: {change.table_name}")
    print(f"Operation: {change.operation}")
    print(f"Command: {change.command}")
    print(f"Details: {change.details}")

def on_schema_changes(manager: LiveManager, change: Attribute):
    print("Schema change detected:")
    print(f"Timestamp: {change.timestamp}")
    print(f"Database: {change.db_name}")
    print(f"Table: {change.table_name}")
    print(f"Operation: {change.operation}")
    print(f"Command: {change.command}")
    print(f"Details: {change.details}")

if __name__ == "__main__":
    databaseLive = LiveManager("test.db", 0.1)
    databaseLive.register_callback('change_detected', on_changes)
    databaseLive.register_callback('schema_change_detected', on_schema_changes)
    databaseLive.run()