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.
Tuple-query APIs (query_multiple and query_multiple_async) apply this validation to the complete tuple: the
placeholders of every query are scanned and every referenced value is resolved before a cursor is acquired or the
first query executes, so a missing parameter in any query means no query reaches the database. This is client-side
prevalidation, not transaction atomicity; after validation succeeds, a later query can still fail at runtime after
earlier queries have executed.
One Statement Per Call🔗
A pydapper command executes exactly one SQL statement. SQL containing more than one statement raises
MultipleStatementsError before a cursor is acquired, so nothing reaches the driver — the refusal is the same on every
adapter regardless of what that adapter's DBAPI would have done with the text.
A single trailing ; followed only by whitespace and comments terminates one statement and is accepted. Anything else
after a top-level ; — including a second ; — is more than one statement and is refused.
Detection is the same lexical scan pydapper already uses to find ?name? placeholders, so a ; inside single-quoted
text, double-quoted text, a -- line comment, or a /* ... */ block comment is ordinary text and does not trip the
guard. It is not a SQL parser and has no dialect awareness: a top-level ; inside a PostgreSQL dollar-quoted body
(DO $$ ... ; ... $$) or an Oracle anonymous PL/SQL block (BEGIN ... ; ... END;) is refused too.
Known limits🔗
Because it is a lexer and not a parser, the guard is not a security boundary on its own — it is a guard rail. It keys
on a top-level ;, and its notion of "literal" and "comment" is one fixed ANSI-flavored set rather than each server's
own grammar. Two consequences worth knowing:
- A backslash before a closing quote currently hides a separator. The scanner treats
\as a string escape, which is MySQL's rule but not PostgreSQL's, SQLite's, Oracle's, or SQL Server's. On those,select 'a\'; drop table tis genuinely two statements and the guard does not catch it. This is the most important of the known gaps and is the headline item of #590. - SQL Server does not need a
;at all.select 1 as a insert into t values (1)is two statements in one T-SQL batch with no separator, and nothing lexical can see that. Detecting it needs a real parser, which pydapper deliberately does not have. - Dialect-specific literal forms are not recognized yet, so a
;inside a MySQL backtick identifier, a T-SQL bracket identifier, a MySQL#comment, or some BigQuery triple-quoted strings is currently misread — in either direction. Also tracked in #590.
Parameter binding, not this guard, is what actually protects you from injection: pydapper never formats a value into the SQL string.
Run scripts, PL/SQL blocks, and anything else that genuinely needs more than one statement against the DBAPI connection directly:
with pydapper.connect(dsn) as commands:
cursor = commands.connection.cursor()
cursor.execute(plsql_block)
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',
),
],
),
]
"""