dbdaora 0.22.6

Communicates with databases using repository pattern and service patterns

dbdaora

Communicates with NoSQL (and SQL for future) databases using repository and service patterns and python dataclasses

---

Documentation: https://dutradda.github.io/dbdaora/

Source Code: https://github.com/dutradda/dbdaora

---

Key Features

• Creates an optional service layer with cache and circuit breaker

• Supports for redis data types:

  • Hash
  • Sorted Set
  • (Others data types are planned)

• Backup redis data into other databases:

  • Google Datastore
  • Mongodb (soon)
  • SQL databases with SQLAlchemy (soon)
  • (Others data bases are planned)

• Support for other databases are in development.

Requirements

• Python 3.8+

• jsondaora for data validation/parsing

• circuitbreaker

• cachetools

• Optionals:

  • aioredis
  • google-cloud-datastore

Instalation

$ pip install dbdaora

Simple redis hash example

import asyncio
from dataclasses import dataclass

from dbdaora import (
    DictFallbackDataSource,
    DictMemoryDataSource,
    HashRepository,
)


@dataclass
class Person:
    id: str
    name: str
    age: int


def make_person(name: str, age: int) -> Person:
    return Person(name.replace(' ', '_').lower(), name, age)


class PersonRepository(HashRepository[str]):
    entity_cls = Person
    key_attrs = ('id',)


repository = PersonRepository(
    memory_data_source=DictMemoryDataSource(),
    fallback_data_source=DictFallbackDataSource(),
    expire_time=60,
)
person = make_person('John Doe', 33)
asyncio.run(repository.add(person))

geted_person = asyncio.run(repository.query(person.id).entity)
print(geted_person)

Person(id='john_doe', name='John Doe', age=33)

Simple redis sorted set example

import asyncio

from dbdaora import (
    DictFallbackDataSource,
    DictMemoryDataSource,
    SortedSetEntity,
    SortedSetRepository,
)


class PlaylistRepository(SortedSetRepository[str]):
    ...


repository = PlaylistRepository(
    memory_data_source=DictMemoryDataSource(),
    fallback_data_source=DictFallbackDataSource(),
    expire_time=60,
)
values = [('m1', 1), ('m2', 2), ('m3', 3)]
playlist = SortedSetEntity(id='my_plalist', values=values)
asyncio.run(repository.add(playlist))

geted_playlist = asyncio.run(repository.query(playlist.id).entity)
print(geted_playlist)

SortedSetEntity(id='my_plalist', values=[b'm1', b'm2', b'm3'], max_size=None)

Using the service layer

The service layer uses the backup dataset when redis is offline, opening a circuit breaker.

It has an optional cache system too.

import asyncio
from dataclasses import dataclass

from dbdaora import (
    DictFallbackDataSource,
    DictMemoryDataSource,
    HashRepository,
    make_hash_service,
)


@dataclass
class Person:
    id: str
    name: str
    age: int


def make_person(name: str, age: int) -> Person:
    return Person(name.replace(' ', '_').lower(), name, age)


class PersonRepository(HashRepository[str], entity_cls=Person):
    ...


async def make_memory_data_source() -> DictMemoryDataSource:
    return DictMemoryDataSource()


async def make_fallback_data_source() -> DictFallbackDataSource:
    return DictFallbackDataSource()


service = asyncio.run(
    make_hash_service(
        PersonRepository,
        memory_data_source_factory=make_memory_data_source,
        fallback_data_source_factory=make_fallback_data_source,
        repository_expire_time=60,
    )
)
person = make_person('John Doe', 33)
asyncio.run(service.add(person))

geted_person = asyncio.run(service.get_one(person.id))
print(geted_person)

Person(id='john_doe', name='John Doe', age=33)