feat(result): add schema_type parameter to SQLResult helper methods (#106)

Adds optional `schema_type` parameter to `SQLResult` data retrieval methods, enabling direct type-safe data transformation across the entire result API. Also refactors `to_schema()` functionality into a standalone utility module to eliminate circular imports.

Author: cofin

docs/examples/litestar_asyncpg.py

@@ -47,8 +47,9 @@ async def get_version(db_session: AsyncpgDriver) -> dict[str, Any]:
     """Get PostgreSQL version information."""
     result = await db_session.execute(SQL("SELECT version() as version"))
 
-    if result.data:
-        return {"database": "PostgreSQL", "version": result.data[0]["version"][:50] + "..."}
+    version_row = result.one_or_none()
+    if version_row:
+        return {"database": "PostgreSQL", "version": version_row["version"][:50] + "..."}
     return {"error": "Could not retrieve version"}
 
 
@@ -62,8 +63,9 @@ async def list_tables(db_session: AsyncpgDriver) -> dict[str, Any]:
         ORDER BY table_name
         """)
 
-    if result.data:
-        tables = [row["table_name"] for row in result.data]
+    tables_data = result.all()
+    if tables_data:
+        tables = [row["table_name"] for row in tables_data]
         return {"tables": tables, "count": len(tables)}
     return {"tables": [], "count": 0}
 

docs/examples/litestar_duckllm.py

@@ -33,8 +33,9 @@ class ChatMessage(Struct):
 
 @post("/chat", sync_to_thread=True)
 def duckllm_chat(db_session: DuckDBDriver, data: ChatMessage) -> ChatMessage:
-    results = db_session.execute("SELECT open_prompt(?)", data.message).get_first()
-    return db_session.to_schema(results or {"message": "No response from DuckLLM"}, schema_type=ChatMessage)
+    result = db_session.execute("SELECT open_prompt(?)", data.message)
+    messages = result.get_data(schema_type=ChatMessage)
+    return messages[0] if messages else ChatMessage(message="No response from DuckLLM")
 
 
 spec = SQLSpec()

docs/examples/standalone_demo.py

@@ -659,16 +659,17 @@ def interactive() -> None:
                             spec_temp.add_config(db_config)
                             with spec_temp.provide_session(type(db_config)) as driver:
                                 result = driver.execute(sql_obj)
-                                if result.data:
-                                    console.print(f"[green]Returned {len(result.data)} rows[/green]")
-                                    if len(result.data) <= MAX_ROWS_TO_DISPLAY:
-                                        for row in result.data:
+                                data = result.all()
+                                if data:
+                                    console.print(f"[green]Returned {len(data)} rows[/green]")
+                                    if len(data) <= MAX_ROWS_TO_DISPLAY:
+                                        for row in data:
                                             console.print(f"  {row}")
                                     else:
                                         console.print("  First 3 rows:")
-                                        for row in result.data[:3]:
+                                        for row in data[:3]:
                                             console.print(f"    {row}")
-                                        console.print(f"  ... and {len(result.data) - 3} more")
+                                        console.print(f"  ... and {len(data) - 3} more")
                         except Exception as e:
                             console.print(f"[yellow]Query built successfully but execution failed: {e}[/yellow]")
 

docs/extensions/litestar/dependency_injection.rst

@@ -69,7 +69,7 @@ Injects the SQLSpec driver instance with full query capabilities (recommended).
    @get("/users")
    async def get_users(db_session: AsyncpgDriver) -> dict:
        result = await db_session.execute("SELECT * FROM users")
-       return {"users": result.data}
+       return {"users": result.all()}
 
 **When to use**: All standard database operations (recommended).
 
@@ -92,7 +92,7 @@ Dependencies are resolved by type annotation:
    async def handler(db_session: AsyncpgDriver) -> dict:
        # SQLSpec injects AsyncpgDriver instance
        result = await db_session.execute("SELECT * FROM users")
-       return {"users": result.data}
+       return {"users": result.all()}
 
 By Dependency Key
 -----------------
@@ -234,7 +234,7 @@ Use specific driver types for better type checking:
        # IDE knows exact driver types
        pg_result = await postgres.execute("SELECT * FROM users")
        duck_result = await duckdb.execute("SELECT * FROM events")
-       return {"pg": pg_result.data, "duck": duck_result.data}
+       return {"pg": pg_result.all(), "duck": duck_result.all()}
 
 Best Practices
 ==============
@@ -252,7 +252,7 @@ Prefer ``db_session`` for standard database operations:
    @get("/users")
    async def get_users(db_session: AsyncpgDriver) -> dict:
        result = await db_session.execute("SELECT * FROM users")
-       return {"users": result.data}
+       return {"users": result.all()}
 
    # Advanced: Use connection only when needed
    @get("/bulk-import")

docs/extensions/litestar/index.rst

@@ -66,7 +66,7 @@ Here's a simple example of creating a Litestar application with SQLSpec integrat
    @get("/users")
    async def list_users(db_session: AsyncpgDriver) -> dict:
        result = await db_session.execute("SELECT * FROM users LIMIT 10")
-       return {"users": result.data}
+       return {"users": result.all()}
 
    @post("/users")
    async def create_user(

docs/extensions/litestar/quickstart.rst

@@ -89,7 +89,7 @@ Define route handlers that use dependency injection to access the database:
    @get("/users")
    async def list_users(db_session: AsyncpgDriver) -> dict:
        result = await db_session.execute("SELECT * FROM users LIMIT 10")
-       return {"users": result.data}
+       return {"users": result.all()}
 
    @get("/users/{user_id:int}")
    async def get_user(
@@ -184,7 +184,7 @@ Here's a complete working example:
        result = await db_session.execute(
            "SELECT id, name, email FROM users ORDER BY id LIMIT 10"
        )
-       return {"users": result.data}
+       return {"users": result.all()}
 
    @get("/users/{user_id:int}")
    async def get_user(

docs/reference/core.rst

@@ -137,36 +137,48 @@ Result Processing
 
    **Attributes:**
 
-   - ``data`` - List of result rows (dicts)
+   - ``data`` - List of result rows (dicts) - prefer using helper methods instead
    - ``rows_affected`` - Number of rows affected by INSERT/UPDATE/DELETE
    - ``columns`` - Column names
    - ``metadata`` - Query metadata
 
-   **Methods:**
+   **Recommended Helper Methods:**
 
    .. code-block:: python
 
       result = await session.execute("SELECT * FROM users")
 
-      # Access data
-      all_rows = result.data
+      # Get all rows (replaces result.data)
+      all_rows = result.all()
+
+      # Get exactly one row (raises if not exactly one)
+      user = result.one()
+
+      # Get one row or None (raises if multiple)
+      user = result.one_or_none()
+
+      # Get first row without validation
       first_row = result.get_first()
-      first_row_or_none = result.get_first_or_none()
 
-      # Map to types
+      # Get scalar value (first column of first row)
+      count = result.scalar()
+
+      # Map to typed models
       from pydantic import BaseModel
 
       class User(BaseModel):
           id: int
           name: str
           email: str
 
-      users = result.as_type(User)  # List[User]
-      user = result.as_type_one(User)  # User
-      user_or_none = result.as_type_one_or_none(User)  # User | None
+      # Get all rows as typed models
+      users: list[User] = result.all(schema_type=User)
+
+      # Get exactly one row as typed model
+      user: User = result.one(schema_type=User)
 
-      # Get scalar value
-      count = result.scalar()  # First column of first row
+      # Get one or none as typed model
+      user: User | None = result.one_or_none(schema_type=User)
 
 **Type Mapping:**
 

docs/reference/extensions.rst

@@ -156,7 +156,7 @@ Plugin
       @get("/users")
       async def get_users(db: AsyncpgDriver) -> list[dict]:
           result = await db.select("SELECT * FROM users")
-          return result.data
+          return result.all()
 
       app = Litestar(route_handlers=[get_users], plugins=[plugin])
 

docs/reference/index.rst

@@ -139,9 +139,10 @@ Common Workflows
 **Processing Results:**
 
 1. Execute query: ``result = await session.execute(sql)``
-2. Access data: ``result.data`` (list of dicts)
-3. Get first row: ``result.get_first()``
-4. Map to models: ``result.as_type(User)``
+2. Get all rows: ``result.all()`` (list of dicts)
+3. Get one row: ``result.one()`` (raises if not exactly one)
+4. Get first row: ``result.get_first()`` (returns first or None)
+5. Map to models: ``result.all(schema_type=User)`` or ``result.one(schema_type=User)``
 
 See Also
 ========

docs/usage/data_flow.rst

@@ -269,15 +269,15 @@ SQLSpec can automatically map results to typed objects:
        email: str
        is_active: Optional[bool] = True
 
-   # Execute with schema type
-   result = session.execute(
-       "SELECT id, name, email, is_active FROM users",
-       schema_type=User
-   )
+   # 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)
 
-   # Results are typed User instances
-   users: list[User] = result.to_schema()
-   user: User = result.one()  # Type-safe!
+   # 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!
 
 **Supported Schema Types**