fractal-specifications 2.2.2

Fractal Specifications is an implementation of the specification pattern for building SOLID logic for your Python applications.

Fractal Specifications

Fractal Specifications is an implementation of the specification pattern for building SOLID logic for your Python applications.

Installation

pip install fractal-specifications

Background

This project comes with an article on Medium, which sets out what the specification pattern is, what the benefits are and how it can be used.

Development

Setup the development environment by running:

make deps
pre-commit install

Happy coding.

Occasionally you can run:

make lint

This is not explicitly necessary because the git hook does the same thing.

Do not disable the git hooks upon commit!

Usage

Specifications can be used to encapsulate business rules. An example specification is EqualsSpecification("maximum_speed", 25).

A specification implements the is_satisfied_by(obj) function that returns True or False, depending on the state of the obj that is passed into the function as parameter. In our example, the obj needs to provide the attribute maximum_speed.

Full code example

This example includes a repository to show an application of specifications.

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import List

from fractal_specifications.generic.operators import EqualsSpecification
from fractal_specifications.generic.specification import Specification


@dataclass
class Road:
    maximum_speed: int

    @staticmethod
    def slow_roads_specification() -> Specification:
        return EqualsSpecification("maximum_speed", 25)


class RoadRepository(ABC):
    @abstractmethod
    def get_all(self, specification: Specification) -> List[Road]:
        ...

    def slow_roads(self) -> List[Road]:
        return self.get_all(Road.slow_roads_specification())


class PythonListRoadRepository(RoadRepository):
    def __init__(self, roads: List[Road]):
        self.roads = roads

    def get_all(self, specification: Specification) -> List[Road]:
        return [
            road for road in self.roads
            if specification.is_satisfied_by(road)
        ]


if __name__ == "__main__":
    road_repository = PythonListRoadRepository([
        Road(maximum_speed=25),
        Road(maximum_speed=50),
        Road(maximum_speed=80),
        Road(maximum_speed=100),
    ])

    print(road_repository.slow_roads())

Contrib

This library also comes with some additional helpers to integrate the specifications easier with existing backends, such as the Django ORM.

Django

Specifications can easily be converted to (basic) Django ORM filters with DjangoOrmSpecificationBuilder.
Using this contrib package requires django to be installed.

Query support:

• Direct model fields field=value
• Indirect model fields field__sub_field=value
  • Implies recursive subfields field__sub_field__sub_sub_field=value
  • This holds for all operators below as well
• Equals field=value or __exact
• Less than __lt
• Less than equal __lte
• Greater than __gt
• Greater than equal __gte
• In __in
• And Q((field_a=value_a) & (field_b=value_b))
• Or Q((field_a=value_a) | (field_b=value_b))
• Partial regex __regex=r".* value .*"
• Full regex __regex
• Contains regex __contains
• Is null __isnull

from abc import ABC, abstractmethod
from django.db import models
from typing import List

from fractal_specifications.contrib.django.specifications import DjangoOrmSpecificationBuilder
from fractal_specifications.generic.operators import EqualsSpecification
from fractal_specifications.generic.specification import Specification


class Road(models.Model):
    maximum_speed = models.IntegerField()

    @staticmethod
    def slow_roads_specification() -> Specification:
        return EqualsSpecification("maximum_speed", 25)


class RoadRepository(ABC):
    @abstractmethod
    def get_all(self, specification: Specification) -> List[Road]:
        ...

    def slow_roads(self) -> List[Road]:
        return self.get_all(Road.slow_roads_specification())


class DjangoRoadRepository(RoadRepository):
    def get_all(self, specification: Specification) -> List[Road]:
        if q := DjangoOrmSpecificationBuilder.build(specification):
            return Road.objects.filter(q)
        return Road.objects.all()


if __name__ == "__main__":
    road_repository = DjangoRoadRepository()

    print(road_repository.slow_roads())

You could of course also skip the repository in between and do the filtering directly:

from fractal_specifications.contrib.django.specifications import DjangoOrmSpecificationBuilder

q = DjangoOrmSpecificationBuilder.build(Road.slow_roads_specification())
Road.objects.filter(q)

SQLAlchemy

Query support:

• Direct model fields {field: value}
• And {field: value, field2: value2}
• Or [{field: value}, {field2: value2}]

from fractal_specifications.contrib.sqlalchemy.specifications import SqlAlchemyOrmSpecificationBuilder

q = SqlAlchemyOrmSpecificationBuilder.build(specification)

Elasticsearch

Using this contrib package requires elasticsearch to be installed.

Query support:

• Exact term match (Equals) {"match": {"%s.keyword" % field: value}}
• String searches (In) {"query_string": {"default_field": field, "query": value}}
• And {"bool": {"must": [...]}}
• Or {"bool": {"should": [...]}}
• Less than {"bool": {"filter": [{"range": {field: {"lt": value}}}]}}
• Less than equal {"bool": {"filter": [{"range": {field: {"lte": value}}}]}}
• Greater than {"bool": {"filter": [{"range": {field: {"gt": value}}}]}}
• Greater than equal {"bool": {"filter": [{"range": {field: {"gte": value}}}]}}

from elasticsearch import Elasticsearch
from fractal_specifications.contrib.elasticsearch.specifications import ElasticSpecificationBuilder

q = ElasticSpecificationBuilder.build(specification)
Elasticsearch(...).search(body={"query": q})

Google Firestore

Query support:

• Equals (field, "==", value)
• And [(field, "==", value), (field2, "==", value2)]
• Contains (field, "array-contains", value)
• In (field, "in", value)
• Less than (field, "<", value)
• Less than equal (field, "<=", value)
• Greater than (field, ">", value)
• Greater than equal (field, ">=", value)

from fractal_specifications.contrib.google_firestore.specifications import FirestoreSpecificationBuilder

q = FirestoreSpecificationBuilder.build(specification)

Mongo

Query support:

• Equals {field: {"$eq": value}}
• And {"$and": [{field: {"$eq": value}}, {field2: {"$eq": value2}}]}
• Or {"or": [{field: {"$eq": value}}, {field2: {"$eq": value2}}]}
• In {field: {"$in": value}}
• Less than {field: {"$lt": value}}
• Less than equal {field: {"$lte": value}}
• Greater than {field: {"$gt": value}}
• Greater than equal {field: {"$gte": value}}
• Regex string match {field: {"$regex": ".*%s.*" % value}}

from fractal_specifications.contrib.mongo.specifications import MongoSpecificationBuilder

q = MongoSpecificationBuilder.build(specification)

Pandas

Query support:

• Equals df[field] == value
• And df[field] == value & df[field2] == value2
• Or df[field] == value | df[field2] == value2
• In df[field].isin[value]
• Less than df[field] < value
• Less than equal df[field] <= value
• Greater than df[field] > value
• Greater than equal df[field] >= value
• Is null df[field].isna()

Filtering on columns:

import pandas as pd

from fractal_specifications.contrib.pandas.specifications import PandasSpecificationBuilder
from fractal_specifications.generic.operators import EqualsSpecification, IsNoneSpecification


df = pd.DataFrame(
    {
        "id": [1, 2, 3, 4],
        "name": ["aa", "bb", "cc", "dd"],
        "field": ["x", "y", "z", None],
    }
)

print(df)
#    id name field
# 0   1   aa     x
# 1   2   bb     y
# 2   3   cc     z
# 3   4   dd  None


specification = EqualsSpecification("id", 4)
f1 = PandasSpecificationBuilder.build(specification)

print(f1(df))
#    id name field
# 3   4   dd  None


specification = IsNoneSpecification("field")
f2 = PandasSpecificationBuilder.build(specification)

print(f2(df))
#    id name field
# 3   4   dd  None


print(df.pipe(f1).pipe(f2))
#    id name field
# 3   4   dd  None


specification = EqualsSpecification("id", 4) & IsNoneSpecification("field")
f3 = PandasSpecificationBuilder.build(specification)

print(f3(df))
#    id name field
# 3   4   dd  None

Filtering on indexes:

import pandas as pd

from fractal_specifications.contrib.pandas.specifications import PandasIndexSpecificationBuilder
from fractal_specifications.generic.operators import EqualsSpecification, GreaterThanSpecification


df = pd.DataFrame({"month": [1, 4, 7, 10],
                   "year": [2012, 2014, 2013, 2014],
                   "sale": [55, 40, 84, 31]})
df = df.set_index("month")

print(df)
#        year  sale
# month
# 1      2012    55
# 4      2014    40
# 7      2013    84
# 10     2014    31

specification = EqualsSpecification("month", 4)
f1 = PandasIndexSpecificationBuilder.build(specification)

print(f1(df))
#        year  sale
# month
# 4      2014    40


df = df.reset_index()
df = df.set_index("year")

specification = GreaterThanSpecification("year", 2013)
f2 = PandasIndexSpecificationBuilder.build(specification)

print(f2(df))
#       month  sale
# year
# 2014      4    40
# 2014     10    31


df = df.reset_index()
df = df.set_index(["month", "year"])

print(df.pipe(f1).pipe(f2))
#             sale
# month year
# 4     2014    40


specification = EqualsSpecification("month", 4) & GreaterThanSpecification("year", 2013)
f3 = PandasIndexSpecificationBuilder.build(specification)

print(f3(df))
#             sale
# month year
# 4     2014    40