Created
May 12, 2021 14:54
-
-
Save banagale/fc98216a55e2271f41d404845efe1216 to your computer and use it in GitHub Desktop.
A DRF Serializer and APIView defining an endpoint for behavioral analytics events posted from a frontend application
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import logging | |
from rest_framework import serializers | |
from rest_framework.authentication import SessionAuthentication, TokenAuthentication | |
from rest_framework.permissions import IsAuthenticated | |
from rest_framework.response import Response | |
from rest_framework.views import APIView | |
from core.analytics.helpers import track_event | |
from frontend.api.event_constants import GATHERING_EVENT_ID_MAP | |
logger = logging.getLogger('main') | |
# noinspection PyAbstractClass | |
class CaptureGatheringEventSerializer(serializers.Serializer): | |
event_id = serializers.IntegerField(required=True, min_value=0) | |
def validate(self, data: dict): | |
# Validate event_id exists in event id to name tuple constants map | |
if len([i for i in GATHERING_EVENT_ID_MAP if i[0] == data.get('event_id')]) > 0: | |
return data | |
logger.warning('An improper event ID was sent for tracking') | |
# Do not provide useful response | |
raise serializers.ValidationError('Unexpected data received') | |
class CaptureGatheringEvent(APIView): | |
authentication_classes = (SessionAuthentication, TokenAuthentication) | |
permission_classes = [IsAuthenticated] | |
@staticmethod | |
def track_gathering_event(user_id: int, event_id: int): | |
# Get event name matching event id | |
event = [i for i in GATHERING_EVENT_ID_MAP if i[0] == event_id][0][1] | |
event_properties = {} | |
track_event(user_id, event, event_properties) | |
def post(self, request): | |
serializer = CaptureGatheringEventSerializer(data=request.data) | |
if serializer.is_valid(): | |
self.track_gathering_event(request.user.id, serializer.data.get('event_id')) | |
return Response(serializer.data) | |
else: | |
# Give no indication of expected data format | |
raise serializers.ValidationError("Bad request") |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment