query_multiple

query_multiple can execute multiple queries with the same cursor and serialize the results. This method will throw a ValueError if you don't supply the same number of queries and models, or if a tuple of mapper functions does not match the number of queries. A single mapper function applies to every result set.

Parameters

All command methods also accept keyword-only options=; see Command options.

name	type	description	optional	default
queries	tuple[str, ...]	the sql query strings to execute in order
params	ParamType	params to substitute in the query		None
models	tuple[Any, ...]	callables to serialize each result set; each callable must accept column names as kwargs.		dict
mapper	Callable[[RawRow], Any] or tuple	one mapper for every result set, or a tuple of mapper functions. Mutually exclusive with models.		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.

Example

Query two tables and return the serialized results.

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_id: int


with connect() as commands:
    task, owner = commands.query_multiple(
        ("select * from task limit 1", "select * from owner limit 1"), models=(Task, Owner)
    )

print(task)
# [Task(id=1, description='Set up a test database', due_date=datetime.date(2021, 12, 31), owner_id=1)]
print(owner)
# [Owner(id=1, name='Zach Schumacher')]

(This script is complete, it should run "as is")

Example - Mapper Functions

Project each result set with RawRow mapper functions.

from pydapper import RawRow
from pydapper import connect


def to_task_description(row: RawRow) -> str:
    return row["description"]


def to_owner_name(row: RawRow) -> str:
    return row["name"]


with connect() as commands:
    task_descriptions, owner_names = commands.query_multiple(
        ("select description from task limit 1", "select name from owner limit 1"),
        mapper=(to_task_description, to_owner_name),
    )

print(task_descriptions)
# ['Set up a test database']
print(owner_names)
# ['Zach Schumacher']

(This script is complete, it should run "as is")