Repositories

Repositories provide useful functions for Record interact with the database. They are also a Repository pattern implementation, as hinted by the naming. These constructs are quite handy in the segregation of the data access layer and the business logic or service layer in multi-layer applications, but can also be quite useful as a shortcut for common read, write or delete operations.

For more details on all the available methods, check the Repository class documentation.

Declaring Repositories

A Repository class can be instantiated directly, or declared via direct inheritance. Typically, direct inheritance has the advantage of resulting in a properly named class that can be easily extendable by adding methods. Regardless, there are usage scenarios where direct instantiation can be quite convenient:

from rick_db import fieldmapper, Repository
from rick_db.conn.pg import PgConnection

@fieldmapper(tablename='character', pk='id_character')
class Character:
    id = 'id_character'
    name = 'name'


# declare a Repository class for Character using inheritance
class CharacterRepository(Repository):

    def __init__(self, db):
        # new constructor
        super().__init__(db, Character)


db_cfg = {...}
conn = PgConnection(**db_cfg)

# instantiate declared repository class
repo = CharacterRepository(conn)

# insert some records
repo.insert(Character(id=1, name='Sarah Connor'))
repo.insert(Character(id=2, name='John Connor'))

# read all records using the repo object
for record in repo.fetch_all():
    print(record.name)

# alternative method, instantiate a repository directly
repo = Repository(conn, Character)

# use the repo object
repo.insert(Character(id=3, name='T-1000'))

Extending Repositories

Repository classes can also contain custom methods; This is the preferred approach when implementing additional database functionality. However, the methods should be designed to be stateless - they should not add or modify internal object attributes, nor depend on previous or future executions to carry their functions. By respecting this, Repository objects can be instantiated or reused in different contexts without any extra supervision:

from rick_db import fieldmapper, Repository
from rick_db.sql import Select, Literal


@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)

Advanced usage

While Repository objects are stateless from an operational perspective, there is actually an internal, thread-safe, global cache that is used to accelerate query building. Most Repository methods perform actions with SQL generated from the Query Builder, and these generated SQL statements can often be cached, due to the fact that any required values are passed in a separate structure.

The cache operations within the Repository internal scope can be performed by using the protected methods _cache_get(key) and _cache_set(key, value). The key parameter is a unique identifier for the specific query within the Repository, and will be internally concatenated with a database identifier, module and repository identifier.

The typical usage scenario for cache interaction is:

sql_string = _cache_get(string_method_name)
if sql_string is None:
   sql_string, values = some_query.assemble()
   _cache_set(string_method_name, sql_string)
else:
    values = list_of_required_values

(...)

The implementation of the actual Registry.fetch_pk() provides a good example of the typical pattern usage of the cache manipulation methods:

    (...)   

    def fetch_pk(self, pk_value) -> Optional[object]:

        if self._pk is None:
            raise RepositoryError("find_pk(): missing primary key in Record %s" % str(type(self._record)))

        # try to fetch a finished SQL string from the cache, ready to use
        # the cache key is the method name
        qry = self._cache_get('find_pk')
        if qry is None:
            # if no cache match, we need to build the SQL string using the query builder
            qry, values = self.select().where(self._pk, '=', pk_value).limit(1).assemble()
            # ...and store the generated SQL string in the cache, for future usage
            self._cache_set('find_pk', qry)
        else:
            # SQL string was successfully found in the cache, no need to use query builder
            # just build the required values list
            values = [pk_value]

        # execute SQL query and return the record           
        with self._db.cursor() as c:
            return c.fetchone(qry, values, self._record)

    (...)

Accessing/Purging the Repository Cache

If necessary, the cache used for the Repository SQL statements can be inspected or purged. The global variable is a rick_db.cache.StrCache instance, and can be accessed via rick_db.repository.query_cache:

from rick_db.repository import  query_cache

# purge all cached items
query_cache.purge()