Eluvia Base Django

Django Rest Framework

Library provides extension for the DRF framework in the package eluvia_base_django.rest_framework.

Rest fields/fieldsets

The classes eluvia_base_django.rest_framework.fieldsets.RestField and eluvia_base_django.rest_framework.fieldsets.RestFieldset are data structures for storing tree structure of the resource field names. For example if a rest endpoint returns this response:

{
  "id": "59f96d64-2344-437c-8f0c-50c0d72b1d80",
  "name": "Petr",
  "email": "petr@eluvia.com",
  "role": {
    "id": "64fcee15-3af6-4c7b-a55c-b8dbfdd7e930",
    "name": "capitan"
  }
}

The response fieldset are the keys of the response:

id,name,email,role(id,name)

The tree key format can be converted to the RestFieldset (RFS) with this way:

from eluvia_base_django.rest_framework.fieldsets import RFS, RF

fieldset = RFS.create_from_string('id,name,email,role(id,name)')

# It will create RFS(RF('id'), RF('name'), RF('email'), RF('role', RFS(RF('id'), RF('name'))))

Now you can work with fieldset with this way:

'id' in fieldset             # return True
fieldset['id']               # return FR('id')
fieldset['role'].fieldset    # return RFS(RF('id'), RF('name'))
del fieldset['id']           # remove first field from the fieldset
bool(fieldset)               # return True (False for empty fieldset)
str(fieldset)                # return 'id,name,email,role(id,name)'
fieldset.append('username')  # add username field to the fieldset
fieldset.flat()              # return ['id', 'name', 'email', 'role']

# iterate over fieldset
for field in fieldset:
    field

# Intersection of two fieldsets (return new fieldset with values 'id,role(id)')
fieldset & RFS.create_from_string('id,role(id),extra')
fieldset.intersection(RFS.create_from_string('id,role(id),extra'))

# Join two fieldsets (return new fieldset with values 'id,name,email,role(id,name),extra')
fieldset | RFS.create_from_string('id,role(id),extra')
fieldset.join(RFS.create_from_string('id,role(id),extra'))

Serializers

Eluvia base django provides extension of the model and base serializer with the classes:

• eluvia_base_django.rest_framework.serializers.Serializer
• eluvia_base_django.rest_framework.serializers.ModelSerializer.
• eluvia_base_django.rest_framework.serializers.ModelListSerializer.
• eluvia_base_django.rest_framework.serializers.LazySerializer.

exclude_fields_per_actions

Meta attribute exclude_fields_per_actions of the serializer classes defines which writable fields will be excluded for a specific view action (create/update/delete/partial_update). Example:

from rest_framework import serializers
from eluvia_base_django.rest_framework.serializers import Serializer


class UserSerializer(Serializer):
    
    id = serializers.UUIDField()
    name = serializers.CharField()
    
    class Meta:
        fields = ['id', 'name']
        exclude_fields_per_actions = {
            'update': ['id']  # We do not want to change ID with the object update for read and write
        }
        exclude_write_fields_per_actions = {
            'update': ['id']  # We do not want to change ID with the object update for input data
        }
        exclude_read_fields_per_actions = {
            'update': ['id']  # We do not want to change ID with the object update for output data
        }

default_read_fields

The fields which will be returned from the resource response can be defined on a user side with the fields query string. For example with the URL /users?fields=id the endpoint will return only {"id": "64fcee15-3af6-4c7b-a55c-b8dbfdd7e930"}. You can specify which fields will be returned by default with the default_read_fields parameter.

from rest_framework import serializers
from eluvia_base_django.rest_framework.serializers import Serializer


class UserSerializer(Serializer):
    
    id = serializers.UUIDField()
    name = serializers.CharField()
    
    class Meta:
        fields = ['id', 'name']
         # Only id will be returned by default but client may request name field with `?fields=id,name`
        default_read_fields = ['id']

Without default_read_fields the all serializer fields are returned. There are two special values __all__ and __default__ that can be used in the fields query string. The value __all__ means that all fields will be returned and __default__ means that default endpoint fields will be returned.

You can get RFS fieldset which will be returned to the client from the serializer with the method get_read_fieldset:

method_name

For the ModelSerializer and ModelListSerializer you can specify which serializer method will be used to get the value from the object. Example:

from rest_framework import serializers
from eluvia_base_django.rest_framework.serializers import ModelSerializer


class RoleSerializer(ModelSerializer):
    ref = serializers.CharField()
    label = serializers.CharField()

class UserSerializer(ModelSerializer):
    
    id = serializers.UUIDField()
    name = serializers.CharField()
    roles = RoleSerializer(many=True, method_name='get_roles')
    
    def get_roles(self, obj):
        # Return only active roles
        return obj.roles.filter(is_active=True)

auto_prefetch

ModelSerializer and ModelListSerializer automatically prefetch related fields. You can disable this behavior with the Meta parameter auto_prefetch=False.

# The serializers requires view and request in the context
serializer = UserSerializer(context={'request': request, 'view': view})
serializer.get_read_fieldset()

LazySerializer

Lazy serializer is serializer which data is loading with a lazy manner. The main purpose is get all data from external services with only one request. The ID of the external service objects are stored in the lazy evaluators and objects data will be loaded just right before rendering with the eluvia_base_django.rest_framework.renderers.LazyJSONRenderer. The renderer must be defined in the settings:

REST_FRAMEWORK = {
    ...
    'DEFAULT_RENDERER_CLASSES': [
        'eluvia_base_django.rest_framework.renderers.LazyJSONRenderer',
    ],
    ...
}

The lazy serializer usage is very simple. You just need to extend LazySerializer class and implement get_data_dict_by_ids methods:

# Models file
from django.db import models
import uuid


class Issue(models.Model):

    id = models.UUIDField(default=uuid.uuid4, primary_key=True)
    name = models.CharField(max_length=250)
    reporter_id = models.UUIDField()


# serializers file
from eluvia_auth import User
from eluvia_base_django.rest_framework.serializers import LazySerializer, ModelSerializer
from rest_framework import serializers


class LazyUserSerializer(LazySerializer):
    
    id = serializers.UUIDField()
    name = serializers.CharField()
    
    class Meta:
        fields = ['id', 'name']
    
    def get_data_dict_by_ids(self, ids, fields):
        # Return data from the external service
        return User.objects.fields(fields).filter(user_ids=ids).to_dict()


class IssueSerializer(ModelSerializer):
  
    id = serializers.UUIDField()
    name = serializers.CharField()
    reporter = LazyUserSerializer(source_pk_field='reporter_id')
    
    class Meta:
        model = Issue
        fields = ['id', 'name', 'reporter']

BulkModelListSerializer

BulkModelListSerializer is a DRF serializer that allows bulk creating and updating objects.

You only need to implement it into ModelSerializer Meta class as value of 'list_serializer_class' attribute. Then the ModelSerializer needs to be used in viewset as usual.

If you want to create multiple objects at once you need to put a list of objects in 'data' attribute and True in 'many' attribute when serializer instance is creating.

When updating multiple objects you need to do the same and put a queryset or list of objects in 'instance' attribute. You also may to set the 'source_field' attribute in serializer class. It is the name of the field that will be used as identifier of the objects (default is 'id').

# ViewSets file

ModelViewSet

A viewset that provides soft deleting.

It should be used by setting 'soft_delete_attribute_name' and 'soft_delete_attribute_value.

Another work with this viewset is the same as with standard ModelViewSet.

BulkModelViewSet

A viewset that provides default bulk_update(), bulk_partial_update(), bulk_destroy() and bulk_script() actions.

It should be used as a parent class of your viewset and ModelSerializer with defined BulkModelListSerializer as list_serializer_class should be set as serializer_class.

Another work with this viewset is the same as with standard ModelViewSet.

# Models file
from django.db import models
import uuid


class Issue(models.Model):

    id = models.UUIDField(default=uuid.uuid4, primary_key=True)
    name = models.CharField(max_length=250)
    reporter_id = models.UUIDField()


# serializers file
from django.db.models.functions import Concat
from django.db.models import F, Value, CharField
from eluvia_base_django.rest_framework.serializers import BulkModelListSerializer, ModelSerializer, BulkScriptSerializer
from rest_framework import serializers


class IssueBulkSerializer(BulkModelListSerializer):
  
    class Meta:
        bulk_lookup_field = 'id'
  
    def validate(self, attrs):
      # Validate data
      return attrs


class IssueSerializer(ModelSerializer):
  
    id = serializers.UUIDField()
    name = serializers.CharField()
    
    class Meta:
        model = Issue
        fields = ['id', 'name']

        
class IssueBulkScriptSerializer(BulkScriptSerializer):
  
    def parse_script(self, value):
        """
        Concat input value to the name field
        """
        
        return {
            'name': Concat(F('name'), Value(' '), Value(value), output_field=CharField())
        }
        

# views file
from eluvia_base_django.rest_framework.viewsets import BulkModelViewSet


class IssueViewSet(BulkModelViewSet):
    
    serializer_class = IssueSerializer
    bulk_serializer_class = IssueBulkSerializer
    bulk_script_serializer_class = IssueBulkScriptSerializer
    queryset = Issue.objects.all()

Filter backends

Ordering

The class eluvia_base_django.rest_framework.filters.OrderingFilter is extension of the original DRF OrderingFilter. Its logic is the same only query string name is order_by and input format is::

?order_by=field_name:(asc|desc)[,field_name:(asc|desc)]

The query string is validated. If the value is invalid or if field is not allowed to filter the 422 response is returned.

You can specify ordering_serializer_fields in the view to specify which serializer fields can be used for ordering.

Filters

Eluvia filter sets and filter backends are based on django-filter library. Therefore this library must be installed.

Filter backends

eluvia_base_django.rest_framework.filters.ComplexFilterBackend is filter backend derived from the django-filter framework. It is very similar to the django_filters.rest_framework.DjangoFilterBackend. The complex filter backend only supports more ways how to parse the input value for the filtersets.

• filter with every query string value where with no query string name duplicities (default django-filter format)

  ?term_a=value_a&term_b=value_1,value_2
  

• filter with duplicate keys

  ?term_a=value_a&term_b=value_1&term_b=value_2
  

• one filter query string which is fully validated (exception is raised for non-existent filter keys).

  ?filter=term_a="value_a"&term_b=["value_1", "value_2"]  # Both term_a and term_b must be valid filters
  

Filter sets

• eluvia_base_django.rest_framework.filters.BaseFilterSet - filter set for Django model queryset.
• eluvia_base_django.rest_framework.filters.NoModelFilterSet - filter set for non Django model objects. It is useful for filtering NoSql databases like ElasticSearch. Filterset supports only method filters.
• eluvia_base_django.rest_framework.filters.SchemaFilterSet - filter set for eluvia schema objects (REST queryset). Filterset supports only method filters.

Pagination

The list of paginators that can be used in the DRF views:

• eluvia_base_django.rest_framework.paginations.ListPagination - standard limit/offset pagination for list of objects.
• eluvia_base_django.rest_framework.paginations.ModelPagination - standard limit/offset pagination for Django model queryset.
• eluvia_base_django.rest_framework.paginations.SchemaPagination - standard limit/offset pagination for eluvia schema objects (REST queryset).
• eluvia_base_django.rest_framework.paginations.anysearch.AnySearchCursorBasedPagination - cursor pagination for search clients exposing a Search-like queryset (Elasticsearch or OpenSearch). Detects search_after support via duck-typing, so neither elasticsearch-dsl/elasticsearch nor opensearch-py needs to be installed to use it.

All of the above resolve their limit query parameter through LimitResolutionMixin, configurable with two Django settings:

• DEFAULT_PAGE_LIMIT (optional, default 20) - the limit used when the limit query parameter is missing or empty.
• MAX_PAGE_LIMIT (optional, default 1000) - the upper bound for the limit query parameter.

A limit above MAX_PAGE_LIMIT is rejected with a 422 Unprocessable Entity, with the error detail attached to the limit field, instead of being silently truncated. The same applies to a non-numeric or negative limit. A missing or empty limit still falls back to DEFAULT_PAGE_LIMIT. ListPagination (and ModelPagination/SchemaPagination) additionally treat limit=0 as a valid, deliberate value used to request only the meta block without any objects; AnySearchCursorBasedPagination rejects limit=0 the same way as any other invalid value.

Both settings can also be overridden per paginator instance via constructor arguments (ListPagination(default_limit=..., max_limit=...)), which take precedence over the Django settings and work outside of a DRF view as well. Each value is resolved in the same order: the constructor argument first, then the Django setting, then the default_limit/max_limit class attribute of the paginator class - so neither setting has to be defined for the paginators to work. The 20/1000 defaults match set_pagination(default_limit: 20) and Eluvia::Base::Config.max_page_limit in the Ruby eluvia-base library.

Passing max_limit=None explicitly - either as the constructor argument, as the MAX_PAGE_LIMIT setting, or as a max_limit = None class attribute on a subclass - removes the cap entirely: any numeric limit is then accepted as-is, however large. Omitting max_limit altogether (the default) keeps the 1000 cap; only an explicit None opts out of it, so this is never triggered by accident.

A separate opt-in flag, disable_errors=True (constructor argument or class attribute, default False), makes an invalid limit - non-numeric, negative, zero (for AnySearchCursorBasedPagination), or above max_limit - silently fall back to DEFAULT_PAGE_LIMIT instead of raising. This is independent of the cap itself: with the default max_limit (1000) and disable_errors=True, a limit above 1000 doesn't raise, but it's replaced by the default rather than honoured as-is.

Dramatiq

This library extends dramatiq (https://dramatiq.io/) library with some useful features.

Actor class

The eluvia_base_django.dramatiq.DjangoActor class adds possibility to run task with on commit signal to achieve that model instances will be stored in the database before the task is executed.

# tasks file
import dramatiq

from eluvia_base_django.dramatiq.actors import DjangoActor


@dramatiq.actor(actor_class=DjangoActor)
def my_task(user_id):
    user = User.objects.get(pk=user_id)

# call the task

my_task.send_with_options(args=(user.id,), on_commit=True)

Encoder

The eluvia_base_django.dramatiq.encoder.DjangoJSONEncoder class adds possibility to pass UUID, date and datetime as a task argument.

# Django settings file

DRAMATIQ_ENCODER = "eluvia_base_django.dramatiq.encoder.DjangoJSONEncoder"

# tasks file (no you can send UUID, date and datetime as a task argument)
import dramatiq

from eluvia_base_django.dramatiq.actors import DjangoActor


@dramatiq.actor(actor_class=DjangoActor)
def my_task(id: UUID, from: date, to: datetime):
    pass

Test helpers

The eluvia_base_django.dramatiq.tests module contains synchronous worker and test context manager for testing dramatiq tasks for a Django unit tests.

# django test settings file, you have to configure StubBroker

DRAMATIQ_BROKER = {
    "BROKER": "dramatiq.brokers.stub.StubBroker",
    "OPTIONS": {},
    "MIDDLEWARE": [
        "dramatiq.middleware.AgeLimit",
        "dramatiq.middleware.TimeLimit",
        "dramatiq.middleware.Callbacks",
        "dramatiq.middleware.Retries",
    ]
}

# tests file

from eluvia_base_django.dramatiq.tests import dramatiq_test_broker


class MyTestCase(TestCase):

    def test_run_task():
        with dramatiq_test_broker() as run_tasks:
            my_task.send_with_options(args=(user.id,))
            
            run_tasks()  # run all tasks

Tasks are executed in the thread of the test, therefore they see the data of the surrounding test transaction. Tasks sent by a running task are executed too and a delayed task (send_with_options(delay=...)) is executed right away.

Broker middleware is not applied, so a failing task raises its exception directly from run_tasks and is never retried, no matter how the middleware is configured.

Storage

Package eluvia_base_django.storage contains classes for storage management. Right now there is support for S3 storage which extends django-storages (https://django-storages.readthedocs.io/en/latest/) library. Usage:

# storages file

from pathlib import Path

from django.conf import settings
from django.core.files.storage import FileSystemStorage
from germanium.storage import TestInMemoryStorage

from eluvia_base_django.storages import BasePrivateS3Storage, BasePublicS3Storage, get_storage_class


class TestPrivateMediaInMemoryStorage(TestInMemoryStorage):
    filesystem_name = "media/private"


class PrivateFileSystemStorage(FileSystemStorage):
    location = Path(settings.BASE_DIR, "media", "private")


class PrivateS3Storage(BasePrivateS3Storage):
    bucket_name = settings.AWS_S3_PRIVATE_BUCKET


# get_storage_class returns the storage class according to STORAGE_TYPE (s3 will return PrivateS3Storage, 
# if tests are running the TestPrivateMediaInMemoryStorage is returned otherwise PrivateFileSystemStorage is returned)
PrivateMediaStorage = get_storage_class(PrivateS3Storage, PrivateFileSystemStorage, TestPrivateMediaInMemoryStorage)

private_storage = PrivateMediaStorage()


# Django settings file
STORAGE_TYPE = "S3"

# Set file storage
DEFAULT_FILE_STORAGE = "data_uploader.storages.PrivateMediaStorage"
if STORAGE_TYPE == "local":
    MEDIA_URL = "/media/"

Integration

Module eluvia_base_django.integration.cache contains helpers for API integration to cache API data in Django models.

eluvia_base_django.integration.cache.CacheModel

CacheModel is a Django model that can be used for caching API data. For its usage you need to create a model that extends CacheModel and define its fields and CacheMeta class which defines the cache configuration.

from eluvia_base_django.integration.cache import CacheModel
from eluvia_auth.schemas import User

class UserCache(CacheModel):
    id = models.UUIDField(default=uuid.uuid4, primary_key=True)
    created_at = models.DateTimeField()
    updated_at = models.DateTimeField()
    first_name = models.CharField(max_length=255, null=True)
    last_name = models.CharField(max_length=255, null=True)
    username = models.CharField(max_length=255, unique=True)
    email = models.CharField(max_length=255, null=True)

    class CacheMeta:
        schema = User  # API schema
        view_name = view_name  # name of the view if you want to use DB view as a cache model

Model can be related to other models with CacheForeignKey field.

from eluvia_base_django.integration.cache import CacheForeignKey

class MyModel(Model):
    user = CacheForeignKey(UserCache)
```