Как создавать и использовать эндпоинты (endpoints) в Django ORM?

Создание и использование endpoints в Django ORM

Эндпоинты — это URL маршруты которые обрабатывают HTTP запросы и возвращают данные. В Django это реализуется через views и URLs. Разберём полный цикл создания.

1. Базовая архитектура Django endpoints

# СТРУКТУРА ПРОЕКТА
#
myproject/
├── myproject/
│   ├── settings.py
│   ├── urls.py          # Главный роутер
│   └── wsgi.py
├── myapp/
│   ├── models.py        # Модели БД
│   ├── views.py         # Бизнес-логика (view handlers)
│   ├── urls.py          # Маршруты приложения
│   ├── serializers.py   # Преобразование данных JSON
│   └── tests.py         # Тесты
└── manage.py

# FLOW запроса:
# HTTP Request → myproject.urls → myapp.urls → views.py → models.py → Response

2. Создание моделей (начало)

# myapp/models.py
from django.db import models
from django.utils import timezone

class User(models.Model):
    """Модель пользователя"""
    name = models.CharField(max_length=100)
    email = models.EmailField(unique=True)
    phone = models.CharField(max_length=20, null=True, blank=True)
    created_at = models.DateTimeField(default=timezone.now)
    is_active = models.BooleanField(default=True)
    
    class Meta:
        db_table = "users"
        ordering = ["-created_at"]
    
    def __str__(self):
        return self.name
    
    def __repr__(self):
        return f"<User(id={self.id}, email={self.email})>"

class Post(models.Model):
    """Модель поста блога"""
    author = models.ForeignKey(User, on_delete=models.CASCADE, related_name="posts")
    title = models.CharField(max_length=200)
    content = models.TextField()
    published_at = models.DateTimeField(default=timezone.now)
    updated_at = models.DateTimeField(auto_now=True)
    
    class Meta:
        db_table = "posts"
        ordering = ["-published_at"]
    
    def __str__(self):
        return self.title

3. Миграции БД

# Создать миграцию
python manage.py makemigrations

# Применить миграцию
python manage.py migrate

# Показать статус миграций
python manage.py showmigrations

4. Создание views (обработчики)

Function-Based Views (FBV)

# myapp/views.py
from django.http import JsonResponse
from django.views.decorators.http import require_http_methods
from django.views.decorators.csrf import csrf_exempt
from django.core.paginator import Paginator
import json

from .models import User, Post

@require_http_methods(["GET"])
def list_users(request):
    """GET /users — список пользователей"""
    # Фильтрация
    users = User.objects.filter(is_active=True)
    
    # Пагинация
    page = request.GET.get("page", 1)
    paginator = Paginator(users, per_page=10)
    page_obj = paginator.get_page(page)
    
    # Преобразование в JSON
    data = {
        "count": paginator.count,
        "next": page_obj.next_page_number() if page_obj.has_next() else None,
        "previous": page_obj.previous_page_number() if page_obj.has_previous() else None,
        "results": [
            {
                "id": user.id,
                "name": user.name,
                "email": user.email,
                "created_at": user.created_at.isoformat(),
            }
            for user in page_obj
        ]
    }
    
    return JsonResponse(data)

@require_http_methods(["GET"])
def get_user(request, user_id):
    """GET /users/{id} — получить пользователя"""
    try:
        user = User.objects.get(id=user_id)
    except User.DoesNotExist:
        return JsonResponse({"error": "User not found"}, status=404)
    
    data = {
        "id": user.id,
        "name": user.name,
        "email": user.email,
        "phone": user.phone,
        "created_at": user.created_at.isoformat(),
        "is_active": user.is_active,
        "posts_count": user.posts.count(),
    }
    
    return JsonResponse(data)

@csrf_exempt
@require_http_methods(["POST"])
def create_user(request):
    """POST /users — создать пользователя"""
    try:
        data = json.loads(request.body)
    except json.JSONDecodeError:
        return JsonResponse({"error": "Invalid JSON"}, status=400)
    
    # Валидация
    if not data.get("name") or not data.get("email"):
        return JsonResponse(
            {"error": "name and email are required"},
            status=400
        )
    
    # Проверка на дубликат
    if User.objects.filter(email=data["email"]).exists():
        return JsonResponse({"error": "Email already exists"}, status=400)
    
    # Создание
    user = User.objects.create(
        name=data["name"],
        email=data["email"],
        phone=data.get("phone"),
    )
    
    return JsonResponse(
        {
            "id": user.id,
            "name": user.name,
            "email": user.email,
        },
        status=201
    )

@csrf_exempt
@require_http_methods(["PUT"])
def update_user(request, user_id):
    """PUT /users/{id} — обновить пользователя"""
    try:
        user = User.objects.get(id=user_id)
    except User.DoesNotExist:
        return JsonResponse({"error": "User not found"}, status=404)
    
    try:
        data = json.loads(request.body)
    except json.JSONDecodeError:
        return JsonResponse({"error": "Invalid JSON"}, status=400)
    
    # Обновление
    user.name = data.get("name", user.name)
    user.email = data.get("email", user.email)
    user.phone = data.get("phone", user.phone)
    user.save()
    
    return JsonResponse({"id": user.id, "name": user.name})

@csrf_exempt
@require_http_methods(["DELETE"])
def delete_user(request, user_id):
    """DELETE /users/{id} — удалить пользователя"""
    try:
        user = User.objects.get(id=user_id)
    except User.DoesNotExist:
        return JsonResponse({"error": "User not found"}, status=404)
    
    user.delete()
    return JsonResponse({"message": "User deleted"}, status=204)

Class-Based Views (CBV)

# myapp/views.py (альтернатива)
from django.views import View
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.utils.decorators import method_decorator
import json

class UserListView(View):
    """Обработчик для /users"""
    
    def get(self, request):
        """GET /users"""
        users = User.objects.filter(is_active=True)[:10]
        data = [
            {"id": u.id, "name": u.name, "email": u.email}
            for u in users
        ]
        return JsonResponse({"results": data})
    
    @method_decorator(csrf_exempt)
    def post(self, request):
        """POST /users"""
        data = json.loads(request.body)
        user = User.objects.create(name=data["name"], email=data["email"])
        return JsonResponse({"id": user.id}, status=201)

class UserDetailView(View):
    """Обработчик для /users/{id}"""
    
    def get_object(self, user_id):
        try:
            return User.objects.get(id=user_id)
        except User.DoesNotExist:
            return None
    
    def get(self, request, user_id):
        """GET /users/{id}"""
        user = self.get_object(user_id)
        if not user:
            return JsonResponse({"error": "Not found"}, status=404)
        return JsonResponse({"id": user.id, "name": user.name})
    
    @method_decorator(csrf_exempt)
    def put(self, request, user_id):
        """PUT /users/{id}"""
        user = self.get_object(user_id)
        if not user:
            return JsonResponse({"error": "Not found"}, status=404)
        
        data = json.loads(request.body)
        user.name = data.get("name", user.name)
        user.save()
        return JsonResponse({"id": user.id, "name": user.name})
    
    @method_decorator(csrf_exempt)
    def delete(self, request, user_id):
        """DELETE /users/{id}"""
        user = self.get_object(user_id)
        if not user:
            return JsonResponse({"error": "Not found"}, status=404)
        user.delete()
        return JsonResponse({}, status=204)

5. URL маршруты

# myapp/urls.py
from django.urls import path
from . import views

urlpatterns = [
    # Function-based views
    path("users", views.list_users, name="user-list"),
    path("users/<int:user_id>", views.get_user, name="user-detail"),
    path("users/create", views.create_user, name="user-create"),
    path("users/<int:user_id>/update", views.update_user, name="user-update"),
    path("users/<int:user_id>/delete", views.delete_user, name="user-delete"),
    
    # Class-based views (альтернатива)
    # path("users", views.UserListView.as_view(), name="user-list"),
    # path("users/<int:user_id>", views.UserDetailView.as_view(), name="user-detail"),
]

# myproject/urls.py (главный)
from django.contrib import admin
from django.urls import path, include

urlpatterns = [
    path("admin", admin.site.urls),
    path("api/v1/", include("myapp.urls")),  # /api/v1/users
]

6. REST Framework (рекомендуется для API)

# Установка: pip install djangorestframework

# myapp/serializers.py
from rest_framework import serializers
from .models import User, Post

class UserSerializer(serializers.ModelSerializer):
    posts_count = serializers.SerializerMethodField()
    
    class Meta:
        model = User
        fields = ["id", "name", "email", "phone", "created_at", "is_active", "posts_count"]
        read_only_fields = ["id", "created_at", "posts_count"]
    
    def get_posts_count(self, obj):
        return obj.posts.count()

class PostSerializer(serializers.ModelSerializer):
    author = UserSerializer(read_only=True)
    
    class Meta:
        model = Post
        fields = ["id", "author", "title", "content", "published_at", "updated_at"]
        read_only_fields = ["id", "published_at", "updated_at"]

# myapp/views.py (с REST Framework)
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 .models import User, Post
from .serializers import UserSerializer, PostSerializer

class UserViewSet(viewsets.ModelViewSet):
    """REST endpoint для users"""
    queryset = User.objects.filter(is_active=True)
    serializer_class = UserSerializer
    permission_classes = [IsAuthenticated]
    
    def get_queryset(self):
        # Фильтрация по параметрам
        qs = super().get_queryset()
        email = self.request.query_params.get("email")
        if email:
            qs = qs.filter(email__icontains=email)
        return qs
    
    @action(detail=True, methods=["get"])
    def posts(self, request, pk=None):
        """GET /users/{id}/posts — посты пользователя"""
        user = self.get_object()
        posts = user.posts.all()
        serializer = PostSerializer(posts, many=True)
        return Response(serializer.data)
    
    @action(detail=False, methods=["post"])
    def bulk_create(self, request):
        """POST /users/bulk_create — создать много пользователей"""
        users_data = request.data
        users = [User(**data) for data in users_data]
        User.objects.bulk_create(users)
        return Response({"message": "Created"}, status=status.HTTP_201_CREATED)

# myapp/urls.py
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from . import views

router = DefaultRouter()
router.register(r"users", views.UserViewSet)

urlpatterns = [
    path("", include(router.urls)),
]

7. Query Optimization

# ПРОБЛЕМА: N+1 queries
def get_users_with_posts(request):
    """ПЛОХО — для каждого user по 1 запросу к posts"""
    users = User.objects.all()
    result = []
    for user in users:
        result.append({
            "id": user.id,
            "name": user.name,
            "posts_count": user.posts.count()  # N+1! Дополнительный запрос
        })
    return JsonResponse({"users": result})

# РЕШЕНИЕ 1: select_related (для ForeignKey)
users = User.objects.select_related("profile").all()

# РЕШЕНИЕ 2: prefetch_related (для ManyToMany и обратные FK)
from django.db.models import Prefetch, Count

users = User.objects.prefetch_related("posts").all()

# РЕШЕНИЕ 3: annotate (для aggregation)
users = User.objects.annotate(
    posts_count=Count("posts")
).all()

def get_users_optimized(request):
    """ХОРОШО — оптимизированный запрос"""
    users = User.objects.annotate(
        posts_count=Count("posts")
    ).filter(is_active=True)
    
    result = [
        {
            "id": user.id,
            "name": user.name,
            "posts_count": user.posts_count,  # Из annotation
        }
        for user in users
    ]
    return JsonResponse({"users": result})

8. Фильтрация и поиск

# myapp/views.py
from django.db.models import Q

def search_users(request):
    """GET /users/search?q=john — поиск пользователей"""
    q = request.GET.get("q", "")
    
    users = User.objects.filter(
        Q(name__icontains=q) | Q(email__icontains=q)
    ).distinct()
    
    return JsonResponse({
        "results": [
            {"id": u.id, "name": u.name, "email": u.email}
            for u in users
        ]
    })

# REST Framework с фильтрацией
from django_filters import FilterSet, CharFilter
from django_filters.rest_framework import DjangoFilterBackend

class UserFilter(FilterSet):
    name = CharFilter(field_name="name", lookup_expr="icontains")
    email = CharFilter(field_name="email", lookup_expr="icontains")
    
    class Meta:
        model = User
        fields = ["name", "email", "is_active"]

class UserViewSet(viewsets.ModelViewSet):
    queryset = User.objects.all()
    serializer_class = UserSerializer
    filter_backends = [DjangoFilterBackend]
    filterset_class = UserFilter
    search_fields = ["name", "email"]
    ordering_fields = ["created_at", "name"]

9. Тестирование endpoints

# myapp/tests.py
from django.test import TestCase, Client
from django.urls import reverse
from .models import User
import json

class UserAPITestCase(TestCase):
    def setUp(self):
        self.client = Client()
        self.user = User.objects.create(
            name="John",
            email="john@example.com"
        )
    
    def test_list_users(self):
        """Test GET /users"""
        response = self.client.get("/api/v1/users")
        self.assertEqual(response.status_code, 200)
        data = json.loads(response.content)
        self.assertGreaterEqual(len(data["results"]), 1)
    
    def test_get_user(self):
        """Test GET /users/{id}"""
        response = self.client.get(f"/api/v1/users/{self.user.id}")
        self.assertEqual(response.status_code, 200)
        data = json.loads(response.content)
        self.assertEqual(data["email"], "john@example.com")
    
    def test_create_user(self):
        """Test POST /users"""
        response = self.client.post(
            "/api/v1/users",
            data=json.dumps({"name": "Jane", "email": "jane@example.com"}),
            content_type="application/json"
        )
        self.assertEqual(response.status_code, 201)
        self.assertTrue(User.objects.filter(email="jane@example.com").exists())
    
    def test_delete_user(self):
        """Test DELETE /users/{id}"""
        user_id = self.user.id
        response = self.client.delete(f"/api/v1/users/{user_id}")
        self.assertEqual(response.status_code, 204)
        self.assertFalse(User.objects.filter(id=user_id).exists())

Итоговая структура endpoints

GET    /api/v1/users              — список (paginated)
GET    /api/v1/users/{id}         — деталь
POST   /api/v1/users              — создание
PUT    /api/v1/users/{id}         — полное обновление
PATCH  /api/v1/users/{id}         — частичное обновление
DELETE /api/v1/users/{id}         — удаление
GET    /api/v1/users/{id}/posts   — related resource
GET    /api/v1/users/search       — поиск

Рекомендация: Используй Django REST Framework для API — намного удобнее чем вручную писать JSON responses!