Created v1 API.

Author: epou

.github/workflows/ci.yml

@@ -44,7 +44,7 @@ jobs:
 
       - name: Run Django Tests
         # Need to run as root because we are writing the results into the shared volume.
-        run: docker compose -f docker-compose-local.yml run -u root --rm django coverage run manage.py test --noinput
+        run: docker compose -f docker-compose-local.yml run -u root --rm django coverage run -m pytest
 
       - name: Create coverage XML report
         # Need to run as root because we are writing the results into the shared volume.

Dockerfile

@@ -37,6 +37,7 @@ RUN pip install numpy==1.23.1
 
 WORKDIR /app
 COPY . /app
+ENV PYTHONPATH "${PYTHONPATH}:/app/"
 
 RUN pip install -r requirements/dev.pip
 

api/__init__.py

@@ -0,0 +1 @@
+default_app_config = "api.apps.ApiConfig"

api/apps.py

@@ -0,0 +1,10 @@
+from django.apps import AppConfig
+
+
+class ApiConfig(AppConfig):
+    name = "api"
+    label = "api"
+
+    def ready(self) -> None:
+        import api.schema
+        import api.auth.schema

api/auth/__init__.py

@@ -0,0 +1,9 @@
+from rest_framework_simplejwt.authentication import JWTAuthentication
+
+from tigaserver_app.models import TigaUser
+
+
+class AppUserJWTAuthentication(JWTAuthentication):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.user_model = TigaUser

api/auth/authentication.py

@@ -0,0 +1,5 @@
+from drf_spectacular.contrib.rest_framework_simplejwt import SimpleJWTScheme
+
+
+class AppUserJWTAuthentication(SimpleJWTScheme):
+    target_class = "api.auth.authentication.AppUserJWTAuthentication"

api/auth/schema.py

@@ -0,0 +1,45 @@
+from django.contrib.auth import authenticate
+
+from rest_framework import exceptions, serializers
+from rest_framework_simplejwt.settings import api_settings
+from rest_framework_simplejwt.serializers import (
+    TokenObtainSerializer,
+    TokenObtainPairSerializer,
+)
+
+
+class AppUserTokenObtainSerializer(TokenObtainSerializer):
+    uuid = serializers.UUIDField(required=True)
+    device_token = serializers.CharField(allow_blank=True)
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+        del self.fields[self.username_field]
+        del self.fields["password"]
+
+    def validate(self, attrs):
+        authenticate_kwargs = {
+            "uuid": attrs["uuid"],
+            "device_token": attrs["device_token"],  # or None,
+        }
+        try:
+            authenticate_kwargs["request"] = self.context["request"]
+        except KeyError:
+            pass
+
+        self.user = authenticate(**authenticate_kwargs)
+
+        if not api_settings.USER_AUTHENTICATION_RULE(self.user):
+            raise exceptions.AuthenticationFailed(
+                self.error_messages["no_active_account"],
+                "no_active_account",
+            )
+
+        return {}
+
+
+class AppUserTokenObtainPairSerializer(
+    TokenObtainPairSerializer, AppUserTokenObtainSerializer
+):
+    pass

api/auth/serializers.py

@@ -0,0 +1,108 @@
+from django.db.models import Model
+from django.core.exceptions import ImproperlyConfigured
+
+from rest_framework import serializers
+from rest_framework.exceptions import ValidationError
+from rest_framework.fields import empty
+
+
+class FieldPolymorphicSerializer(serializers.Serializer):
+    field_value_serializer_mapping = None
+    resource_type_field_name = "type"
+    model = None
+
+    def __new__(cls, *args, **kwargs):
+        if cls.field_value_serializer_mapping is None:
+            raise ImproperlyConfigured(
+                "`{cls}` is missing a "
+                "`{cls}.field_value_serializer_mapping` attribute".format(
+                    cls=cls.__name__
+                )
+            )
+        if not isinstance(cls.resource_type_field_name, str):
+            raise ImproperlyConfigured(
+                "`{cls}.resource_type_field_name` must be a string".format(
+                    cls=cls.__name__
+                )
+            )
+        if not issubclass(cls.model, Model):
+            raise ImproperlyConfigured(
+                "`{cls}.model` must be a django Model".format(cls=cls.__name__)
+            )
+        return super().__new__(cls, *args, **kwargs)
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        field_value_serializer_mapping = self.field_value_serializer_mapping
+        self.field_value_serializer_mapping = {}
+        for field_value, serializer in field_value_serializer_mapping.items():
+            if callable(serializer):
+                serializer = serializer(*args, **kwargs)
+                serializer.parent = self
+
+            self.field_value_serializer_mapping[field_value] = serializer
+
+    def to_representation(self, instance):
+        serializer = self._get_serializer_for_instance(instance=instance)
+        return serializer.to_representation(instance)
+
+    def to_internal_value(self, data):
+        if self.instance:
+            serializer = self._get_serializer_for_instance(instance=self.instance)
+        else:
+            serializer = self._get_serializer_for_data(data=data)
+        return serializer.to_internal_value(data=data)
+
+    def create(self, validated_data):
+        serializer = self._get_serializer_for_data(data=validated_data)
+        return serializer.create(validated_data)
+
+    def update(self, instance, validated_data):
+        serializer = self._get_serializer_for_data(data=validated_data)
+        return serializer.update(instance, validated_data)
+
+    def is_valid(self, raise_exception=False):
+        valid = super().is_valid(raise_exception)
+
+        try:
+            if self.instance:
+                serializer = self._get_serializer_for_instance(instance=self.instance)
+            else:
+                serializer = self._get_serializer_for_data(data=self.initial_data)
+        except ValidationError:
+            child_valid = False
+        else:
+            child_valid = serializer.is_valid(raise_exception)
+            self._errors.update(serializer.errors)
+
+        return valid and child_valid
+
+    def run_validation(self, data=empty):
+        if self.instance:
+            serializer = self._get_serializer_for_instance(instance=self.instance)
+        else:
+            serializer = self._get_serializer_for_data(data=data)
+        return serializer.run_validation(data)
+
+    def _get_serializer_for_type(self, type_value):
+        try:
+            return self.field_value_serializer_mapping[type_value]
+        except KeyError:
+            raise ValidationError(
+                f"Value {type_value} has not serializer assigned for field '{self.resource_type_field_name}'"
+            )
+
+    def _get_serializer_for_instance(self, instance):
+        return self._get_serializer_for_type(
+            type_value=getattr(instance, self.resource_type_field_name)
+        )
+
+    def _get_serializer_for_data(self, data):
+        try:
+            return self._get_serializer_for_type(
+                type_value=data[self.resource_type_field_name]
+            )
+        except KeyError:
+            raise ValidationError(
+                f"{self.resource_type_field_name} is a required field."
+            )

api/base_serializers.py

@@ -0,0 +1,82 @@
+import pytz
+
+try:
+    import zoneinfo
+except ImportError:
+    from backports import zoneinfo
+
+from django.utils.dateparse import parse_datetime
+from django.core.exceptions import ValidationError
+from rest_framework import serializers
+
+from drf_extra_fields.geo_fields import PointField
+from timezone_field.utils import use_pytz_default
+from timezone_field.rest_framework import TimeZoneSerializerField
+
+
+class TimezoneAwareDateTimeField(serializers.DateTimeField):
+    def to_internal_value(self, value):
+        # Parse the datetime string into a datetime object
+        parsed_dt = parse_datetime(value)
+
+        # Ensure that the datetime contains timezone information
+        if parsed_dt.tzinfo is None:
+            raise ValidationError("Datetime value must include timezone information.")
+
+        return super().to_internal_value(value=value)
+
+
+class ExpandedPointField(PointField):
+    def __init__(self, *args, **kwargs):
+        super().__init__(str_points=True, *args, **kwargs)
+
+
+class IntegerDefaultField(serializers.IntegerField):
+    # When db value is None, use the default.
+    def get_attribute(self, instance):
+        attibute = super().get_attribute(instance)
+        if attibute is None and self.default != serializers.empty:
+            attibute = self.default
+        return attibute
+
+
+class TimeZoneSerializerChoiceField(TimeZoneSerializerField, serializers.ChoiceField):
+    def __init__(self, **kwargs):
+        self.use_pytz = kwargs.pop("use_pytz", use_pytz_default())
+
+        _tzstrs = (
+            pytz.common_timezones if self.use_pytz else zoneinfo.available_timezones()
+        )
+
+        super().__init__(choices=_tzstrs, html_cutoff=5, **kwargs)
+
+
+class WritableSerializerMethodField(serializers.SerializerMethodField):
+    # Reference: https://stackoverflow.com/a/64274128/8450576
+    def __init__(self, **kwargs):
+        self.setter_method_name = kwargs.pop("setter_method_name", None)
+        self.deserializer_field = kwargs.pop("deserializer_field")
+
+        super().__init__(**kwargs)
+
+        self.read_only = False
+        self.required = True
+
+    def bind(self, field_name, parent):
+        retval = super().bind(field_name, parent)
+        if not self.setter_method_name:
+            self.setter_method_name = f"set_{field_name}"
+
+        return retval
+
+    def get_default(self):
+        default = super().get_default()
+
+        return {self.field_name: default}
+
+    def to_internal_value(self, data):
+        value = self.deserializer_field.to_internal_value(data)
+        method = getattr(self.parent, self.setter_method_name)
+        return {
+            self.field_name: self.deserializer_field.to_internal_value(method(value))
+        }