Skip to main content

Simple Typed Python Database Client based PEP 249

Project description

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 is especially difficult to automatically test cloud-based SQL databases 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

  • :rocket: Simple - Turu is a simple database api wrapper of PEP 249.
  • :bulb: Type Hint - Full support for type hints.
  • :zap: Async/Await - Async/Await supports.
  • :test_tube: 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.execute_map(Row, "select 1, 'a'") as cursor:
    assert cursor.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.execute_map(Row, "select 1, 'a'") as cursor:
        assert cursor.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.execute_map(Row, "select 1, 'a'"),
        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

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

turu-0.9.8.tar.gz (3.3 kB view hashes)

Uploaded Source

Built Distribution

turu-0.9.8-py3-none-any.whl (2.9 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page