Django Rest Framework - Utils

A set of util functions used in EBS Projects

Install:

pip install drf_util

Functions

Get value from an object by path

Definition:

gt(obj, path, default=None, sep='.')

Usage:

>>> data = {"a":{"b": 1}}
>>> print(gt(data, 'a.b'))
1

Get recursive values from a dict by keys

Definition:

get_object_labels(obj, path, default=None)

Usage:

>>> data = {"a": {"b": 'title'}, "c": 'test'}
>>> print(get_object_labels(data))
['title', 'test']

>>> data = {"a": {"b": 'title'}, "c": 'test'}
>>> print(get_object_labels(data, ['c']))
['test']

map() alternative with chunk select

Definition:

fetch_objects(instance, function, select=50)

Usage:

>>> def print_name(obj):
        print obj.name
>>>
>>> fetch_objects(UserBigList.objects.order_by('id'), print_name, 500)

Select a first true value

Definition:

any_value(items: list)

Usage:

>>> print(any_value('', None, 0, "Some text", 50000))
Some text

Recursive merge two dict

Definition:

dict_merge(a, b, path=None)

Usage:

>>> a = {'a': {'c': 1, 'd': {'x': 1}}}
>>> b = {'a': {'e': 1, 'd': {'y': 1}}}
>>> print(dict_merge(a, b))
{'a': {'c': 1, 'e': 1, 'd': {'x': 1, 'y': 1}}}

Iterate big query

Definition:

iterate_query(queryset, offset_field='pk', offset_start=0, limit=100)

Usage:

queryset = Thing.objects.all()
for _ in utils.iterate_query(queryset, 'id', 0):
    ...

Get applications from folder

Definition:

get_applications(base_folder='apps', inside_file='', only_directory=True)

Usage:

# settings.py
APPS_PATH = 'path_to_aps'  # default is apps
...

# any file
get_applications()  # ['path_to_aps.app1', 'path_to_aps.app2']
get_applications(inside_file='models.py',
                 only_directory=False)  # ['path_to_aps.app1.models', 'path_to_aps.app2.models']

Tricks:

# settings.py
INSTALLED_APPS = get_applications()
...

# urls.py
urlpatterns = [
    path("", include(application_urls))
    
    for application_urls in get_applications(
        inside_file='urls.py', only_directory=False
    ))
]

Prefetch and select related by serializer

Definition:

add_related(queryset, serializer)

Usage:

queryset = add_related(Thing.objects.all(), ThingSerializer)

Compare dicts

Definition:

dict_diff(a, b)

Usage:

>>> dict_diff({'a': 2, 'b': {'c': 2, 'd': 1}}, {'b': {'c': 3, 'd': 1}})
({'a': 2, 'b': {'c': 2}}, {'a': None, 'b': {'c': 3}})

Tighten dicts

Definition:

dict_normalise(data, separator='__')

Usage:

>>> dict_normalise({'a': 1, 'b': {'c': 2, 'd': {'e': 3, 'f': 3}}})
{'a': 1, 'b__c': 2, 'b__d__e': 3, 'b__d__f': 3}

Decorators

serialize_decorator

Definition:

serialize_decorator(serializer_method, preview_function=None, read_params=False)

Usage:

class RestoreUserPassword(GenericAPIView):
    @serialize_decorator(RestoreUserSerializer)
    def post(self, request, *args, **kwargs):
        return Response({"valid": True})

await_process_decorator

Decorator for creating a queue for using a function, it is needed to adjust the call of a function from different processes (Сelery, Threads). For example, this decorator can be used to limit the number of requests in the parser.

Definition:

# rate : count of usage some function, by default it's 20 times
# period : period of usage some function,  by default it's 1 minute
await_process_decorator(rate=20, period=60)

Usage:

@await_process_decorator(rate=10, period=5)  # 10 times per 5 seconds 
def simple_print(text):
    print(text)

Managers

• NoDeleteManager
• BaseManager

BaseManager

Features:

• fetch_only_annotation - make annotation only on iteration

Simple annotation:

queryset = User.objects.annotate(
    contacts_count=Count('contacts')
)

print(queryset.count())
# SELECT COUNT(*) from users join contacts on contacts.owner = user.id;

print(queryset.first())
# SELECT *, COUNT(contacts) as contacts_count from users join contacts on contacts.owner = user.id;

queryset = User.objects.fetch_only_annotation(
    contacts_count=Count('contacts')
)

print(queryset.count())
# SELECT COUNT(*) from users;

print(queryset.first())
# SELECT *, COUNT(contacts) as contacts_count from users join contacts on contacts.owner = user.id;

Models

BaseModel - with created_at and updated_at

class Thing(BaseModel):
    title = models.CharField(max_length=20)

    class Meta:
        db_table = 'another_things'

• CommonModel - with date_created and date_updated
• NoDeleteModel - with date_deleted
• AbstractJsonModel - with languages

Validators

• ObjectExistValidator - check if object exists
• ObjectUniqueValidator - check if object not exists
• PhoneValidator - check phone

Serializers

BaseModelSerializer - simple serializer for BaseModel class

class ThingSerializer(BaseModelSerializer):
    class Meta(BaseModelSerializer.Meta):
        model = Thing

ElasticFilterSerializer - make easy conversion between serializer data and elastic filters

class TenderFilterSerializer(PaginatorSerializer, ElasticFilterSerializer):
    sort_criteria = [{"date_updated": {"order": "desc"}}, "_score"]

    status = StringListField(required=False)
    date_start = serializers.DateField(required=False)
    date_end = serializers.DateField(required=False)

    def filter_status(self, value):
        return {'terms': {
            'search_status.keyword': value
        }}

    def filter_date_start(self, value):
        return {
            "range": {
                "tenderPeriod.startDate": {'gte': value}
            }
        }

    def filter_date_end(self, value):
        return {
            "range": {
                "tenderPeriod.startDate": {'lte': value}
            }
        }


class TenderListView(GenericAPIView):
    @serialize_decorator(TenderFilterSerializer)
    def get(self, request, *args, **kwargs):
        return Response(es_app.search_response(request.serializer, 'tenders_index'))

FilterSerializer - filter queryset by serializer fields

class ServiceListQuerySerializer(FilterSerializer):
    name = CharField(required=False)
    tag_id = CharField(required=False)
    type = CharField(required=False)
    status = CharField(required=False)

    def filter_name(self, value, queryset):
        return queryset.filter(name__icontains=value)

    def filter_tag_id(self, value, queryset):
        return queryset.filter(tags__contains=value)

    def filter_type(self, value, queryset):
        return queryset.filter(type=value)

    def filter_status(self, value, queryset):
        return queryset.filter(status=value)


class ServiceListView(ListAPIView):
    serializer_class = ServiceListQuerySerializer

    @swagger_auto_schema(query_serializer=ServiceListQuerySerializer)
    @serialize_decorator(ServiceListQuerySerializer)
    def get(self, request):
        services = request.serializer.get_filter(request.valid, Service.objects.all())
        return Response(ServiceSerializer(instance=services, many=True).data)

ChangebleSerializer - metamorphic serializer

class ContractNoticeCancelView(GenericAPIView):
    def put(self, request):
        serializer_meta = {
            'id': PrimaryKeyRelatedField(queryset=Tender.objects.all(), required=True),
            'info': {
                'rationale': CharField(required=True),
                'description': CharField(required=True),
            },
            'documents': DocumentFileSerializer(required=True, many=True)
        }
        serializer = ChangebleSerializer(data=request.data)
        serializer.update_properties(serializer_meta)
        serializer.is_valid(raise_exception=True)

        return Response({"valid": True})

PaginatorSerializer - serializer for paginating

class ListUserNotification(GenericAPIView):
    @serialize_decorator(PaginatorSerializer)
    def get(self, request):
        notifications = NotificationEvent.objects.filter(user=request.user)
        return request.serializer.response(notifications, serializer=ListNotificationSerializer)

Another serializers

• StringListField - simple string list of chars
• EmptySerializer - simple empty serializer
• IdSerializer - simple id serializer
• ReturnSuccessSerializer - simple success, message serializer

Serializers functions

build_model_serializer - build serializer with Inheritance

Definition:

build_model_serializer(base=BaseModelSerializer, add_bases=True, **kwargs)

Usage:

ThingSerializer = build_model_serializer(
    meta_model=Thing,
)

CreateThingSerializer = build_model_serializer(
    ThingSerializer,
    meta_fields=('name', 'desctiption')
)

CreateThingSerializer = build_model_serializer(
    ThingSerializer,
    meta_fields=('name', 'desctiption')  # 'id', 'created_at' and 'updated_at' is added automatically
)

ShortThingSerializer = build_model_serializer(
    ThingSerializer,
    meta_fields=('name', 'desctiption'),
    add_bases=False  # so as not to add 'id', 'created_at' and 'updated_at'
)


AnotherThingSerializer = build_model_serializer(
    things=ThingSerializer(many=True),
    meta_model=AnotherThing,
)

---

Note: Parameters with prefix 'meta_' is added to the meta class, the rest are added in the serializer class

Views

Note: for them to work, set in swagger settings DEFAULT_AUTO_SCHEMA_CLASS=drf_util.mixins.CustomAutoSchema

BaseModelViewSet

Usage:

class ThingViewSet(BaseModelViewSet):
    queryset = Thing.objects.all()
    serializer_class = ThingSerializer

Attributes:

queryset = None  # QuerySet

query_serializer = None  # Serializer for query
serializer_class = None  # Default and response serializer
serializer_create_class = None  # Body serializer
serializer_by_action = {}  # Serializer by action {[action]: [serializer]}

pagination_class = CustomPagination  # Pagination

filter_backends = (filters.OrderingFilter, CustomFilterBackend, filters.SearchFilter,)  # Filter backends
filter_class = None  # FilterSet
search_fields = ()  # Fields for search query_param
ordering_fields = '__all__'  # Fields for ordering query_param
ordering = ['-id']  # Default ordering fields

permission_classes_by_action = {"default": [IsAuthenticated]}  # Permission class by action {[action]: [permissions]}

Another views

• BaseViewSet
• BaseCreateModelMixin
• BaseUpdateModelMixin
• BaseListModelMixin
• BaseReadOnlyViewSet
• BaseModelItemViewSet
• BaseModelViewSet

Pagination

CustomPagination

Declaration:

class CustomPagination(PageNumberPagination):
    page = DEFAULT_PAGE
    page_size = 10
    page_size_query_param = 'per_page'

    def get_paginated_response(self, data):
        custom_paginator = dict(
            count=self.page.paginator.count, # noqa
            total_pages=self.page.paginator.num_pages, # noqa
            per_page=int(self.request.GET.get('per_page', self.page_size)),
            current_page=int(self.request.GET.get('page', DEFAULT_PAGE)), results=data
        )
        return Response(custom_paginator)

Tests

CustomClient - client which check response for status code

Usage:

class BaseTestCase(TestCase):
    client_class = CustomClient
    base_view = 'things'
    
    def test_list(self) -> None:
        self.client.get(reverse(f'{self.base_view}-list'))
        
    def test_duplicate(self):
        self.client.post(
            reverse(f'{self.base_view}-duplicate', args=(test_instance.pk,)),
            assert_status_code=status.HTTP_200_OK
        ).json()

BaseTestCase - test case with custom client

Usage:

class ViewsTestCase(BaseTestCase, TestCase):
    def test_swagger(self):
        response = self.client.get('/swagger/?format=openapi').json()
        self.assertEqual(len(response['schemes']), 2)

---

Note: Default setUp function authenticates the user

CRUDTestCase - test case with crud

Usage:

class ThingCRUDTestCase(CRUDTestCase, TestCase):
    fixtures = ['tests/fixtures.json']
    base_view = 'things'
    queryset = Thing.objects.all()
    fake_data = {
        'title': 'Thing name'
    }

Middlewares

PrintSQlMiddleware - middleware to print sql request and their statistics

Usage:

MIDDLEWARE = [
    'drf_util.middlewares.PrintSQlMiddleware',
    ...
]

Swagger utils

CustomAutoSchema - render schema with custom serializers methods

Usage:

SWAGGER_SETTINGS = {
    'DEFAULT_AUTO_SCHEMA_CLASS': 'drf_util.mixins.CustomAutoSchema'
    ...
}

get_custom_schema_view - function to get swagger with HTTP and HTTPS

Declaration:

get_custom_schema_view(title, default_version='v1', description='', *args, **kwargs)

Usage:

schema_view = get_custom_schema_view(
    title="API Documentation",
    description="This is API Documentation"
)

urlpatterns = [
    path("", schema_view.with_ui('swagger', cache_timeout=0), name='schema-swagger-ui'),
    path("redoc", schema_view.with_ui('redoc', cache_timeout=0), name='schema-redoc'),    
]

Lookups

Like

Usage:

queryset = User.objects.filter(first_name__like='%ia').values('first_name')

# SQL: SELECT name FROM users WHERE name LIKE '%ia';
# queryset: <QuerySet [{'first_name': 'Olivia'}, {'first_name': 'Amelia'}, '...(remaining elements truncated)...']>

Not

Usage:

queryset = queryset.objects.filter(first_name__not='Doe').values_list('first_name', flat=True)

# SQL: SELECT name FROM users WHERE NOT name = 'Doe';
# queryset: <QuerySet [{'first_name': 'Not Doe'}, {'first_name': 'Realy not Doe'}, '...(remaining elements truncated)...']>

Distinct

Usage:

queryset = queryset.objects.filter(first_name__distinct='Doe').values_list('first_name', flat=True)

# SQL: SELECT name FROM users WHERE NOT name IS DISTINCT FROM 'Doe';
# queryset: <QuerySet [{'first_name': 'Not Doe'}, {'first_name': 'Realy not Doe'}, '...(remaining elements truncated)...']>

Functions

StringToArr

Usage:

queryset = User.objects.annotate(split_name=StringToArr('name', Value(' ')).values('name', 'split_name')

# SQL: SELECT name FROM users WHERE name LIKE '%ia';
# queryset: <QuerySet [{'name': 'Olivia Marchel', 'split_name': ['Olivia', 'Marchel']}, {'name': 'Amelia Hust', 'split_name': ['Amelia', 'Hust']}, '...(remaining elements truncated)...']>

ArrIntersection

Usage:

queryset = User.objects.annotate(teams_intersection=ArrIntersection('featured_teams', 'teams')).values('featured_teams', 'teams', 'teams_intersection')

# SQL: SELECT name FROM users WHERE name LIKE '%ia';
# queryset: <QuerySet [{'featured_teams': ['Team1', 'Team2'], 'teams': ['Team1', 'Team3'], 'teams_intersection': ['Team1']}, {'featured_teams': ['Team1', 'Team2'], 'teams': ['Team3', 'Team4'], 'teams_intersection': []}, '...(remaining elements truncated)...']>

ArrayLength

Usage:

queryset = User.objects.annotate(tokens_count=ArrayLength('tokens')).values('tokens', 'tokens_count')

# SQL: SELECT name FROM users WHERE name LIKE '%ia';
# queryset: <QuerySet [{'tokens': ['tok1', 'tok2'], 'tokens_count': 2}, {'tokens': [], 'tokens_count': 0}, '...(remaining elements truncated)...']>

Exceptions

ValidationException

Usage:

value = -3

if value < 0:
    raise ValidationException("Invalid value.")

# Status code: 400
# Body: {"detail": "Invalid value."} 

FailedDependency

Usage:

response = request.get(url)

if not response.ok:
    raise FailedDependency(response.content)

# Status code: 424

IMATeapot

Usage:

try:
    number = 13 / 0
except Exception:
    raise IMATeapot()

# Status code: 418