rick_db - Simple SQL database layer

rick_db is a simple SQL database layer for Python3. It includes connection management, Object Mapper, Query Builder, and a Repository pattern implementation. It is not an ORM, and it's not meant to replace one.

Features

• Object Mapper;
• Fluent Sql Query builder;
• High level connectors for PostgreSQL, SQLite3, ClickHouse;
• SQL query builder dialect for MySQL;
• Pluggable SQL query profiler;
• Simple migration manager for SQL files;
• Forward-only Keyset pagination using SQL row-value comparison, with custom base queries;

Note: SQLite may have different behaviour based on Python versions; notably, DDL statements in a transaction may not be affected by rollback on Python <3.12. Also, there are limitations on text search when using the Grid helper

Note: rick_db version >=2.0.0 are not backwards compatible with 1.x versions; Code changes are required to migrate between versions; see the documentation for further details

Usage scenarios

rick_db was built to cater to a schema-first approach: Database schema is built and managed directly with SQL DDL commands, and the application layer has no responsibility on the structure of the database.

Installation

$ pip3 install rick-db

To use the PostgreSQL backend, install with one of the optional extras to pull in a psycopg2 driver:

$ pip3 install rick-db[pgsql]         # builds psycopg2 from source
$ pip3 install rick-db[pgsql-binary]  # prebuilt psycopg2-binary wheel

Documentation

Project documentation can be found on the Documentation website, including the Keyset pagination guide.

TL;DR; example

Showcasing the Connection, DTO, Repository and Query Builder objects:

from rick_db import fieldmapper, Repository
from rick_db.backend.pg import PgConnectionPool
from rick_db.sql import Select, Literal


@fieldmapper(tablename='publisher', pk='id_publisher')
class Publisher:
    id = 'id_publisher'
    name = 'name'


@fieldmapper(tablename='book', pk='id_book')
class Book:
    id = 'id_book'
    title = 'title'
    total_pages = 'total_pages'
    rating = 'rating'
    isbn = 'isbn'
    published = 'published_date'
    fk_publisher = 'fk_publisher'


@fieldmapper(tablename='author', pk='id_author')
class Author:
    id = 'id_author'
    first_name = 'first_name'
    middle_name = 'middle_name'
    last_name = 'last_name'


@fieldmapper(tablename='book_author', pk='id_book_author')
class BookAuthor:
    id = 'id_book_author'
    fk_book = 'fk_book'
    fk_author = 'fk_author'


class AuthorRepository(Repository):

    def __init__(self, db):
        super().__init__(db, Author)

    def calc_avg_rating(self, id_author: int):
        """
        Calculate average rating for a given author
        :param id_author:
        :return: average rating, if any
        """

        # generated query:
        # SELECT avg(rating) AS "rating" FROM "book" INNER JOIN "book_author" ON
        # "book"."id_book"="book_author"."fk_book" WHERE ("fk_author" = %s)
        qry = Select(self.dialect). \
            from_(Book, {Literal("avg({})".format(Book.rating)): 'rating'}). \
            join(BookAuthor, BookAuthor.fk_book, Book, Book.id). \
            where(BookAuthor.fk_author, '=', id_author)

        # retrieve result as list of type Book (to get the rating field)
        rset = self.fetch(qry, cls=Book)
        if len(rset) > 0:
            return rset.pop(0).rating
        return 0

    def books(self, id_author: int) -> list[Book]:
        """
        Retrieve all books for the given author
        :return: list[Book]
        """

        qry = Select(self.dialect). \
            from_(Book). \
            join(BookAuthor, BookAuthor.fk_book, Book, Book.id). \
            where(BookAuthor.fk_author, '=', id_author)

        return self.fetch(qry, cls=Book)


def dump_author_rating(repo: AuthorRepository):
    for author in repo.fetch_all():

        # calculate average
        rating = repo.calc_avg_rating(author.id)

        # print book list
        print("Books by {firstname} {lastname}:".format(firstname=author.first_name, lastname=author.last_name))
        for book in repo.books(author.id):
            print(book.title)

        # print average rating
        print("Average rating for {firstname} {lastname} is {rating}".
              format(firstname=author.first_name, lastname=author.last_name, rating=rating))


if __name__ == '__main__':
    db_cfg = {
        'dbname': "rickdb-bookstore",
        'user': "rickdb_user",
        'password': "rickdb_pass",
        'host': "localhost",
        'port': 5432,
        'sslmode': 'require'
    }

    pool = PgConnectionPool(**db_cfg)
    repo = AuthorRepository(pool)
    dump_author_rating(repo)

Running tests

To run the tests, you need a local docker daemon and the current user must have access to it. The PostgreSQL and ClickHouse containers are started and stopped automatically by testcontainers — no manual setup or environment variables are required.

$ pip3 install -r requirements-dev.txt

# run the whole suite directly
$ pytest

# or across all supported Python versions
$ tox