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
param	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
buffered	bool	whether to buffer reading the results of the query		True

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)]

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

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"),
    )
]
"""

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

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.

from pydapper import connect

with connect() as commands:
    data = commands.query("select * from task", buffered=False)
    print(type(data))
    # <class 'generator'>

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

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',
            ),
        ],
    ),
]
"""

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