fix(query builder): correct edge cases and typing of query builder

.github/workflows/ci.yml

@@ -106,7 +106,10 @@ jobs:
         run: uv sync --all-extras --dev
 
       - name: Test
-        run: uv run pytest -n 2 --dist=loadgroup
+        env:
+          PYTHONFAULTHANDLER: "1"
+          PYTEST_ADDOPTS: "--max-worker-restart=0 -s"
+        run: timeout 900s uv run pytest -n 2 --dist=loadgroup
 
   # test-windows:
   #   runs-on: windows-latest

.pre-commit-config.yaml

@@ -17,7 +17,7 @@ repos:
       - id: mixed-line-ending
       - id: trailing-whitespace
   - repo: https://github.com/charliermarsh/ruff-pre-commit
-    rev: "v0.14.5"
+    rev: "v0.14.6"
     hooks:
       - id: ruff
         args: ["--fix"]
@@ -40,7 +40,7 @@ repos:
       - id: slotscheck
         exclude: "docs|.github"
   - repo: https://github.com/sphinx-contrib/sphinx-lint
-    rev: "v1.0.1"
+    rev: "v1.0.2"
     hooks:
       - id: sphinx-lint
         args: ["--jobs", "1"]

docs/examples/usage/usage_configuration_1.py

@@ -2,6 +2,7 @@
 
 
 def test_sqlite_memory_db() -> None:
+    from rich import print
 
     # start-example
     from sqlspec import SQLSpec
@@ -15,6 +16,7 @@ def test_sqlite_memory_db() -> None:
 
     # Use the database
     with db_manager.provide_session(db) as session:
-        result = session.execute("SELECT 1")
+        value = session.select_value("SELECT 1")
+        print(value)
         # end-example
-        assert result[0] == {"1": 1}
+        assert value == 1

docs/examples/usage/usage_data_flow_11.py

@@ -5,6 +5,8 @@
 
 def test_sql_result_object() -> None:
     """Test accessing SQLResult object properties."""
+    from rich import print
+
     from sqlspec import SQLSpec
     from sqlspec.adapters.sqlite import SqliteConfig
 
@@ -13,15 +15,13 @@ def test_sql_result_object() -> None:
 
     # start-example
     with db_manager.provide_session(db) as session:
-        result = session.execute("SELECT 'test' as col1, 'value' as col2")
+        result = session.select("SELECT 'test' as col1, 'value' as col2")
+        print(len(result))  # length of result set
 
-        # Access result data
-        result.data  # List of dictionaries
-        result.rows_affected  # Number of rows modified (INSERT/UPDATE/DELETE)
-        result.column_names  # Column names for SELECT
-        result.operation_type  # "SELECT", "INSERT", "UPDATE", "DELETE", "SCRIPT"
+        # Access result data (SQLResult is iterable over rows)
+        print(result)
     # end-example
 
     # Verify result properties
-    assert result.data is not None
-    assert result.column_names is not None
+    assert result is not None
+    assert len(result) == 1

docs/examples/usage/usage_data_flow_13.py

@@ -5,9 +5,9 @@
 
 def test_schema_mapping() -> None:
     """Test mapping results to typed objects."""
-
     # start-example
     from pydantic import BaseModel
+    from rich import print
 
     from sqlspec import SQLSpec
     from sqlspec.adapters.sqlite import SqliteConfig
@@ -27,17 +27,17 @@ class User(BaseModel):
         session.execute("INSERT INTO users VALUES (1, 'Alice', '[email protected]', 1)")
 
         # Execute query
-        result = session.execute("SELECT id, name, email, is_active FROM users")
-
-        # Map results to typed User instances
-        users: list[User] = result.all(schema_type=User)
+        result = session.select("SELECT id, name, email, is_active FROM users", schema_type=User)
+        print(len(result))
 
         # Or get single typed user
-        single_result = session.execute("SELECT id, name, email, is_active FROM users WHERE id = ?", 1)
-        user: User = single_result.one(schema_type=User)  # Type-safe!
+        single_result = session.select_one(
+            "SELECT id, name, email, is_active FROM users WHERE id = ?", 1, schema_type=User
+        )
+        print(single_result)
     # end-example
 
     # Verify typed results
-    assert len(users) == 1
-    assert isinstance(user, User)
-    assert user.id == 1
+    assert len(result) == 1
+    assert isinstance(single_result, User)
+    assert single_result.id == 1

docs/examples/usage/usage_data_flow_8.py

@@ -16,6 +16,8 @@ def test_statement_execution() -> None:
 
     with db_manager.provide_session(db) as session:
         result = session.execute(sql_statement)
+        print(result.rows_affected)
+        print(result.parameters)
         message = result.scalar()
     # end-example
 

docs/examples/usage/usage_data_flow_9.py

@@ -5,6 +5,8 @@
 
 def test_driver_execution() -> None:
     """Test driver execution pattern."""
+    from rich import print
+
     from sqlspec import SQLSpec
     from sqlspec.adapters.sqlite import SqliteConfig
 
@@ -13,7 +15,8 @@ def test_driver_execution() -> None:
     db_manager = SQLSpec()
     db = db_manager.add_config(SqliteConfig(pool_config={"database": ":memory:"}))
     with db_manager.provide_session(db) as session:
-        result = session.execute("SELECT 'test' as message")
+        message = session.select_value("SELECT 'test' as message")
+        print(message)
     # end-example
 
-    assert result is not None
+    assert message == "test"

docs/examples/usage/usage_query_builder_1.py

@@ -0,0 +1,31 @@
+from pathlib import Path
+
+__all__ = ("test_example_1",)
+
+
+def test_example_1(tmp_path: Path) -> None:
+    from sqlspec import SQLSpec, sql
+    from sqlspec.adapters.sqlite.config import SqliteConfig
+
+    db = SQLSpec()
+    database = tmp_path / "example1.db"  # Database file path
+    config = SqliteConfig(
+        pool_config={
+            "database": database.name,
+            "timeout": 5.0,  # Lock timeout in seconds
+            "check_same_thread": False,  # Allow multi-thread access
+            "cached_statements": 100,  # Statement cache size
+            "uri": False,  # Enable URI mode
+        }
+    )
+    with db.provide_session(config) as session:
+        create_table_query = """CREATE TABLE if not exists users(id int primary key,name text,email text, status text, created_at timestamp)"""
+        _ = session.execute(create_table_query)
+        # start-example
+        # Build SELECT query
+        query = (
+            sql.select("id", "name", "email").from_("users").where("status = ?").order_by("created_at DESC").limit(10)
+        )
+        # Execute with session
+        session.execute(query, "active")
+        # end-example

docs/examples/usage/usage_query_builder_10.py

@@ -0,0 +1,37 @@
+from pathlib import Path
+
+__all__ = ("test_example_10",)
+
+
+def test_example_10(tmp_path: Path) -> None:
+    from sqlspec import SQLSpec, sql
+    from sqlspec.adapters.sqlite.config import SqliteConfig
+
+    db = SQLSpec()
+    database = tmp_path / "example10.db"  # Database file path
+    config = SqliteConfig(
+        pool_config={
+            "database": database.name,
+            "timeout": 5.0,  # Lock timeout in seconds
+            "check_same_thread": False,  # Allow multi-thread access
+            "cached_statements": 100,  # Statement cache size
+            "uri": False,  # Enable URI mode
+        }
+    )
+
+    with db.provide_session(config) as session:
+        session.execute(
+            """CREATE TABLE if not exists users(id integer primary key autoincrement, name text, email text)"""
+        )
+        # start-example
+        # Multiple value sets
+        query = (
+            sql.insert("users")
+            .columns("name", "email")
+            .values("Alice", "[email protected]")
+            .values("Bob", "[email protected]")
+            .values("Charlie", "[email protected]")
+        )
+
+        session.execute(query)
+        # end-example

docs/examples/usage/usage_query_builder_11.py

@@ -0,0 +1,30 @@
+from pathlib import Path
+
+__all__ = ("test_example_11",)
+
+
+def test_example_11(tmp_path: Path) -> None:
+    from sqlspec import SQLSpec, sql
+    from sqlspec.adapters.sqlite.config import SqliteConfig
+
+    db = SQLSpec()
+    database = tmp_path / "example11.db"
+    config = SqliteConfig(
+        pool_config={
+            "database": database.name,
+            "timeout": 5.0,
+            "check_same_thread": False,
+            "cached_statements": 100,
+            "uri": False,
+        }
+    )
+    with db.provide_session(config) as session:
+        session.execute(
+            """CREATE TABLE if not exists users(id integer primary key autoincrement, name text, email text, created_at timestamp)"""
+        )
+        # start-example
+        # PostgreSQL RETURNING clause
+        (sql.insert("users").columns("name", "email").values("?", "?").returning("id", "created_at"))
+
+        # SQLite does not support RETURNING, so we skip execution for this example.
+        # end-example

docs/examples/usage/usage_query_builder_12.py

@@ -0,0 +1,35 @@
+from pathlib import Path
+
+__all__ = ("test_example_12",)
+
+
+def test_example_12(tmp_path: Path) -> None:
+    from sqlspec import SQLSpec, sql
+    from sqlspec.adapters.sqlite.config import SqliteConfig
+
+    db = SQLSpec()
+    database = tmp_path / "example12.db"
+    config = SqliteConfig(
+        pool_config={
+            "database": database.name,
+            "timeout": 5.0,
+            "check_same_thread": False,
+            "cached_statements": 100,
+            "uri": False,
+        }
+    )
+    with db.provide_session(config) as session:
+        session.execute(
+            """CREATE TABLE if not exists users(id integer primary key autoincrement, name text, email text)"""
+        )
+        # Insert test data
+        session.execute("INSERT INTO users (name, email) VALUES ('Alice', '[email protected]')")
+
+        # start-example
+        # Update with WHERE
+        query = sql.update("users").set("email", "[email protected]").where("id = 1")
+        # SQL: UPDATE users SET email = :email WHERE id = 1
+
+        session.execute(query)
+        # print(f"Updated {result.rows_affected} rows")
+        # end-example

docs/examples/usage/usage_query_builder_13.py

@@ -0,0 +1,41 @@
+from pathlib import Path
+
+__all__ = ("test_example_13",)
+
+
+def test_example_13(tmp_path: Path) -> None:
+    from sqlspec import SQLSpec, sql
+    from sqlspec.adapters.sqlite.config import SqliteConfig
+
+    db = SQLSpec()
+    database = tmp_path / "example13.db"
+    config = SqliteConfig(
+        pool_config={
+            "database": database.name,
+            "timeout": 5.0,
+            "check_same_thread": False,
+            "cached_statements": 100,
+            "uri": False,
+        }
+    )
+    with db.provide_session(config) as session:
+        session.execute(
+            """CREATE TABLE if not exists users(id integer primary key autoincrement, name text, email text, updated_at timestamp)"""
+        )
+        # Insert test data
+        session.execute("INSERT INTO users (name, email) VALUES ('Alice', '[email protected]')")
+
+        # start-example
+        # Update multiple columns
+        from sqlglot import exp
+
+        query = (
+            sql.update("users")
+            .set("name", "New Name")
+            .set("email", "[email protected]")
+            .set("updated_at", exp.CurrentTimestamp())
+            .where("id = 1")
+        )
+
+        session.execute(query)
+        # end-example

docs/examples/usage/usage_query_builder_14.py

@@ -0,0 +1,43 @@
+from pathlib import Path
+from typing import Any
+
+__all__ = ("test_example_14",)
+
+
+def test_example_14(tmp_path: Path) -> None:
+    from sqlspec import SQLSpec, sql
+    from sqlspec.adapters.sqlite.config import SqliteConfig
+
+    db = SQLSpec()
+    database = tmp_path / "example14.db"
+    config = SqliteConfig(
+        pool_config={
+            "database": database.name,
+            "timeout": 5.0,
+            "check_same_thread": False,
+            "cached_statements": 100,
+            "uri": False,
+        }
+    )
+    with db.provide_session(config) as session:
+        session.execute(
+            """CREATE TABLE if not exists users(id integer primary key autoincrement, name text, email text, status text)"""
+        )
+        # Insert test data
+        session.execute("INSERT INTO users (name, email, status) VALUES ('Alice', '[email protected]', 'inactive')")
+
+        # start-example
+        # Dynamic update builder
+        def update_user(user_id: Any, **fields: Any) -> Any:
+            query = sql.update("users")
+
+            for field, value in fields.items():
+                query = query.set(field, value)
+
+            query = query.where(f"id = {user_id}")
+
+            return session.execute(query)
+
+        # Usage
+        update_user(1, name="Alice Updated", email="[email protected]", status="active")
+        # end-example