Django Clean Architecture DRF · Celery · Channels · PostgreSQL · Redis

from rest_framework import serializers
from .models import User


class UserSerializer(serializers.ModelSerializer):
    class Meta:
        model  = User
        fields = ['id', 'email', 'name', 'role', 'created_at']
        read_only_fields = ['id', 'created_at']


class RegisterSerializer(serializers.Serializer):
    email    = serializers.EmailField()
    name     = serializers.CharField(max_length=255)
    password = serializers.CharField(
        min_length=8, write_only=True,
        style={'input_type': 'password'},
    )

    def validate_email(self, value):
        if User.objects.filter(email=value).exists():
            raise serializers.ValidationError('Email already registered.')
        return value.lower()


class LoginSerializer(serializers.Serializer):
    email    = serializers.EmailField()
    password = serializers.CharField(write_only=True)

from rest_framework import viewsets, status
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticated
from .services import UserService
from .serializers import RegisterSerializer, UserSerializer


class AuthViewSet(viewsets.ViewSet):
    """Thin view — call service, return response. Nothing else."""

    @action(detail=False, methods=['post'])
    def register(self, request):
        serializer = RegisterSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)

        user = UserService.register(**serializer.validated_data)
        return Response(
            UserSerializer(user).data,
            status=status.HTTP_201_CREATED,
        )

    @action(detail=False, methods=['post'])
    def login(self, request):
        serializer = LoginSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)

        tokens = UserService.login(**serializer.validated_data)
        return Response(tokens)

    @action(detail=False, methods=['get'], permission_classes=[IsAuthenticated])
    def me(self, request):
        return Response(UserSerializer(request.user).data)

from rest_framework.routers import DefaultRouter
from .views import AuthViewSet, UserViewSet

router = DefaultRouter()
router.register(r'auth',  AuthViewSet,  basename='auth')
router.register(r'users', UserViewSet, basename='user')

urlpatterns = router.urls

# Generates routes:
# POST /auth/register/
# POST /auth/login/
# GET  /auth/me/
# GET  /users/          (list — admin only)
# GET  /users/{id}/     (retrieve)
# PUT  /users/{id}/     (update)
# DELETE /users/{id}/   (destroy)

import environ

env = environ.Env(
    DEBUG=(bool, False),
    ALLOWED_HOSTS=(list, []),
)
environ.Env.read_env()  # reads .env file

SECRET_KEY    = env('SECRET_KEY')
DEBUG         = env('DEBUG')
ALLOWED_HOSTS = env('ALLOWED_HOSTS')

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'rest_framework',
    'corsheaders',
    'channels',
    'django_filters',
    'apps.common',
    'apps.users',
    'apps.chat',
]

DATABASES = {'default': env.db('DATABASE_URL')}
CACHES    = {'default': env.cache('REDIS_URL')}

AUTH_USER_MODEL  = 'users.User'
JWT_SECRET       = env('JWT_SECRET')
FIELD_ENCRYPTION_KEY = env('FIELD_ENCRYPTION_KEY')

REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'apps.common.authentication.JWTAuthentication',
    ],
    'DEFAULT_PERMISSION_CLASSES': [
        'rest_framework.permissions.IsAuthenticated',
    ],
    'DEFAULT_PAGINATION_CLASS':
        'apps.common.pagination.PageNumberPagination',
    'PAGE_SIZE': 20,
    'EXCEPTION_HANDLER':
        'apps.common.exceptions.custom_exception_handler',
}

import functools, json
from django.core.cache import cache


def cache_result(ttl: int = 300, key_prefix: str = ''):
    """Decorator that caches the return value in Redis."""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            key = f'{key_prefix}:{":".join(str(a) for a in args)}'
            hit = cache.get(key)
            if hit is not None:
                return hit
            result = func(*args, **kwargs)
            cache.set(key, result, timeout=ttl)
            return result
        return wrapper
    return decorator


def invalidate_cache(key_prefix: str, *args) -> None:
    key = f'{key_prefix}:{":".join(str(a) for a in args)}'
    cache.delete(key)

# Usage in selectors.py:
@cache_result(ttl=300, key_prefix='user')
def by_id(user_id: str) -> User | None:
    return User.objects.filter(id=user_id).first()

import factory
from factory.django import DjangoModelFactory
from apps.users.models import User


class UserFactory(DjangoModelFactory):
    class Meta:
        model = User

    email = factory.Sequence(lambda n: f'user{n}@example.com')
    name  = factory.Faker('name')
    role  = 'user'

    @classmethod
    def _create(cls, model_class, *args, **kwargs):
        return model_class.objects.create_user(
            password='testpassword123', **kwargs
        )


class AdminFactory(UserFactory):
    role     = 'admin'
    is_staff = True


# test_services.py — fast, no HTTP
def test_register_creates_user(db):
    user = UserService.register(
        email='new@example.com', name='New', password='pass1234!'
    )
    assert user.email == 'new@example.com'
    assert user.check_password('pass1234!')