query_async
query_async can execute a query and serialize the results to a model.
Parameters🔗
| name | type | description | optional | default |
|---|---|---|---|---|
| sql | str |
the sql query str to execute |  |
|
| params | ParamType |
params to substitute in the query |  |
None |
| model | Any |
the callable to serialize the model; callable must be able to accept column names as kwargs. | |
dict |
| mapper | Callable[[RawRow], Any] |
callable that receives a RawRow and returns a projected value. Mutually exclusive with model. |
|
None |
| buffered | bool |
whether to buffer reading the results of the query | |
True |
| options | CommandOptions or None |
command execution options; see Command options | |
None |
param= remains accepted as a 1.x compatibility alias for params=. Pass only one of the two names.
Parameter Shapes🔗
params accepts one parameter record: a mapping, mapping subclass, mutable mapping, or object/dataclass with attributes
matching the placeholder names. Top-level list params are only for execute and execute_async; read and scalar methods
raise InvalidParameterShapeException for top-level lists before opening a cursor.
params=None, param=None, or omitting both names means there is no parameter object. If the SQL contains pydapper
placeholders such as ?id?, every referenced placeholder must be supplied or pydapper raises
MissingParameterException before calling the DBAPI. An empty mapping is a real parameter record with no keys. A list
inside one parameter record, such as {"ids": []} or {"ids": [1, 2, 3]}, is one value and is reserved for future IN
list expansion support.
Rows🔗
When rows are returned as dictionaries, they are insertion-ordered dict[str, Any] values. Key order follows the column order reported by the DB-API cursor.
Column names must be unique exactly as the driver reports them. If a result includes duplicate names, pydapper raises DuplicateColumnException with columns, duplicate_columns, and duplicate_indexes attributes. Alias joined columns instead of using ambiguous select * joins.
Missing keys on returned dict rows raise normal Python KeyError. When custom model construction is requested with model= or models=, pydapper uses the same column-name keyword argument mapping path for results with unique column names.
Use mapper= when column-name mapping is too restrictive. The mapper receives a RawRow with columns, values, and as_dict(). RawRow preserves duplicate column names and positional values in cursor order, so mapper functions can intentionally project joined rows, nested objects, aliases, or duplicate names. Positional indexing and slicing read from values; name indexing and RawRow.as_dict() require unique column names and raise DuplicateColumnException when a dict would be ambiguous.
Cardinality🔗
- 0 rows with unique column names and
buffered=True: returns an empty list. - 0 rows with unique column names and
buffered=False: returns an async generator that yields no rows.
Example - Serialize to a dataclass🔗
The raw sql query can be executed using the query_async method and map the results to a list of dataclasses.
import asyncio
import datetime
from dataclasses import dataclass
from pydapper import connect_async
@dataclass
class Task:
id: int
description: str
due_date: datetime.date
owner_id: int
async def main():
async with connect_async() as commands:
data = await commands.query_async("select * from task limit 1", model=Task)
print(data)
# [Task(id=1, description='Set up a test database', due_date=datetime.date(2021, 12, 31), owner_id=1)]
asyncio.run(main())
Example - Serialize a one to one relationship🔗
You can get creative with what you pass in to the model kwarg of query
import asyncio
import datetime
from dataclasses import dataclass
from pydapper import connect_async
@dataclass
class Owner:
id: int
name: str
@dataclass
class Task:
id: int
description: str
due_date: datetime.date
owner: Owner
@classmethod
def from_query_row(cls, id, description, due_date, owner_id, owner_name):
return cls(id, description, due_date, Owner(owner_id, owner_name))
query = """
select t.id, t.description, t.due_date, o.id as owner_id, o.name as owner_name
from task t join owner o on t.owner_id = o.id
limit 1
"""
async def main():
async with connect_async() as commands:
data = await commands.query_async(query, model=Task.from_query_row)
print(data)
"""
[
Task(
id=1,
description="Set up a test database",
due_date=datetime.date(2021, 12, 31),
owner=Owner(id=1, name="Zach Schumacher"),
)
]
"""
asyncio.run(main())
Example - Project joined rows with duplicate column names🔗
Use mapper= when a join intentionally selects duplicate column names or when projection logic needs positional values.
import asyncio
from dataclasses import dataclass
from pydapper import RawRow
from pydapper import connect_async
@dataclass
class Owner:
id: int
name: str
@dataclass
class TaskWithOwner:
id: int
description: str
owner: Owner
def to_task_with_owner(row: RawRow) -> TaskWithOwner:
return TaskWithOwner(
id=row.values[0],
description=row.values[1],
owner=Owner(id=row.values[2], name=row.values[3]),
)
query = """
select
t.id,
t.description,
o.id,
o.name
from task t
join owner o on t.owner_id = o.id
limit 1
"""
async def main():
async with connect_async() as commands:
data = await commands.query_async(query, mapper=to_task_with_owner)
print(data)
# [TaskWithOwner(id=1, description='Set up a test database', owner=Owner(id=1, name='Zach Schumacher'))]
asyncio.run(main())
Example - Project aliased rows by name🔗
RawRow.as_dict() and row["column_name"] are available when the referenced column names are unique.
import asyncio
from typing import Any
from pydapper import RawRow
from pydapper import connect_async
def to_summary(row: RawRow) -> dict[str, Any]:
values = row.as_dict()
return {
"task_id": values["task_id"],
"owner_name": row["owner_name"],
}
query = """
select
t.id as task_id,
o.name as owner_name
from task t
join owner o on t.owner_id = o.id
limit 1
"""
async def main():
async with connect_async() as commands:
data = await commands.query_async(query, mapper=to_summary)
print(data)
# [{'task_id': 1, 'owner_name': 'Zach Schumacher'}]
asyncio.run(main())
Example - Buffering queries🔗
By default, query_async fetches all results and stores them in a list (buffered). By setting buffered=False, you can
instead have query_async return an async generator that fetches one record from the result set at a time. This may be
useful if querying a large amount of data that would not fit into memory, but note that this keeps both the connection
and cursor open while you're retrieving results. Breaking out of a plain async generator does not by itself guarantee
immediate cleanup while the generator remains referenced, so explicitly close it when stopping early.
rows = await db.query_async(sql, buffered=False)
try:
async for row in rows:
break
finally:
await rows.aclose()
import asyncio
from pydapper import connect_async
async def main():
async with connect_async() as commands:
rows = await commands.query_async("select * from task", buffered=False)
print(type(rows))
# <class 'async_generator'>
try:
async for row in rows:
print(row)
break
finally:
await rows.aclose()
asyncio.run(main())
Example - Serializing a one-to-many relationship🔗
Using model is nice for simple serialization, but more complex serializations might require more complex logic. In this case, it is recommended to return an unbuffered result and serialize it as you iterate. See the example below:
import asyncio
import typing
from dataclasses import dataclass
from pydapper import connect_async
@dataclass
class Owner:
id: int
name: str
tasks: typing.List["Task"]
@dataclass
class Task:
id: int
description: str
query = """
select task.id as task_id,
owner.id as owner_id,
owner.name as owner_name,
task.description as description
from owner
join task on owner.id = task.owner_id
"""
async def main():
async with connect_async() as commands:
owners = dict()
async for record in await commands.query_async(query, buffered=False):
if (owner_id := record["owner_id"]) not in owners:
owners[owner_id] = Owner(id=owner_id, name=record["owner_name"], tasks=list())
owners[owner_id].tasks.append(Task(id=record["task_id"], description=record["description"]))
print(list(owners.values()))
"""
[
Owner(
id=1,
name='Zach Schumacher',
tasks=[
Task(
id=1,
description='Set up a test database',
),
Task(
id=2,
description='Seed the test database',
),
Task(
id=3,
description='Run the test suite',
),
],
),
]
"""
asyncio.run(main())