query
query 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 a generator that yields no rows.
Example - Serialize to a dataclass🔗
The raw sql query can be executed using the query method and map the results to a list of dataclasses.
import datetime
from dataclasses import dataclass
from pydapper import connect
@dataclass
class Task:
id: int
description: str
due_date: datetime.date
owner_id: int
with connect() as commands:
data = commands.query("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)]
Example - Serialize a one-to-one relationship🔗
You can get creative with what you pass in to the model kwarg of query
import datetime
from dataclasses import dataclass
from pydapper import connect
@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
"""
with connect() as commands:
data = commands.query(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"),
)
]
"""
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.
from dataclasses import dataclass
from pydapper import RawRow
from pydapper import connect
@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
"""
with connect() as commands:
data = commands.query(query, mapper=to_task_with_owner)
print(data)
# [TaskWithOwner(id=1, description='Set up a test database', owner=Owner(id=1, name='Zach Schumacher'))]
Example - Project aliased rows by name🔗
RawRow.as_dict() and row["column_name"] are available when the referenced column names are unique.
from typing import Any
from pydapper import RawRow
from pydapper import connect
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
"""
with connect() as commands:
data = commands.query(query, mapper=to_summary)
print(data)
# [{'task_id': 1, 'owner_name': 'Zach Schumacher'}]
Example - Buffering queries🔗
By default, query fetches all results and stores them in a list (buffered). By setting buffered=False, you can
instead have query act as a generator function, fetching 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 generator does not by itself guarantee immediate
cleanup while the generator remains referenced, so explicitly close it when stopping early.
rows = db.query(sql, buffered=False)
try:
for row in rows:
break
finally:
rows.close()
from pydapper import connect
with connect() as commands:
rows = commands.query("select * from task", buffered=False)
print(type(rows))
# <class 'generator'>
try:
for row in rows:
print(row)
break
finally:
rows.close()
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 typing
from dataclasses import dataclass
import pydapper
@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
"""
with pydapper.connect() as commands:
owners = dict()
for record in commands.query(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',
),
],
),
]
"""