sqlite3 adapter for turu


Keywords
async, bigquery, pep249, postgres, postgresql, postgresql-database, python, snowflake, snowflakedb, sqlite, sqlite-database, sqlite3, sqlite3-database, typed-python
License
MIT
Install
pip install turu-sqlite3==0.10.0

Documentation

Turu: Simple Database Client for Typed Python

docs test pypi package

logo


Documentation: https://yassun7010.github.io/turu-py/

Source Code: https://github.com/yassun7010/turu-py


Installation

pip install "turu[snowflake]"

Why Turu?

SQL is a powerful language, but it has many dialects, and Cloud Native Databases are especially difficult to test automatically in a local environment.

Turu was developed as a simple tool to assist local development. It provides a simple interface according to PEP 249 โ€“ Python Database API Specification v2.0 and allows for easy recording of query results and injection mock data.

Features

  • ๐Ÿš€ Simple - Turu is a simple database api wrapper of PEP 249.
  • ๐Ÿ’ก Type Hint - Full support for type hints.
  • โšก Async/Await - Async/Await supports.
  • ๐Ÿงช Recoed and Mock - Record and mock database queries for testing.

Supprted Database

Database Sync Support Async Support Installation
SQLite3 Yes Yes pip install "turu[sqlite3]"
MySQL Yes Yes pip install "turu[mysql]"
PostgreSQL Yes Yes pip install "turu[postgres]"
Snowflake Yes Yes pip install "turu[snowflake]"
BigQuery Yes No pip install "turu[bigquery]"

Usage

Basic Usage

from pydantic import BaseModel


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

connection = turu.sqlite3.connect("test.db")

with connection.cursor() as cursor:
    assert cursor.execute_map(Row, "select 1, 'a'").fetchone() == Row(id=1, name="a")

Testing

import turu.sqlite3

from pydantic import BaseModel


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

expected1 = [Row(id=1, name="a"), Row(id=2, name="b")]
expected2 = [Row(id=3, name="c"), Row(id=4, name="d")]
expected3 = [Row(id=5, name="e"), Row(id=6, name="f")]

connection = turu.sqlite3.MockConnection()

(
    connection.chain()
    .inject_response(Row, expected1)
    .inject_response(Row, expected2)
    .inject_response(Row, expected3)
)

for expected in [expected1, expected2, expected3]:
    with connection.cursor() as cursor:
        assert cursor.execute_map(Row, "select 1, 'a'").fetchall() == expected

Recording and Testing

Your Production Code

import os

import turu.sqlite3
from turu.core.record import record_to_csv

from your_package.data import RECORD_DIR
from your_package.schema import Row


def do_something(connection: turu.sqlite3.Connection):
    with record_to_csv(
        RECORD_DIR / "test.csv",
        connection.curosr(),
        enable=os.environ.get("ENABLE_RECORDING"),
        limit=100,
    ) as cursor:
        ... # Your logic

Your Test Code

import turu.sqlite3

from your_package.data import RECORD_DIR
from your_package.schema import Row


def test_do_something(connection: turu.sqlite3.MockConnection):
    connection.inject_response_from_csv(Row, RECORD_DIR / "test.csv")

    assert do_something(connection) is None