Skip to content

teff.tool.builtin.sql

teff.tool.builtin.sql

SQL tools — read-only queries and schema inspection for SQLite/PostgreSQL.

Classes:

Name Description
SQLDescribeTool

Describe a table's columns and types.
SQLListTablesTool

List the tables in a database.
SQLQueryTool

Run a read-only SQL query against a database.

SQLDescribeTool

Bases: _SQLBase

Describe a table's columns and types.

Source code in teff/tool/builtin/sql.py 
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
class SQLDescribeTool(_SQLBase):
    """Describe a table's columns and types."""

    name = "sql_describe"
    description = "Describe a table's columns and types"

    def run(self, table: str = "") -> str:  # type: ignore[override]
        if not table:
            raise ValueError("table is required")
        conn = self._connect()
        try:
            if self.db_type == "sqlite":
                cursor = conn.execute(f'PRAGMA table_info("{table}")')
                return self._format(cursor.description, cursor.fetchall())
            with conn.cursor() as cursor:
                cursor.execute(
                    "SELECT column_name, data_type, is_nullable "
                    "FROM information_schema.columns "
                    "WHERE table_schema='public' AND table_name=%s "
                    "ORDER BY ordinal_position",
                    (table,),
                )
                return self._format(cursor.description, cursor.fetchall())
        finally:
            conn.close()

SQLListTablesTool

Bases: _SQLBase

List the tables in a database.

Source code in teff/tool/builtin/sql.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
class SQLListTablesTool(_SQLBase):
    """List the tables in a database."""

    name = "sql_list_tables"
    description = "List the tables in a database"

    def run(self) -> str:  # type: ignore[override]
        conn = self._connect()
        try:
            if self.db_type == "sqlite":
                cursor = conn.execute(_SQLITE_TABLES)
            else:
                with conn.cursor() as cursor:
                    cursor.execute(_POSTGRES_TABLES)
                    return self._format(cursor.description, cursor.fetchall())
            return self._format(cursor.description, cursor.fetchall())
        finally:
            conn.close()

SQLQueryTool

Bases: _SQLBase

Run a read-only SQL query against a database.

Only SELECT/WITH (read) statements are allowed; anything that would mutate data (INSERT, UPDATE, DELETE, DDL, …) is rejected. Placeholders match the backend: ? for SQLite, %s for PostgreSQL.

Source code in teff/tool/builtin/sql.py 
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
class SQLQueryTool(_SQLBase):
    """Run a read-only SQL query against a database.

    Only ``SELECT``/``WITH`` (read) statements are allowed; anything that
    would mutate data (``INSERT``, ``UPDATE``, ``DELETE``, DDL, …) is
    rejected. Placeholders match the backend: ``?`` for SQLite, ``%s``
    for PostgreSQL.
    """

    name = "sql_query"
    description = "Run a read-only SQL SELECT query against a database"

    def _guard(self, query: str) -> None:
        first = query.lstrip().split(None, 1)[0].upper() if query.strip() else ""
        if first not in ("SELECT", "WITH", "EXPLAIN"):
            msg = f"sql_query is read-only: unsupported statement '{first or query}'"
            raise ValueError(msg)

    def run(self, query: str = "", params: list | None = None, limit: int = 100) -> str:  # type: ignore[override]
        if not query:
            raise ValueError("query is required")
        self._guard(query)
        conn = self._connect()
        try:
            if self.db_type == "sqlite":
                cursor = conn.execute(query, params or ())
            else:
                with conn.cursor() as cursor:
                    cursor.execute(query, params or ())
                    rows = cursor.fetchmany(limit)
                    return self._format(cursor.description, rows)
            return self._format(cursor.description, cursor.fetchmany(limit))
        finally:
            conn.close()