SQLiteEntityStore Class — langchain Architecture

Architecture documentation for the SQLiteEntityStore class in entity.py from the langchain codebase.

Class python

Entity Profile

Dependency Diagram

graph TD
  40387f29_12a3_f291_fc18_4ce17ac91db3["SQLiteEntityStore"]
  d6a371b0_afe3_1ceb_cd58_c254c31b562e["BaseEntityStore"]
  40387f29_12a3_f291_fc18_4ce17ac91db3 -->|extends| d6a371b0_afe3_1ceb_cd58_c254c31b562e
  1ed59830_902e_f3ce_f9f6_76963a6fd122["entity.py"]
  40387f29_12a3_f291_fc18_4ce17ac91db3 -->|defined in| 1ed59830_902e_f3ce_f9f6_76963a6fd122
  28535b08_6603_aa75_f05d_abb90636cd88["__init__()"]
  40387f29_12a3_f291_fc18_4ce17ac91db3 -->|method| 28535b08_6603_aa75_f05d_abb90636cd88
  b84a6136_5602_e674_0b9e_90bbc1cefdb2["full_table_name()"]
  40387f29_12a3_f291_fc18_4ce17ac91db3 -->|method| b84a6136_5602_e674_0b9e_90bbc1cefdb2
  bb3de870_d9a2_b2ef_b746_608a90b30ec8["_execute_query()"]
  40387f29_12a3_f291_fc18_4ce17ac91db3 -->|method| bb3de870_d9a2_b2ef_b746_608a90b30ec8
  4913d94d_bd60_eefc_8e42_e5664da21e4e["_create_table_if_not_exists()"]
  40387f29_12a3_f291_fc18_4ce17ac91db3 -->|method| 4913d94d_bd60_eefc_8e42_e5664da21e4e
  323eba30_b983_8952_8d7a_c69c375e843a["get()"]
  40387f29_12a3_f291_fc18_4ce17ac91db3 -->|method| 323eba30_b983_8952_8d7a_c69c375e843a
  5815fe41_90c9_034c_6b77_d9442e944639["set()"]
  40387f29_12a3_f291_fc18_4ce17ac91db3 -->|method| 5815fe41_90c9_034c_6b77_d9442e944639
  bcb05110_0d1a_dff9_9d77_188d5452df3f["delete()"]
  40387f29_12a3_f291_fc18_4ce17ac91db3 -->|method| bcb05110_0d1a_dff9_9d77_188d5452df3f
  51c4ff54_3688_e0d2_d041_9ea9d10b2773["exists()"]
  40387f29_12a3_f291_fc18_4ce17ac91db3 -->|method| 51c4ff54_3688_e0d2_d041_9ea9d10b2773
  de8bc7d0_c825_fe9e_a5db_d7c69aae0700["clear()"]
  40387f29_12a3_f291_fc18_4ce17ac91db3 -->|method| de8bc7d0_c825_fe9e_a5db_d7c69aae0700

Relationship Graph

Source Code

libs/langchain/langchain_classic/memory/entity.py lines 343–454

class SQLiteEntityStore(BaseEntityStore):
    """SQLite-backed Entity store with safe query construction."""

    session_id: str = "default"
    table_name: str = "memory_store"
    conn: Any = None

    model_config = ConfigDict(
        arbitrary_types_allowed=True,
    )

    def __init__(
        self,
        session_id: str = "default",
        db_file: str = "entities.db",
        table_name: str = "memory_store",
        *args: Any,
        **kwargs: Any,
    ):
        """Initializes the SQLiteEntityStore.

        Args:
            session_id: Unique identifier for the session.
            db_file: Path to the SQLite database file.
            table_name: Name of the table to store entities.
            *args: Additional positional arguments.
            **kwargs: Additional keyword arguments.
        """
        super().__init__(*args, **kwargs)
        try:
            import sqlite3
        except ImportError as e:
            msg = (
                "Could not import sqlite3 python package. "
                "Please install it with `pip install sqlite3`."
            )
            raise ImportError(msg) from e

        # Basic validation to prevent obviously malicious table/session names
        if not table_name.isidentifier() or not session_id.isidentifier():
            # Since we validate here, we can safely suppress the S608 bandit warning
            msg = "Table name and session ID must be valid Python identifiers."
            raise ValueError(msg)

        self.conn = sqlite3.connect(db_file)
        self.session_id = session_id
        self.table_name = table_name
        self._create_table_if_not_exists()

    @property
    def full_table_name(self) -> str:
        """Returns the full table name with session ID."""
        return f"{self.table_name}_{self.session_id}"

    def _execute_query(self, query: str, params: tuple = ()) -> "sqlite3.Cursor":
        """Executes a query with proper connection handling."""
        with self.conn:
            return self.conn.execute(query, params)

    def _create_table_if_not_exists(self) -> None:
        """Creates the entity table if it doesn't exist, using safe quoting."""
        # Use standard SQL double quotes for the table name identifier
        create_table_query = f"""
            CREATE TABLE IF NOT EXISTS "{self.full_table_name}" (
                key TEXT PRIMARY KEY,
                value TEXT
            )
        """
        self._execute_query(create_table_query)

    def get(self, key: str, default: str | None = None) -> str | None:
        """Retrieves a value, safely quoting the table name."""
        # `?` placeholder is used for the value to prevent SQL injection
        # Ignore S608 since we validate for malicious table/session names in `__init__`
        query = f'SELECT value FROM "{self.full_table_name}" WHERE key = ?'  # noqa: S608
        cursor = self._execute_query(query, (key,))
        result = cursor.fetchone()
        return result[0] if result is not None else default

    def set(self, key: str, value: str | None) -> None:
        """Inserts or replaces a value, safely quoting the table name."""