Pymnesia

Pymnesia provides with a real in memory database and ORM to be used in unit tests and in staging environment when persistence adapters are not available yet. This tool is likely to be used within decoupled architectures.

Overview

The current version is beta, but the project is stable and offers a wide range of features that can already be used in your projects :

• Declare entities with various field types (For a complete list, please refer to the table below.)
• Save entities in an in memory database
• Commit and rollback
• Query stored entities using a very lightweight api and intuitive syntax

Fields support

	Supported for declaration	Supported for query
int	yes ☑	yes ☑
float	yes ☑	yes ☑
str	yes ☑	yes ☑
bool	yes ☑	yes ☑
date	not officially	not officially
datetime	not officially	not officially
tuple	yes ☑	not officially
list	no ☒	no ☒
dict	not officially	no ☒
one to one relation	yes ☑	yes ☑
one to many relation	yes ☑	yes ☑
many to one relation	no ☒	no ☒
many to many relation	no ☒	no ☒

not officially supported: The feature should work but may not be fully functional (querying for instance) and stability is not guarantied.

Basic user documentation

Until a more detailed documentation is released, please refer to the examples below for usage.

Please keep in mind that the features will usually be made available through an api import. Core classes, functions and such should not be imported and used directly.

Entities

Declaring a simple entity

from uuid import UUID

from pymnesia.api.entities.base import declarative_base


class CarEngine(declarative_base()):
    __tablename__ = "car_engines"

    id: UUID

    horsepower: int

Declaring an entity with a 'one to one' relation

from uuid import UUID

from pymnesia.api.entities.base import declarative_base


class Car(declarative_base()):
    __tablename__ = "cars"

    id: UUID

    name: str

    engine: CarEngine

Declaring an entity with a 'one to many' relation

The relation api can be used to specify custom options (for now the reverse name and whether the relation is optional or not).

from uuid import UUID

from pymnesia.api.entities.base import declarative_base
from pymnesia.api.entities.fields import field, relation


class Car(declarative_base()):
    __tablename__ = "cars"

    id: UUID

    name: str = field(default="Peugeot 3008")

    engine: CarEngine

    drivers: List[Driver] = relation(reverse="a_list_of_drivers")

Command

Save and commit

from uuid import uuid4

from pymnesia.api.unit_of_work import uow
from pymnesia.api.command import transaction

unit_of_work = uow()
new_transaction = transaction(unit_of_work=unit_of_work)

v12_engine = CarEngine(id=uuid4(), horsepower=400)
aston_martin = CarModel(id=uuid4(), engine_id=v12_engine.id)

unit_of_work.save_entity(entity=v12_engine)
unit_of_work.save_entity(entity=aston_martin)
new_transaction.commit()

Querying the database for car models will return one car model (The output will be 400).

for car in unit_of_work.query().cars().fetch():
    print(car.engine.horsepower)

Rollback

from uuid import uuid4

from pymnesia.api.unit_of_work import uow
from pymnesia.api.command import transaction

unit_of_work = uow()
new_transaction = transaction(unit_of_work=unit_of_work)

v12_engine = CarEngine(id=uuid4(), horsepower=400)
unit_of_work.save_entity(entity=v12_engine)
new_transaction.rollback()

v8_engine = CarEngine(id=uuid4(), horsepower=300)
unit_of_work.save_entity(entity=v8_engine)
new_transaction.commit()

Querying the database for car engines will return the v8 engine alone (The output will be 300).

for engine in unit_of_work.query().car_engines().fetch():
    print(engine.horsepower)

Query

Fetch

Fetch allows to retrieve multiple results of a query. To query an entity model, call query() in the unit of work instance containing your entities. Then simply call a method using the tablename you declared for said entity.

For instance, if you declare the entity below:

from uuid import UUID

from pymnesia.api.entities.base import declarative_base


class Address(declarative_base()):
    __tablename__ = "addresses"

    id: UUID

    street_name: str

You will query addresses as follows:

for car in unit_of_work.query().addresses().fetch():
    # whatever you need to do
    pass

Fetch one

Fetch allows to retrieve the first result of a query.

Given you have two cars in your in memory database, fetch_one() will return the entity that was saved first.

car = unit_of_work.query().addresses().fetch_one()

Where Or clauses

One of the great features of Pymnesia is how you can add where or clauses to you queries.

Where clause

To add a where clause use the where method when querying an entity.

'Equal' operator

The query below will return every car engine that has a 400 horsepower:

for car in unit_of_work.query().car_engines().where({"horsepower": 400}).fetch():
    # whatever you need to do
    pass

'Not equal' operator

The query below will return every car engine that doesn't have a 400 horsepower:

for car in unit_of_work.query().car_engines().where({"horsepower::not": 400}).fetch():
    # whatever you need to do
    pass

'Greater than' operator

The query below will return every car engine that have horsepower greater than 400:

for car in unit_of_work.query().car_engines().where({"horsepower::gt": 400}).fetch():
    # whatever you need to do
    pass

'Greater than or equal to' operator

The query below will return every car engine that have horsepower greater than or equal to 400:

for car in unit_of_work.query().car_engines().where({"horsepower::gte": 400}).fetch():
    # whatever you need to do
    pass

'Less than' operator

The query below will return every car engine that have horsepower lesser than 400:

for car in unit_of_work.query().car_engines().where({"horsepower::lt": 400}).fetch():
    # whatever you need to do
    pass

'Less than or equal to' operator

The query below will return every car engine that have horsepower lesser than or equal to 400:

for car in unit_of_work.query().car_engines().where({"horsepower::lte": 400}).fetch():
    # whatever you need to do
    pass

'Match' operator

The query below will return every car that have a name matching the provided regex:

for car in unit_of_work.query().cars().where({"name::match": r'^Peugeot .*$'}).fetch():
    # whatever you need to do
    pass

'In' operator

The query below will return every car that have a value included in the provided list:

for car in unit_of_work.query().cars().where({"name::in": ["Aston Martin Valkyrie", "Porsche GT3"]}).fetch():
    # whatever you need to do
    pass

Relational queries

Every operator documented above can be used to make relational queries:

for car in unit_of_work.query().cars().where({"engine.horsepower::gt": 400}).fetch():
    # whatever you need to do
    pass

Or clauses

You can add one or more 'or clauses' to a query. Every condition in a 'or clause' is evaluated as OR AND.

For instance the query below:

query = unit_of_work.query().cars().where({"name": "Peugeot 3008"})
query.or_({"name::match": r'^Peugeot .*$', "engine.horsepower::gt": 100})

Is the equivalent of an SQL query:

SELECT *
FROM cars
         JOIN car_engines ON car_engines.id = cars.engine_id
WHERE cars.name = 'Peugeot 3008'
   OR (cars.name LIKE '^Peugeot .*$' AND car_engines.horsepower > 100)

Multiple 'or clauses' remain independent of one another:

query = unit_of_work.query().cars().where({"name": "Peugeot 3008"})
query.or_({"name::match": r'^Peugeot .*$', "engine.horsepower::gt": 100})
query.or_({"engine.horsepower::gte": 200})

Is the equivalent of an SQL query:

SELECT *
FROM cars
         JOIN car_engines ON car_engines.id = cars.engine_id
WHERE cars.name = 'Peugeot 3008'
   OR (cars.name LIKE '^Peugeot .*$' AND car_engines.horsepower > 100)
   OR (car_engines.horsepower >= 200)

Where clause using composition

The entities can be queried using composition rather than declarative conditions. The example below makes little sense, but this feature can be powerful to make complex queries when operator functions are not available to perform the requested operation.

from typing import Iterable
from functools import partial


def car_name_func(entities_: Iterable, field: str, value: str, *args, **kwargs) -> filter:
    return filter(lambda e: getattr(e, field) == value, entities_)


partial_k2000_func = partial(
    car_name_func,
    field="name",
    value="K2000",
)

partial_gran_torino_func = partial(
    car_name_func,
    field="name",
    value="Gran Torino",
)

query = unit_of_work.query().cars().where_with_composition([
    partial_k2000_func,
    partial_gran_torino_func,
])

Limit

You can limit the number of result using the limit() method.

The query below will limit the number of results to 5 car engines:

for car in unit_of_work.query().car_engines().limit(5).fetch():
    # whatever you need to do
    pass

Order by

You can order by your results by specifying a direction and a key.

The query below will order the results on the field 'name' in descending order.

for car in unit_of_work.query().cars().order_by(direction="desc", key="name").fetch():
    # whatever you need to do
    pass