Result Processing¶

Retrieve Result Data¶

fetch_one() - Get Single Record¶

async with engine.execute("SELECT * FROM users WHERE id = %s", (1,)) as result:
    user = await result.fetch_one()
    if user:
        print(user)

fetch_many() - Get Multiple Records¶

async with engine.execute("SELECT * FROM users") as result:
    users = await result.fetch_many(10)  # Get 10 records
    for user in users:
        print(user)

fetch_all() - Get All Records¶

async with engine.execute("SELECT * FROM users") as result:
    all_users = await result.fetch_all()
    print(f"Total {len(all_users)} records")

iterate() - Async Iterator¶

async with engine.execute("SELECT * FROM users") as result:
    async for user in result.iterate():
        print(user)

Direct Iteration of Result¶

# Method 1: Direct iteration of execute result
async for user in engine.execute("SELECT * FROM users"):
    print(user)

# Method 2: Using async with then iterate
async with engine.execute("SELECT * FROM users") as result:
    async for user in result:
        print(user)

Result Types¶

Default tuple Type¶

async with engine.execute("SELECT id, name, email FROM users") as result:
    user = await result.fetch_one()
    # user is tuple: (1, "张三", "zhangsan@example.com")

dict Type¶

async with engine.execute(
    "SELECT id, name, email FROM users",
    result_class=dict
) as result:
    user = await result.fetch_one()
    # user is dict: {"id": 1, "name": "张三", "email": "zhangsan@example.com"}

Custom Model Type¶

from pydantic import BaseModel
from asmysql import Engine

class User(BaseModel):
    id: int
    name: str
    email: str

engine = Engine(url="mysql://root:pass@127.0.0.1:3306/")
await engine.connect()

async with engine.execute(
    "SELECT id, name, email FROM users",
    result_class=User
) as result:
    user = await result.fetch_one()
    # user is User instance
    print(user.name)  # Use attribute access

Error Handling¶

Check Error¶

async with engine.execute("SELECT * FROM non_existent_table") as result:
    if result.error:
        print(f"Error code: {result.error_no}")
        print(f"Error message: {result.error_msg}")
        print(f"Error object: {result.error}")
    else:
        data = await result.fetch_all()

Error Attributes¶

result.error: Error exception object (MySQLError or None)
result.error_no: Error code (integer, 0 when no error)
result.error_msg: Error message (string, empty string when no error)

Execution Statistics¶

async with engine.execute("INSERT INTO users (name) VALUES (%s)", ("张三",)) as result:
    # Affected rows
    print(f"Affected rows: {result.row_count}")

    # Last inserted ID (only for INSERT statements)
    print(f"Last inserted ID: {result.last_rowid}")

Note: - row_count returns None when using streaming query (stream=True) - row_count returns None when an error occurs - last_rowid is only valid for INSERT statements