SQL database layer
Project description
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, SqlLite3;
- Pluggable SQL query profiler;
- Simple migration manager for SQL files;
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
Documentation
Project documentation can be found on the Documentation website.
TL;DR; example
Showcasing the Connection, DTO, Repository and Query Builder objects:
from rick_db import fieldmapper, Repository
from rick_db.conn.pg import PgConnection
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: author id
: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'
}
conn = PgConnection(**db_cfg)
repo = AuthorRepository(conn)
dump_author_rating(repo)
Running tests
To run the tests, you should have both tox and tox-docker, as well as a local docker daemon. Make sure the current user has access to the docker daemon.
$ pip3 install -r requirements-dev.txt
$ tox
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.