Skip to content

session_manager

Chat session management utilities.

This module provides the ChatSessionManager class, which offers a high level interface for creating, selecting, and managing chat sessions backed by a pluggable database implementation.

The manager abstracts away the underlying database type through the BaseChatDatabase interface and ChatDatabaseFactory, allowing clients to work with chat sessions using a simple API

ChatSessionManager

Manage chat sessions and their message histories using a backing database.

This class provides a thin abstraction around a BaseChatDatabase implementation so that callers can interact with chat sessions using a simple, stateful interface. A "current" session is tracked via current_session_id, and most operations act on this current session.

The underlying database implementation is created using ChatDatabaseFactory. If no database_config is supplied, a default SQLiteConfig is used. If the provided configuration does not specify a session_id, a new UUID based session ID is created automatically.

Attributes:

Name Type Description
chat_db BaseChatDatabase

Concrete implementation of BaseChatDatabase used to store and retrieve chat data.
current_session_id

Identifier of the active chat session. This is either taken from the provided DatabaseConfig or generated using create_session_id.

Methods:

Name Description
create_session_id

Generates a new unique identifier for a chat session.
select_session

Sets the given session as the current session and returns its message history.
load_history

Loads the message history for the current session from the database.
add_message

Adds a message entry to the current session in the database.
list_sessions

Lists all sessions known to the database, including their last updated time.
delete_session

Deletes the current session from the database.
Source code in src/mada/core/database/session_manager.py 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
class ChatSessionManager:
    """
    Manage chat sessions and their message histories using a backing database.

    This class provides a thin abstraction around a `BaseChatDatabase`
    implementation so that callers can interact with chat sessions using a
    simple, stateful interface. A "current" session is tracked via
    `current_session_id`, and most operations act on this current session.

    The underlying database implementation is created using
    `ChatDatabaseFactory`. If no `database_config` is supplied, a default
    `SQLiteConfig` is used. If the provided configuration does not specify
    a `session_id`, a new UUID based session ID is created automatically.

    Attributes:
        chat_db: Concrete implementation of `BaseChatDatabase` used to
            store and retrieve chat data.
        current_session_id: Identifier of the active chat session. This is
            either taken from the provided `DatabaseConfig` or generated
            using `create_session_id`.

    Methods:
        create_session_id: Generates a new unique identifier for a chat
            session.
        select_session: Sets the given session as the current session and
            returns its message history.
        load_history: Loads the message history for the current session from
            the database.
        add_message: Adds a message entry to the current session in the
            database.
        list_sessions: Lists all sessions known to the database, including
            their last updated time.
        delete_session: Deletes the current session from the database.
    """

    def __init__(self, database_config: DatabaseConfig = None):
        """
        Constructor for the ChatSessionManager.

        Args:
            database_config: The database configuration object to
                use for connecting to the database. If no database
                configuration is provided, default to SQLite.
        """
        self.chat_db: BaseChatDatabase = self._connect_to_db(database_config)

        # Store the current session ID
        if database_config is not None and database_config.session_id is not None:
            self.current_session_id = database_config.session_id
        else:
            self.current_session_id = self.create_session_id()

    def _connect_to_db(self, database_config: DatabaseConfig) -> BaseChatDatabase:
        """
        Given a database configuration, connect to the database.

        Args:
            database_config: The database configuration object to
                use for connecting to the database. If no database
                configuration is provided, default to SQLite.

        Returns:
            A connected database instance.
        """
        if database_config is None:
            database_config = SQLiteConfig()
        chat_db_factory = ChatDatabaseFactory()
        chat_db = chat_db_factory.create(database_config.type, database_config)
        return chat_db

    def create_session_id(self) -> str:
        """
        Create a new chat session ID.

        Returns:
            A new session ID.
        """
        return str(uuid.uuid4())

    def create_new_session(self, session_id: str = None):
        """
        Save an empty session to the database.

        Args:
            session_id (str): Specific session ID to assign to the new session.
        """
        if not session_id:
            session_id = self.current_session_id
        self.chat_db.create_session(session_id)

    def select_session(self, session_id: str) -> List[Dict]:
        """
        Set the given session as the current session and return its history.

        Args:
            session_id (str): The ID of the session to select.

        Returns:
            The session chat history.
        """
        self.current_session_id = session_id
        return self.load_history()

    def load_history(self) -> List[Dict]:
        """
        Load the message history from the current session.

        Returns:
            A list of messages for the given session.
        """
        return self.chat_db.load_session(self.current_session_id) or []

    def add_message(self, role: str, message: str):
        """
        Add a message to the database.

        Args:
            role (str): The role (user or assistant) to designate who wrote the message
            message (str): The message contents
        """
        self.chat_db.add_message(self.current_session_id, role, message)

    def list_sessions(self) -> List[Tuple[str, datetime]]:
        """
        List all sessions currently stored in the database.

        Returns:
            A list of sessions where each session has (session ID, last updated).
        """
        return self.chat_db.list_sessions()

    def delete_session(self, session_id: str = None):
        """
        Given a session ID, delete the session from the database.

        If no session ID is provided, the currently tracked session is deleted.

        Args:
            session_id (str): Specific session ID to delete from the database.
        """
        if session_id is None:
            session_id = self.current_session_id
        self.chat_db.delete_session(session_id)

    def delete_all_sessions(self, confirm: bool = True):
        """
        Delete all sessions from the database.

        Warning:
            This operation is irreversible and will delete all chat history.

        Args:
            confirm (bool): If True, prompt the user for confirmation before
                deleting. If False, delete without prompting.
        """
        self.chat_db.flush_database(confirm=confirm)
        self.current_session_id = None

__init__(database_config=None)

Constructor for the ChatSessionManager.

Parameters:

Name Type Description Default
database_config DatabaseConfig

The database configuration object to use for connecting to the database. If no database configuration is provided, default to SQLite.

 None
Source code in src/mada/core/database/session_manager.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def __init__(self, database_config: DatabaseConfig = None):
    """
    Constructor for the ChatSessionManager.

    Args:
        database_config: The database configuration object to
            use for connecting to the database. If no database
            configuration is provided, default to SQLite.
    """
    self.chat_db: BaseChatDatabase = self._connect_to_db(database_config)

    # Store the current session ID
    if database_config is not None and database_config.session_id is not None:
        self.current_session_id = database_config.session_id
    else:
        self.current_session_id = self.create_session_id()

add_message(role, message)

Add a message to the database.

Parameters:

Name Type Description Default
role str

The role (user or assistant) to designate who wrote the message

 required
message str

The message contents

 required
Source code in src/mada/core/database/session_manager.py 
136
137
138
139
140
141
142
143
144
def add_message(self, role: str, message: str):
    """
    Add a message to the database.

    Args:
        role (str): The role (user or assistant) to designate who wrote the message
        message (str): The message contents
    """
    self.chat_db.add_message(self.current_session_id, role, message)

create_new_session(session_id=None)

Save an empty session to the database.

Parameters:

Name Type Description Default
session_id str

Specific session ID to assign to the new session.

 None
Source code in src/mada/core/database/session_manager.py 
103
104
105
106
107
108
109
110
111
112
def create_new_session(self, session_id: str = None):
    """
    Save an empty session to the database.

    Args:
        session_id (str): Specific session ID to assign to the new session.
    """
    if not session_id:
        session_id = self.current_session_id
    self.chat_db.create_session(session_id)

create_session_id()

Create a new chat session ID.

Returns:

Type Description
str

A new session ID.
Source code in src/mada/core/database/session_manager.py 
 94
 95
 96
 97
 98
 99
100
101
def create_session_id(self) -> str:
    """
    Create a new chat session ID.

    Returns:
        A new session ID.
    """
    return str(uuid.uuid4())

delete_all_sessions(confirm=True)

Delete all sessions from the database.

Warning

This operation is irreversible and will delete all chat history.

Parameters:

Name Type Description Default
confirm bool

If True, prompt the user for confirmation before deleting. If False, delete without prompting.

 True
Source code in src/mada/core/database/session_manager.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
def delete_all_sessions(self, confirm: bool = True):
    """
    Delete all sessions from the database.

    Warning:
        This operation is irreversible and will delete all chat history.

    Args:
        confirm (bool): If True, prompt the user for confirmation before
            deleting. If False, delete without prompting.
    """
    self.chat_db.flush_database(confirm=confirm)
    self.current_session_id = None

delete_session(session_id=None)

Given a session ID, delete the session from the database.

If no session ID is provided, the currently tracked session is deleted.

Parameters:

Name Type Description Default
session_id str

Specific session ID to delete from the database.

 None
Source code in src/mada/core/database/session_manager.py 
155
156
157
158
159
160
161
162
163
164
165
166
def delete_session(self, session_id: str = None):
    """
    Given a session ID, delete the session from the database.

    If no session ID is provided, the currently tracked session is deleted.

    Args:
        session_id (str): Specific session ID to delete from the database.
    """
    if session_id is None:
        session_id = self.current_session_id
    self.chat_db.delete_session(session_id)

list_sessions()

List all sessions currently stored in the database.

Returns:

Type Description
List[Tuple[str, datetime]]

A list of sessions where each session has (session ID, last updated).
Source code in src/mada/core/database/session_manager.py 
146
147
148
149
150
151
152
153
def list_sessions(self) -> List[Tuple[str, datetime]]:
    """
    List all sessions currently stored in the database.

    Returns:
        A list of sessions where each session has (session ID, last updated).
    """
    return self.chat_db.list_sessions()

load_history()

Load the message history from the current session.

Returns:

Type Description
List[Dict]

A list of messages for the given session.
Source code in src/mada/core/database/session_manager.py 
127
128
129
130
131
132
133
134
def load_history(self) -> List[Dict]:
    """
    Load the message history from the current session.

    Returns:
        A list of messages for the given session.
    """
    return self.chat_db.load_session(self.current_session_id) or []

select_session(session_id)

Set the given session as the current session and return its history.

Parameters:

Name Type Description Default
session_id str

The ID of the session to select.

 required

Returns:

Type Description
List[Dict]

The session chat history.
Source code in src/mada/core/database/session_manager.py 
114
115
116
117
118
119
120
121
122
123
124
125
def select_session(self, session_id: str) -> List[Dict]:
    """
    Set the given session as the current session and return its history.

    Args:
        session_id (str): The ID of the session to select.

    Returns:
        The session chat history.
    """
    self.current_session_id = session_id
    return self.load_history()