Database

SQLite persistence APIs for monitoring sessions, foreground activity, and process events.

Schema models

---

source

ProcessEvent

def ProcessEvent(
    *args, **kwargs
):

A recorded process start or stop event.

class ProcessEvent:
    "A recorded process start or stop event."

    session_id: str  # Monitoring session containing the event
    pid: int         # Operating-system process identifier
    app: str         # Process or application name
    event_type: str  # Event type: `start` or `stop`
    timestamp: str   # ISO-formatted event timestamp
    id: int | None = None  # Database-generated primary key

---

source

Session

def Session(
    *args, **kwargs
):

A single continuous monitoring session.

class Session:
    "A single continuous monitoring session."

    session_id: str       # Unique session identifier
    start_time: str       # ISO-formatted session start time
    end_time: str | None  # ISO-formatted end time; `None` while open
    status: str           # Session state: `open` or `closed`
    end_reason: str | None  # Reason the session ended

---

source

ForegroundEvent

def ForegroundEvent(
    *args, **kwargs
):

An interval during which an application or idle state was in the foreground.

class ForegroundEvent:
    "An interval during which an application or idle state was in the foreground."

    session_id: str       # Monitoring session containing the interval
    app: str              # Foreground application name
    pid: int              # Foreground process identifier
    window_title: str     # Foreground window title
    start_time: str       # ISO-formatted interval start time
    end_time: str         # ISO-formatted interval end time
    hwnd: int             # Windows window handle
    is_idle: bool         # Whether the interval represents user idle time
    id: int | None = None  # Database-generated primary key

Database connection

---

source

init_db

def init_db(
    path:NoneType=None, # database path for the application
):

Open the database connection and create tables. Uses DATABASE_PATH if path is None.

from tempfile import mkdtemp
from pathlib import Path
init_db(Path(mkdtemp()) / "snooper_test.db")

<Database <apsw.Connection "/tmp/tmp6k5aw5n7/snooper_test.db" at 0x7efcd45c82c0>>

Process events

---

source

log_process_event

def log_process_event(
    session_id:str, # Session receiving the event
    pid:int, # Operating-system process identifier
    app:str, # Process or application name
    event_type:str, # Event type: `start` or `stop`
    timestamp:str, # ISO-formatted event timestamp
):

Record a process start or stop event.

---

source

get_processes

def get_processes(
    session_id:str, # Session whose completed process intervals are requested
)->list[dict]:

Return paired process start/stop intervals for a session.

Monitoring sessions

---

source

start_session

def start_session(
    session_id:str, # Unique identifier for the new session
    start_time:str, # ISO-formatted session start time
):

Create and return a new open monitoring session.

---

source

end_session

def end_session(
    session_id:str, # Session to close
    end_time:str, # ISO-formatted session end time
    end_reason:str='NA', # Reason the session ended
):

Close a monitoring session and record why it ended.

---

source

get_last_completed_session

def get_last_completed_session()->dict:

Return the most recently completed session, or an empty dictionary.

---

source

close_stale_open_sessions

def close_stale_open_sessions(
    end_time:str, # Timestamp used to close stale sessions
)->list[str]:

Close sessions left open before startup and return their identifiers.

---

source

get_session_by_session_id

def get_session_by_session_id(
    session_id:str, # Identifier of the requested session
)->dict:

Return a session by ID, using the current time for an open session’s end.

Foreground events

---

source

log_fg_app_events

def log_fg_app_events(
    session_id:str, # Session receiving the interval
    app:str, # Foreground application name
    pid:int, # Foreground process identifier
    window_title:str, # Foreground window title
    start_time:str, # ISO-formatted interval start time
    end_time:str, # ISO-formatted interval end time
    hwnd:int, # Windows window handle
    is_idle:bool, # Whether this was an idle interval
):

Record a completed foreground-window or idle interval.

---

source

get_foreground_events

def get_foreground_events(
    session_id:str, # Session whose foreground intervals are requested
)->list[dict]:

Return a session’s foreground intervals in chronological order.

Data retention

---

source

delete_old_data

def delete_old_data(
    now_ts:str | datetime.datetime | None=None, # Retention reference time; defaults to now
)->str:

Delete data belonging to closed sessions older than the retention period.