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}}}

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

Models

• 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

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