Skip to content

base_db

Base class for chat history database implementations.

BaseChatDatabase

Bases: ABC

Abstract base class for chat history database implementations.

This class defines the common interface that all database implementations must adhere to, ensuring consistency across different database types.

Attributes:

Name Type Description
db_config DatabaseConfig

Configuration object containing database connection details.

Methods:

Name Description
init_db

Initializes the database by creating the necessary table if it doesn't exist.
add_message

Saves a message in the database.
create_session

Create a new session entry in the database.
load_session

Loads a session's messages from the database.
list_sessions

Lists all sessions in the database.
delete_session

Deletes a session from the database.
confirm_db_flush

Confirms that the user wants to flush the entire database.
flush_database

Removes all data from the database, effectively resetting it.
Source code in src/mada/core/database/base_db.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 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
class BaseChatDatabase(ABC):
    """
    Abstract base class for chat history database implementations.

    This class defines the common interface that all database implementations
    must adhere to, ensuring consistency across different database types.

    Attributes:
        db_config (core.config.DatabaseConfig): Configuration object containing database
            connection details.

    Methods:
        init_db: Initializes the database by creating the necessary table if it
            doesn't exist.
        add_message: Saves a message in the database.
        create_session: Create a new session entry in the database.
        load_session: Loads a session's messages from the database.
        list_sessions: Lists all sessions in the database.
        delete_session: Deletes a session from the database.
        confirm_db_flush: Confirms that the user wants to flush the entire database.
        flush_database: Removes all data from the database, effectively resetting
            it.
    """

    def __init__(self, db_config: DatabaseConfig):
        """
        Initialize the database by creating necessary tables or structures.

        Args:
            db_config (core.config.DatabaseConfig): Configuration settings for the database.
        """
        self.db_config = db_config
        self.init_db()

    @abstractmethod
    def init_db(self):
        """
        Initialize the database by creating necessary tables or structures.
        """

    @abstractmethod
    def add_message(
        self, session_id: str, role: str, content: str, timestamp: datetime = None
    ):
        """
        Add a single message to the messages table in the database.

        Args:
            session_id (str): The ID of the session that this message is associated with
            role (str): The role (user or assistant) to designate who wrote the message
            content (str): The message contents
            timestamp (datetime): The time that the message was created
        """

    @abstractmethod
    def create_session(self, session_id: str):
        """
        Create a new session entry.

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

    @abstractmethod
    def load_session(self, session_id: str) -> List:
        """
        Load a session's messages from the database.

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

        Returns:
            list: List of messages from the session.
        """

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

        Returns:
            list: List of tuples containing session_id and last_updated.
        """

    @abstractmethod
    def delete_session(self, session_id: str):
        """
        Delete a session from the database.

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

    def confirm_db_flush(self) -> bool:
        """
        Confirm that the user wants to flush the entire database.

        Returns:
            True if the user wants to flush the entire database. False otherwise.
        """
        valid_inputs = ["y", "n"]
        user_input = (
            input("Are you sure you want to flush the entire database? (y/n): ")
            .strip()
            .lower()
        )
        while user_input not in valid_inputs:
            user_input = (
                input("Invalid input. Use 'y' for 'yes' or 'n' for 'no': ")
                .strip()
                .lower()
            )

        if user_input == "y":
            return True
        return False

    @abstractmethod
    def flush_database(self, confirm: bool = True):
        """
        Remove all data from the database, effectively resetting it.

        Warning:
            This operation is irreversible and will delete all sessions.

        Args:
            confirm (bool): If True, confirm that the user wants to flush the
                database prior to deleting anything. Otherwise, don't ask.
        """

__init__(db_config)

Initialize the database by creating necessary tables or structures.

Parameters:

Name Type Description Default
db_config DatabaseConfig

Configuration settings for the database.

 required
Source code in src/mada/core/database/base_db.py 
39
40
41
42
43
44
45
46
47
def __init__(self, db_config: DatabaseConfig):
    """
    Initialize the database by creating necessary tables or structures.

    Args:
        db_config (core.config.DatabaseConfig): Configuration settings for the database.
    """
    self.db_config = db_config
    self.init_db()

add_message(session_id, role, content, timestamp=None) abstractmethod

Add a single message to the messages table in the database.

Parameters:

Name Type Description Default
session_id str

The ID of the session that this message is associated with

 required
role str

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

 required
content str

The message contents

 required
timestamp datetime

The time that the message was created

 None
Source code in src/mada/core/database/base_db.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
@abstractmethod
def add_message(
    self, session_id: str, role: str, content: str, timestamp: datetime = None
):
    """
    Add a single message to the messages table in the database.

    Args:
        session_id (str): The ID of the session that this message is associated with
        role (str): The role (user or assistant) to designate who wrote the message
        content (str): The message contents
        timestamp (datetime): The time that the message was created
    """

confirm_db_flush()

Confirm that the user wants to flush the entire database.

Returns:

Type Description
bool

True if the user wants to flush the entire database. False otherwise.
Source code in src/mada/core/database/base_db.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def confirm_db_flush(self) -> bool:
    """
    Confirm that the user wants to flush the entire database.

    Returns:
        True if the user wants to flush the entire database. False otherwise.
    """
    valid_inputs = ["y", "n"]
    user_input = (
        input("Are you sure you want to flush the entire database? (y/n): ")
        .strip()
        .lower()
    )
    while user_input not in valid_inputs:
        user_input = (
            input("Invalid input. Use 'y' for 'yes' or 'n' for 'no': ")
            .strip()
            .lower()
        )

    if user_input == "y":
        return True
    return False

create_session(session_id) abstractmethod

Create a new session entry.

Parameters:

Name Type Description Default
session_id str

The ID of the session to create.

 required
Source code in src/mada/core/database/base_db.py 
69
70
71
72
73
74
75
76
@abstractmethod
def create_session(self, session_id: str):
    """
    Create a new session entry.

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

delete_session(session_id) abstractmethod

Delete a session from the database.

Parameters:

Name Type Description Default
session_id str

The ID of the session to delete.

 required
Source code in src/mada/core/database/base_db.py 
 99
100
101
102
103
104
105
106
@abstractmethod
def delete_session(self, session_id: str):
    """
    Delete a session from the database.

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

flush_database(confirm=True) abstractmethod

Remove all data from the database, effectively resetting it.

Warning

This operation is irreversible and will delete all sessions.

Parameters:

Name Type Description Default
confirm bool

If True, confirm that the user wants to flush the database prior to deleting anything. Otherwise, don't ask.

 True
Source code in src/mada/core/database/base_db.py 
132
133
134
135
136
137
138
139
140
141
142
143
@abstractmethod
def flush_database(self, confirm: bool = True):
    """
    Remove all data from the database, effectively resetting it.

    Warning:
        This operation is irreversible and will delete all sessions.

    Args:
        confirm (bool): If True, confirm that the user wants to flush the
            database prior to deleting anything. Otherwise, don't ask.
    """

init_db() abstractmethod

Initialize the database by creating necessary tables or structures.

Source code in src/mada/core/database/base_db.py 
49
50
51
52
53
@abstractmethod
def init_db(self):
    """
    Initialize the database by creating necessary tables or structures.
    """

list_sessions() abstractmethod

List all sessions in the database.

Returns:

Name Type Description
list List[Tuple[str, datetime]]

List of tuples containing session_id and last_updated.
Source code in src/mada/core/database/base_db.py 
90
91
92
93
94
95
96
97
@abstractmethod
def list_sessions(self) -> List[Tuple[str, datetime]]:
    """
    List all sessions in the database.

    Returns:
        list: List of tuples containing session_id and last_updated.
    """

load_session(session_id) abstractmethod

Load a session's messages from the database.

Parameters:

Name Type Description Default
session_id str

The ID of the session to load.

 required

Returns:

Name Type Description
list List

List of messages from the session.
Source code in src/mada/core/database/base_db.py 
78
79
80
81
82
83
84
85
86
87
88
@abstractmethod
def load_session(self, session_id: str) -> List:
    """
    Load a session's messages from the database.

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

    Returns:
        list: List of messages from the session.
    """