✨ Modernize bit bit

- bumps
- poetry
- linting

dpdapi/admin.py

@@ -5,9 +5,9 @@ from dpdapi.models import Alias, Domain, User
 
 @admin.register(Alias)
 class AliasAdmin(admin.ModelAdmin):
-    list_display = ['source', 'destination', 'active', 'domain']
-    list_filter = ['domain']
-    search_fields = ['source', 'destination']
+    list_display = ["source", "destination", "active", "domain"]
+    list_filter = ["domain"]
+    search_fields = ["source", "destination"]
 
 
 @admin.register(Domain)

dpdapi/filters.py

@@ -7,10 +7,7 @@ from dpdapi.models import Alias
 class AliasRegexFilter(django_filters.FilterSet):
     class Meta:
         model = Alias
-        fields = {
-            'source': ['iregex', 'regex'],
-            'destination': ['iregex', 'regex']
-        }
+        fields = {"source": ["iregex", "regex"], "destination": ["iregex", "regex"]}
 
 
 class AliasRegexFilterBackend(BaseFilterBackend):

dpdapi/management/commands/import_valiases.py

@@ -1,60 +1,62 @@
+import os
+
 from django.core.management.base import BaseCommand
 from django.db.utils import IntegrityError
-import os
 
 from dpdapi.models import Alias, Domain
 
 
 class Command(BaseCommand):
-    help = 'Import aliases for a single domain at a time, which is the domain of the first source address.'
+    help = "Import aliases for a single domain at a time, which is the domain of the first source address."
 
     def handle(self, *args, **options):
         if len(args) == 1:
             if os.path.exists(args[0]):
                 self.import_file(args[0])
             else:
-                self.stdout.write("{} does not exist".format(args[0]))
+                self.stdout.write(f"{args[0]} does not exist")
         else:
             self.stdout.write("Missing argument")
 
     def import_file(self, filename):
         new_aliases = []
-        counts = {'imported': 0, 'existing': 0}
+        counts = {"imported": 0, "existing": 0}
         # Open file
         with open(filename) as f:
             # Parse
             for l in f:
-                if l.startswith('*: '):
+                if l.startswith("*: "):
                     continue  # Skip catch all
 
-                source, destinations = l.strip().split(': ')
-                for destination in destinations.split(','):
-                    if destination.startswith('"|') or '-admin@' in destination:
+                source, destinations = l.strip().split(": ")
+                for destination in destinations.split(","):
+                    if destination.startswith('"|') or "-admin@" in destination:
                         continue  # Skip mailman-aliases
                     new_aliases.append(Alias(source=source, destination=destination))
 
         if len(new_aliases) == 0:
-            self.stdout.write('No aliases found')
+            self.stdout.write("No aliases found")
             return
 
         # Lookup domain
-        domain_name = new_aliases[0].source.split('@')[1]  # Assumes that every alias is in the same domain
+        domain_name = new_aliases[0].source.split("@")[1]  # Assumes that every alias is in the same domain
         d, created = Domain.objects.get_or_create(name=domain_name)
         if created:
-            self.stdout.write('Created domain {}.'.format(d.name))
+            self.stdout.write(f"Created domain {d.name}.")
 
         # Try to create, log if exists. Slow but works
         for a in new_aliases:
             a.domain = d
             try:
                 a.save()
-                counts['imported'] += 1  # Woop!
+                counts["imported"] += 1  # Woop!
             except IntegrityError:
-                counts['existing'] += 1  # Exists
-                self.stdout.write('Could not import \'{}={}\', already exists.'.format(a.source, a.destination))
+                counts["existing"] += 1  # Exists
+                self.stdout.write(f"Could not import '{a.source}={a.destination}', already exists.")
 
         # Print num imported, existing, total
-        self.stdout.write('Imported {}, skipped {} existing, total {}'.format(
-            counts['imported'],
-            counts['existing'],
-            len(new_aliases)))
+        self.stdout.write(
+            "Imported {}, skipped {} existing, total {}".format(
+                counts["imported"], counts["existing"], len(new_aliases)
+            )
+        )

dpdapi/migrations/0001_initial.py

@@ -1,62 +1,58 @@
-# -*- coding: utf-8 -*-
-from __future__ import unicode_literals
-
-from django.db import models, migrations
+from django.db import migrations, models
 
 
 class Migration(migrations.Migration):
 
-    dependencies = [
-    ]
+    dependencies = []
 
     operations = [
         migrations.CreateModel(
-            name='Alias',
+            name="Alias",
             fields=[
-                ('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)),
-                ('created', models.DateTimeField(auto_now_add=True)),
-                ('updated', models.DateTimeField(auto_now=True)),
-                ('source', models.EmailField(max_length=254)),
-                ('destination', models.EmailField(max_length=254)),
-                ('active', models.BooleanField(default=True)),
+                ("id", models.AutoField(verbose_name="ID", serialize=False, auto_created=True, primary_key=True)),
+                ("created", models.DateTimeField(auto_now_add=True)),
+                ("updated", models.DateTimeField(auto_now=True)),
+                ("source", models.EmailField(max_length=254)),
+                ("destination", models.EmailField(max_length=254)),
+                ("active", models.BooleanField(default=True)),
             ],
             options={
-                'verbose_name_plural': 'aliases',
+                "verbose_name_plural": "aliases",
             },
             bases=(models.Model,),
         ),
         migrations.CreateModel(
-            name='Domain',
+            name="Domain",
             fields=[
-                ('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)),
-                ('created', models.DateTimeField(auto_now_add=True)),
-                ('updated', models.DateTimeField(auto_now=True)),
-                ('name', models.CharField(unique=True, max_length=254)),
+                ("id", models.AutoField(verbose_name="ID", serialize=False, auto_created=True, primary_key=True)),
+                ("created", models.DateTimeField(auto_now_add=True)),
+                ("updated", models.DateTimeField(auto_now=True)),
+                ("name", models.CharField(unique=True, max_length=254)),
             ],
             options={
-                'abstract': False,
+                "abstract": False,
             },
             bases=(models.Model,),
         ),
         migrations.CreateModel(
-            name='User',
+            name="User",
             fields=[
-                ('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)),
-                ('created', models.DateTimeField(auto_now_add=True)),
-                ('updated', models.DateTimeField(auto_now=True)),
-                ('email', models.EmailField(unique=True, max_length=254)),
-                ('password', models.CharField(max_length=109)),
-                ('domain', models.ForeignKey(to='dpdapi.Domain', on_delete=models.CASCADE)),
+                ("id", models.AutoField(verbose_name="ID", serialize=False, auto_created=True, primary_key=True)),
+                ("created", models.DateTimeField(auto_now_add=True)),
+                ("updated", models.DateTimeField(auto_now=True)),
+                ("email", models.EmailField(unique=True, max_length=254)),
+                ("password", models.CharField(max_length=109)),
+                ("domain", models.ForeignKey(to="dpdapi.Domain", on_delete=models.CASCADE)),
             ],
             options={
-                'abstract': False,
+                "abstract": False,
             },
             bases=(models.Model,),
         ),
         migrations.AddField(
-            model_name='alias',
-            name='domain',
-            field=models.ForeignKey(to='dpdapi.Domain', on_delete=models.CASCADE),
+            model_name="alias",
+            name="domain",
+            field=models.ForeignKey(to="dpdapi.Domain", on_delete=models.CASCADE),
             preserve_default=True,
         ),
     ]

dpdapi/migrations/0002_auto_20150118_1638.py

@@ -1,18 +1,15 @@
-# -*- coding: utf-8 -*-
-from __future__ import unicode_literals
-
-from django.db import models, migrations
+from django.db import migrations, models
 
 
 class Migration(migrations.Migration):
 
     dependencies = [
-        ('dpdapi', '0001_initial'),
+        ("dpdapi", "0001_initial"),
     ]
 
     operations = [
         migrations.AlterUniqueTogether(
-            name='alias',
-            unique_together=set([('source', 'destination', 'domain')]),
+            name="alias",
+            unique_together={("source", "destination", "domain")},
         ),
     ]

dpdapi/migrations/0003_auto_20200909_2342.py

@@ -6,13 +6,13 @@ from django.db import migrations, models
 class Migration(migrations.Migration):
 
     dependencies = [
-        ('dpdapi', '0002_auto_20150118_1638'),
+        ("dpdapi", "0002_auto_20150118_1638"),
     ]
 
     operations = [
         migrations.AlterField(
-            model_name='user',
-            name='password',
+            model_name="user",
+            name="password",
             field=models.CharField(max_length=512),
         ),
     ]

dpdapi/models.py

@@ -12,21 +12,21 @@ class BaseModel(models.Model):
 class Alias(BaseModel):
     source = models.EmailField(max_length=254)
     destination = models.EmailField(max_length=254)
-    domain = models.ForeignKey('dpdapi.Domain', models.CASCADE)
+    domain = models.ForeignKey("dpdapi.Domain", models.CASCADE)
     active = models.BooleanField(default=True)
 
     def __str__(self):
-        return "{}={}".format(self.source, self.destination)
+        return f"{self.source}={self.destination}"
 
     def save(self, *args, **kwargs):
         self.source = self.source.lower()
         self.destination = self.destination.lower()
 
-        super(Alias, self).save(*args, **kwargs)
+        super().save(*args, **kwargs)
 
     class Meta:
-        verbose_name_plural = 'aliases'
-        unique_together = ('source', 'destination', 'domain')
+        verbose_name_plural = "aliases"
+        unique_together = ("source", "destination", "domain")
 
 
 class Domain(BaseModel):
@@ -38,13 +38,13 @@ class Domain(BaseModel):
     def save(self, *args, **kwargs):
         self.name = self.name.lower()
 
-        super(Domain, self).save(*args, **kwargs)
+        super().save(*args, **kwargs)
 
 
 class User(BaseModel):
     email = models.EmailField(max_length=254, unique=True)
     password = models.CharField(max_length=512)
-    domain = models.ForeignKey('dpdapi.Domain', models.CASCADE)
+    domain = models.ForeignKey("dpdapi.Domain", models.CASCADE)
 
     def __str__(self):
         return self.email
@@ -52,5 +52,4 @@ class User(BaseModel):
     def save(self, *args, **kwargs):
         self.email = self.email.lower()
 
-        super(User, self).save(*args, **kwargs)
-
+        super().save(*args, **kwargs)

dpdapi/serializers.py

@@ -1,10 +1,10 @@
-from rest_framework import serializers
 from passlib.hash import sha512_crypt
+from rest_framework import serializers
+
 from .models import Alias, Domain, User
 
 
 class AliasSerializer(serializers.ModelSerializer):
-
     def create(self, validated_data):
         # Workaround to accept duplicates
         # FIXME, will fail with existing alias with validated_data['active']=False
@@ -14,36 +14,36 @@ class AliasSerializer(serializers.ModelSerializer):
     class Meta:
         model = Alias
         validators = []  # Disable unique togheter check
-        fields = '__all__'
+        fields = "__all__"
 
 
 class AliasDeleteSerializer(serializers.ModelSerializer):
     class Meta:
         model = Alias
         validators = []  # No Unique togheter check on deletion
-        fields = '__all__'
+        fields = "__all__"
 
 
 class DomainSerializer(serializers.ModelSerializer):
     class Meta:
         model = Domain
-        fields = '__all__'
+        fields = "__all__"
 
 
 class UserSerializer(serializers.ModelSerializer):
     class Meta:
         model = User
-        fields = '__all__'
-        extra_kwargs = {'password': {'write_only': True}}
+        fields = "__all__"
+        extra_kwargs = {"password": {"write_only": True}}
 
     def _hash_password(self, raw_password):
         return sha512_crypt.encrypt(raw_password)
 
     def create(self, validated_data):
         user = User(
-            email=validated_data['email'],
-            domain=validated_data['domain'],
-            password=self._hash_password(validated_data['password'])
+            email=validated_data["email"],
+            domain=validated_data["domain"],
+            password=self._hash_password(validated_data["password"]),
         )
         user.save()
         return user

dpdapi/settings.py

@@ -0,0 +1,106 @@
+""" Ref: https://docs.djangoproject.com/en/3.2/topics/settings/ """
+
+# Build paths inside the project like this: os.path.join(BASE_DIR, ...)
+import os
+
+import sentry_sdk
+from sentry_sdk.integrations.django import DjangoIntegration
+
+BASE_DIR = os.path.dirname(os.path.dirname(__file__))
+SECRET_KEY = os.getenv("SECRET_KEY", "hlaf#yma(8s!cy-bf5e6kc&_5n_l1h!34c(j32-b3v8o7pg$5n")
+DEBUG = bool(os.getenv("DEBUG", True))
+ALLOWED_HOSTS = []
+
+# Application definition
+INSTALLED_APPS = [
+    "django.contrib.admin",
+    "django.contrib.messages",
+    "django.contrib.auth",
+    "django.contrib.contenttypes",
+    "django.contrib.sessions",
+    "django.contrib.staticfiles",
+    "django_extensions",
+    "rest_framework",
+    "django_filters",
+]
+INSTALLED_APPS += [
+    "dpdapi",
+]
+
+MIDDLEWARE = [
+    "django.middleware.security.SecurityMiddleware",
+    "django.contrib.sessions.middleware.SessionMiddleware",
+    "django.middleware.common.CommonMiddleware",
+    "django.middleware.csrf.CsrfViewMiddleware",
+    "django.contrib.auth.middleware.AuthenticationMiddleware",
+    "django.middleware.clickjacking.XFrameOptionsMiddleware",
+    "django.contrib.messages.middleware.MessageMiddleware",
+]
+ROOT_URLCONF = "dpdapi.urls"
+TEMPLATES = [
+    {
+        "BACKEND": "django.template.backends.django.DjangoTemplates",
+        "DIRS": [
+            # insert your TEMPLATE_DIRS here
+        ],
+        "APP_DIRS": True,
+        "OPTIONS": {
+            "context_processors": [
+                "django.template.context_processors.debug",
+                "django.template.context_processors.request",
+                "django.contrib.auth.context_processors.auth",
+                "django.contrib.messages.context_processors.messages",
+            ],
+        },
+    },
+]
+WSGI_APPLICATION = "dpdapi.wsgi.application"
+DATABASES = {
+    "default": {
+        "ENGINE": "django.db.backends.sqlite3",
+        "NAME": os.path.join(BASE_DIR, "db.sqlite3"),
+    }
+}
+LANGUAGE_CODE = "en-us"
+TIME_ZONE = "Europe/Oslo"
+USE_I18N = True
+USE_L10N = True
+USE_TZ = True
+STATIC_URL = "/static/"
+REST_FRAMEWORK = {
+    "DEFAULT_AUTHENTICATION_CLASSES": (
+        "rest_framework.authentication.BasicAuthentication",
+        "rest_framework.authentication.SessionAuthentication",
+    )
+}
+LOGGING = {
+    "version": 1,
+    "disable_existing_loggers": False,
+    "root": {
+        "level": "INFO",
+        "handlers": ["console"],
+    },
+    "formatters": {
+        "verbose": {"format": "%(levelname)s %(asctime)s %(module)s %(process)d %(thread)d %(message)s"},
+    },
+    "handlers": {"console": {"level": "DEBUG", "class": "logging.StreamHandler", "formatter": "verbose"}},
+    "loggers": {
+        "django.db.backends": {
+            "level": "ERROR",
+            "handlers": ["console"],
+            "propagate": False,
+        },
+    },
+}
+
+SENTRY_DSN = os.getenv("SENTRY_DSN", "https://c7e6a47844484c0a81a98e10e48c1b71@sentry.neuf.no/8")
+if SENTRY_DSN:
+    SENTRY_ENVIRONMENT = os.getenv("SENTRY_ENVIRONMENT", "production")
+    sentry_sdk.init(dsn=SENTRY_DSN, environment=SENTRY_ENVIRONMENT, integrations=[DjangoIntegration()])
+
+DEFAULT_AUTO_FIELD = "django.db.models.AutoField"
+
+try:
+    from .local import *  # noqa
+except ImportError:
+    pass

dpdapi/tests.py

@@ -1,43 +1,41 @@
-import json
-
 from django.contrib.auth.models import User
 from django.urls import reverse
 from rest_framework.test import APITestCase
 
-from dpdapi.models import Domain, Alias
+from dpdapi.models import Alias, Domain
 
 
 class RegexTestCase(APITestCase):
-    aliases = ['ops-alert@example.com', 'yolo-announce@example.com']
+    aliases = ["ops-alert@example.com", "yolo-announce@example.com"]
 
     def setUp(self):
-        user = User.objects.create(username='yolo')
-        self.domain = Domain.objects.create(name='example.com')
+        user = User.objects.create(username="yolo")
+        self.domain = Domain.objects.create(name="example.com")
         for a in self.aliases:
-            Alias.objects.create(source=a, destination='ninja@example.com', domain=self.domain)
+            Alias.objects.create(source=a, destination="ninja@example.com", domain=self.domain)
         self.client.force_login(user)
 
     def test_get_all(self):
-        res = self.client.get(reverse('alias-list'))
+        res = self.client.get(reverse("alias-list"))
         self.assertEqual(len(res.data), len(self.aliases))
 
     def test_domain_filter(self):
-        params = {'domain__name': 'example.com'}
-        res = self.client.get(reverse('alias-list'), data=params)
+        params = {"domain__name": "example.com"}
+        res = self.client.get(reverse("alias-list"), data=params)
         self.assertEqual(len(res.data), len(self.aliases))
 
     def test_domain_filter_no_hits(self):
-        params = {'domain__name': 'zexample.com'}
-        res = self.client.get(reverse('alias-list'), data=params)
+        params = {"domain__name": "zexample.com"}
+        res = self.client.get(reverse("alias-list"), data=params)
         self.assertEqual(len(res.data), 0)
 
     def test_regex_filter(self):
-        params = {'source__regex': '^ops-|^ops@'}
-        res = self.client.get(reverse('alias-list'), data=params)
+        params = {"source__regex": "^ops-|^ops@"}
+        res = self.client.get(reverse("alias-list"), data=params)
         self.assertEqual(len(res.data), 1)
 
     def test_delete_bulk_single(self):
-        data = {'destination': 'ninja@example.com', 'source': 'ding@example.com', 'domain': self.domain.pk}
-        Alias.objects.create(source=data['source'], destination=data['destination'], domain=self.domain)
-        res = self.client.delete(reverse('alias-delete-bulk'), data=[data], format='json')
+        data = {"destination": "ninja@example.com", "source": "ding@example.com", "domain": self.domain.pk}
+        Alias.objects.create(source=data["source"], destination=data["destination"], domain=self.domain)
+        res = self.client.delete(reverse("alias-delete-bulk"), data=[data], format="json")
         self.assertEqual(res.status_code, 204, res.data)

dpdapi/urls.py

@@ -1,10 +1,19 @@
+from django.contrib import admin
+from django.urls import include, path, re_path
+from django.views.generic import RedirectView
+from rest_framework import urls as rest_framework_urls
 from rest_framework.routers import DefaultRouter
 
 from .views import AliasViewSet, DomainViewSet, UserViewSet
 
+api_router = DefaultRouter()
+api_router.register(r"aliases", AliasViewSet)
+api_router.register(r"domains", DomainViewSet)
+api_router.register(r"users", UserViewSet)
 
-router = DefaultRouter()
-router.register(r'aliases', AliasViewSet)
-router.register(r'domains', DomainViewSet)
-router.register(r'users', UserViewSet)
-urlpatterns = router.urls
+urlpatterns = [
+    path("", RedirectView.as_view(pattern_name="api-root")),
+    re_path(r"^api/", include(api_router.urls)),
+    re_path(r"^admin/", admin.site.urls),
+    re_path(r"^api-auth/", include(rest_framework_urls, namespace="rest_framework")),
+]

dpdapi/views.py

@@ -1,32 +1,36 @@
 import logging
 import operator
-
 from functools import reduce
+
 from django.db.models import Q
 from django_filters.rest_framework import DjangoFilterBackend
+from rest_framework import status, viewsets
 from rest_framework.decorators import action
 from rest_framework.mixins import ListModelMixin
 from rest_framework.permissions import IsAuthenticated
 from rest_framework.response import Response
-from rest_framework import status, viewsets
 from rest_framework.viewsets import GenericViewSet
 
 from dpdapi.filters import AliasRegexFilterBackend
 from dpdapi.models import Alias, Domain, User
-from dpdapi.serializers import AliasSerializer, AliasDeleteSerializer, DomainSerializer, UserSerializer
+from dpdapi.serializers import AliasDeleteSerializer, AliasSerializer, DomainSerializer, UserSerializer
 
 logger = logging.getLogger(__name__)
 
 
 class AliasViewSet(ListModelMixin, GenericViewSet):
-    """ Aliases """
+    """Aliases"""
+
     queryset = Alias.objects.all()
     serializer_class = AliasSerializer
     permission_classes = [IsAuthenticated]
     filter_backends = (DjangoFilterBackend, AliasRegexFilterBackend)
-    filterset_fields = ('domain', 'domain__name',)
+    filterset_fields = (
+        "domain",
+        "domain__name",
+    )
 
-    @action(detail=False, methods=['post'])
+    @action(detail=False, methods=["post"])
     def create_bulk(self, request):
         serializer = AliasSerializer(data=request.data, many=True)
 
@@ -35,7 +39,7 @@ class AliasViewSet(ListModelMixin, GenericViewSet):
         serializer.save()
         return Response(serializer.data, status=status.HTTP_201_CREATED)
 
-    @action(detail=False, methods=['delete'])
+    @action(detail=False, methods=["delete"])
     def delete_bulk(self, request):
         serializer = AliasDeleteSerializer(data=request.data, many=True)
 
@@ -45,14 +49,14 @@ class AliasViewSet(ListModelMixin, GenericViewSet):
         q_list = map(lambda x: Q(**x), serializer.initial_data)
         aliases = Alias.objects.filter(reduce(operator.or_, q_list))
 
-        initial_set = set(map(lambda x: (x['source'], x['destination'], x['domain']), serializer.initial_data))
-        deleted_set = set(aliases.values_list('source', 'destination', 'domain'))
+        initial_set = set(map(lambda x: (x["source"], x["destination"], x["domain"]), serializer.initial_data))
+        deleted_set = set(aliases.values_list("source", "destination", "domain"))
         diff = initial_set.difference(deleted_set)
-        logger.debug('Deleted: {}'.format(','.join(map(lambda x: '{0[0]}={0[1]}'.format(x), deleted_set))))
+        logger.debug("Deleted: {}".format(",".join(map(lambda x: "{0[0]}={0[1]}".format(x), deleted_set))))
         aliases.delete()
 
         if diff:
-            diff_formatted = map(lambda x: '{0[0]}={0[1]}'.format(x), diff)
+            diff_formatted = map(lambda x: "{0[0]}={0[1]}".format(x), diff)
             res = "Aliases '{}' could not be found and were not deleted.".format(", ".join(diff_formatted))
             logger.info(res)
             return Response({"result": res}, status=status.HTTP_200_OK)
@@ -61,16 +65,18 @@ class AliasViewSet(ListModelMixin, GenericViewSet):
 
 
 class DomainViewSet(viewsets.ModelViewSet):
-    """ Domains """
+    """Domains"""
+
     queryset = Domain.objects.all()
     serializer_class = DomainSerializer
     permission_classes = [IsAuthenticated]
     filter_backends = (DjangoFilterBackend,)
-    filterset_fields = ('name',)
+    filterset_fields = ("name",)
 
 
 class UserViewSet(viewsets.ModelViewSet):
-    """ Users """
+    """Users"""
+
     queryset = User.objects.all()
     serializer_class = UserSerializer
     permission_classes = [IsAuthenticated]

dpdapi/wsgi.py

@@ -0,0 +1,7 @@
+import os
+
+from django.core.wsgi import get_wsgi_application
+
+os.environ.setdefault("DJANGO_SETTINGS_MODULE", "dpdapi.settings")
+
+application = get_wsgi_application()