PEP 249 Compliant¶

Turu defines a simple protocol based on PEP 249.

All Connection and Cursor class are derived from those protcols.

This tool is a simple wrapper for PEP 249 that allows you to specify and inspect the return type with the API you are familiar with.

In addition to execute/executemany, we provide execute_map/executemany_map with processing to map query results to a specified type.

import pydantic
import turu.sqlite3


class User(pydantic.BaseModel):
    id: int
    name: str


connection = turu.sqlite3.connect(":memory:")

with connection.cursor() as cursor:
    user = cursor.execute_map(User, "SELECT 1, 'taro'").fetchone()

    assert user == User(id=1, name="taro")

Note

typing.NamedDict/dataclasses.dataclass/pydantic.BaseModel are available as types to map.

But, for type validation, we reccomend using pydantic.

Connection Pool¶

Todo

Connection Pool may be implemented in the future, but has not yet been started.

Reasons:

Turu is developed for use in data analysis workflows and has no incentive to pool connections.
Caching strategies exist at different layers, such as RDS Proxy.
DbApi3 is not yet in PEP.

Connection¶

Connection can execute directly, but it is only reading the cursor internally.

with connection.execute("select 1") as cursor:
    ...

Equivalent to the following.

with connection.cursor().execute("select 1") as cursor:
    ...

It is recommended that you use this in conjunction with the with syntax and always close the cursor after using it.

Cursor¶

Always assign a new Cursor as the return value of the execute* method in order to update a type hint that can be retrieved by Cursor.

import pydantic
import turu.sqlite3
from typing_extensions import Never


class Row(pydantic.BaseModel):
    id: int
    name: str


connection = turu.sqlite3.connect(":memory:")

cursor1: turu.sqlite3.Cursor[Never] = connection.cursor()
cursor2: turu.sqlite3.Cursor[Row] = cursor1.execute_map(
    Row, "SELECT :id, :name", {"id": 1, "name": "taro"}
)