انطلق في Python Django REST Framework: خطوتك الأولى العملية اليوم

GET    /api/articles/        # سرد
GET    /api/articles/42/     # تفاصيل
POST   /api/articles/        # إنشاء
PUT    /api/articles/42/     # استبدال
PATCH  /api/articles/42/     # تحديث جزئي
DELETE /api/articles/42/     # حذف

# أفعال HTTP + URIs = CRUD عالمي

# أفضل الممارسات
GET    /api/v1/users/                  # قائمة
GET    /api/v1/users/42/               # تفاصيل
GET    /api/v1/users/42/orders/        # مورد فرعي
POST   /api/v1/users/                  # 201 Created + Location
PATCH  /api/v1/users/42/               # جزئي
DELETE /api/v1/users/42/               # 204 No Content

# فلاتر / تصفح
GET    /api/v1/users/?role=admin&page=2&page_size=20

# رموز الحالة
200 OK, 201 Created, 204 No Content
400 Bad Request, 401 Unauthorized, 403 Forbidden
404 Not Found, 409 Conflict, 422 Unprocessable
500 Server Error, 503 Unavailable

{
  "data": [...],
  "meta": {
    "count": 150,
    "page": 2,
    "page_size": 20
  },
  "links": {
    "next": "/api/v1/users/?page=3",
    "prev": "/api/v1/users/?page=1"
  }
}

# الأخطاء
{
  "error": {
    "code": "VALIDATION_ERROR",
    "message": "Email is required",
    "details": {"email": ["This field is required"]}
  }
}

# workspaces/models.py
class Workspace(models.Model):
    class Plan(models.TextChoices):
        FREE = "free", "Free"
        PRO = "pro", "Pro"
        ENTERPRISE = "enterprise", "Enterprise"
    
    name = models.CharField(max_length=100)
    slug = models.SlugField(unique=True)
    plan = models.CharField(max_length=20, choices=Plan.choices, default=Plan.FREE)
    stripe_customer_id = models.CharField(max_length=100, blank=True)
    created_at = models.DateTimeField(auto_now_add=True)

class Membership(models.Model):
    class Role(models.TextChoices):
        OWNER = "owner"
        ADMIN = "admin"
        MEMBER = "member"
        GUEST = "guest"
    
    workspace = models.ForeignKey(Workspace, on_delete=models.CASCADE, related_name="memberships")
    user = models.ForeignKey("accounts.User", on_delete=models.CASCADE)
    role = models.CharField(max_length=20, choices=Role.choices, default=Role.MEMBER)
    invited_at = models.DateTimeField(auto_now_add=True)
    
    class Meta:
        unique_together = ["workspace", "user"]

# projects/models.py
class Project(models.Model):
    workspace = models.ForeignKey(Workspace, on_delete=models.CASCADE, related_name="projects")
    name = models.CharField(max_length=200)
    description = models.TextField(blank=True)
    created_by = models.ForeignKey("accounts.User", on_delete=models.PROTECT)
    created_at = models.DateTimeField(auto_now_add=True)
    
    class Meta:
        indexes = [models.Index(fields=["workspace", "-created_at"])]

class Task(models.Model):
    class Status(models.TextChoices):
        TODO = "todo"
        IN_PROGRESS = "in_progress"
        DONE = "done"
    
    project = models.ForeignKey(Project, on_delete=models.CASCADE, related_name="tasks")
    title = models.CharField(max_length=200)
    description = models.TextField(blank=True)
    status = models.CharField(max_length=20, choices=Status.choices, default=Status.TODO)
    assigned_to = models.ForeignKey("accounts.User", null=True, blank=True, on_delete=models.SET_NULL)
    due_date = models.DateField(null=True, blank=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

# workspaces/middleware.py
class WorkspaceMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
    
    def __call__(self, request):
        host = request.get_host().split(":")[0]
        subdomain = host.split(".")[0]
        
        try:
            request.workspace = Workspace.objects.get(slug=subdomain)
        except Workspace.DoesNotExist:
            request.workspace = None
        
        return self.get_response(request)

# settings.py
MIDDLEWARE += ["workspaces.middleware.WorkspaceMiddleware"]

# URLs : acme.api.example.com -> workspace "acme"

# workspaces/permissions.py
from rest_framework.permissions import BasePermission

class IsWorkspaceMember(BasePermission):
    def has_permission(self, request, view):
        if not request.user.is_authenticated:
            return False
        if not request.workspace:
            return False
        return Membership.objects.filter(
            workspace=request.workspace, user=request.user
        ).exists()

class IsWorkspaceAdmin(BasePermission):
    def has_permission(self, request, view):
        return Membership.objects.filter(
            workspace=request.workspace,
            user=request.user,
            role__in=["owner", "admin"]
        ).exists()

class WorkspaceQuerysetMixin:
    """تصفية تلقائية حسب مساحة العمل الحالية."""
    workspace_field = "workspace"
    
    def get_queryset(self):
        qs = super().get_queryset()
        if self.request.workspace:
            return qs.filter(**{self.workspace_field: self.request.workspace})
        return qs.none()
    
    def perform_create(self, serializer):
        serializer.save(workspace=self.request.workspace, created_by=self.request.user)

class ProjectViewSet(WorkspaceQuerysetMixin, viewsets.ModelViewSet):
    queryset = Project.objects.all()
    serializer_class = ProjectSerializer
    permission_classes = [IsAuthenticated, IsWorkspaceMember]
    
    def get_queryset(self):
        return super().get_queryset().select_related("created_by")

class TaskViewSet(WorkspaceQuerysetMixin, viewsets.ModelViewSet):
    queryset = Task.objects.all()
    serializer_class = TaskSerializer
    permission_classes = [IsAuthenticated, IsWorkspaceMember]
    workspace_field = "project__workspace"
    
    def get_queryset(self):
        return super().get_queryset().select_related("project", "assigned_to")

# workspaces/quotas.py
PLAN_LIMITS = {
    "free": {"max_projects": 3, "max_members": 3, "api_rate": "100/day"},
    "pro": {"max_projects": None, "max_members": 20, "api_rate": "10000/day"},
    "enterprise": {"max_projects": None, "max_members": None, "api_rate": None},
}

def check_project_quota(workspace):
    limit = PLAN_LIMITS[workspace.plan]["max_projects"]
    if limit is None:
        return
    current = workspace.projects.count()
    if current >= limit:
        raise serializers.ValidationError(
            f"Limite atteinte : {limit} projects pour le plan {workspace.plan}. "
            f"Upgrade vers Pro."
        )

# في الـ serializer
class ProjectSerializer(serializers.ModelSerializer):
    def create(self, validated_data):
        check_project_quota(validated_data["workspace"])
        return super().create(validated_data)

class PlanRateThrottle(SimpleRateThrottle):
    scope = "plan"
    
    def allow_request(self, request, view):
        if not request.workspace:
            return True
        
        rate = PLAN_LIMITS[request.workspace.plan]["api_rate"]
        if rate is None:
            return True          # enterprise
        
        self.num_requests, self.duration = self.parse_rate(rate)
        self.request = request
        return super().allow_request(request, view)
    
    def get_cache_key(self, request, view):
        return f"plan_{request.workspace.id}"

# settings.py - Redis backend
CACHES = {
    "default": {
        "BACKEND": "django.core.cache.backends.redis.RedisCache",
        "LOCATION": env("REDIS_URL"),
        "TIMEOUT": 300,                # 5 دقائق افتراضيًا
        "OPTIONS": {
            "db": 1,
            "max_connections": 50,
        }
    }
}

# اختبار سريع باستخدام الذاكرة المحلية (للتطوير)
CACHES = {
    "default": {"BACKEND": "django.core.cache.backends.locmem.LocMemCache"}
}

from django.views.decorators.cache import cache_page
from django.utils.decorators import method_decorator

class PopularBooksView(generics.ListAPIView):
    queryset = Book.objects.filter(popular=True)
    serializer_class = BookSerializer
    
    @method_decorator(cache_page(60 * 15))      # 15 دقيقة
    def dispatch(self, *args, **kwargs):
        return super().dispatch(*args, **kwargs)

# أو باستخدام vary_on_headers
from django.views.decorators.vary import vary_on_headers

@method_decorator(cache_page(600), name="dispatch")
@method_decorator(vary_on_headers("Authorization"), name="dispatch")
class MyView(generics.ListAPIView):
    # تخزين مختلف لكل توكن
    ...

from django.core.cache import cache

class StatsView(APIView):
    def get(self, request):
        cache_key = f"stats_{request.user.id}"
        data = cache.get(cache_key)
        
        if data is None:
            # حساب مكلف
            data = compute_expensive_stats(request.user)
            cache.set(cache_key, data, timeout=3600)
        
        return Response(data)

# get_or_set في سطر واحد
data = cache.get_or_set(f"stats_{user.id}", lambda: compute_stats(user), 3600)

# تخزين متعدد
cache.set_many({"a": 1, "b": 2}, timeout=60)
values = cache.get_many(["a", "b"])

# إبطال
cache.delete("stats_42")
cache.delete_many(["stats_42", "stats_43"])
cache.clear()                                # مسح الكل

from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver

@receiver([post_save, post_delete], sender=Book)
def invalidate_books_cache(sender, instance, **kwargs):
    cache.delete("popular_books")
    cache.delete(f"book_{instance.id}")
    cache.delete(f"books_by_author_{instance.author_id}")

from rest_framework.response import Response
import hashlib

class BookDetailView(generics.RetrieveAPIView):
    queryset = Book.objects.all()
    serializer_class = BookSerializer
    
    def retrieve(self, request, *args, **kwargs):
        book = self.get_object()
        etag = hashlib.md5(str(book.updated_at).encode()).hexdigest()
        
        if request.headers.get("If-None-Match") == etag:
            return Response(status=304)        # غير معدل
        
        response = Response(BookSerializer(book).data)
        response["ETag"] = etag
        response["Cache-Control"] = "public, max-age=300"
        return response

# العميل يرسل تلقائيًا If-None-Match: "abc..."
# يمكن للخادم الرد بـ 304 بدون جسم -> سريع جدًا

from django.views.decorators.cache import cache_control
from django.utils.decorators import method_decorator

class PublicCatalogView(generics.ListAPIView):
    queryset = Product.objects.filter(public=True)
    serializer_class = ProductSerializer
    
    @method_decorator(cache_control(public=True, max_age=3600))
    def dispatch(self, *args, **kwargs):
        return super().dispatch(*args, **kwargs)

# Header : Cache-Control: public, max-age=3600
# ستقوم CDN والمتصفح بتخزينه

class UserDataView(APIView):
    @method_decorator(cache_control(private=True, max_age=60))
    def dispatch(self, *args, **kwargs):
        return super().dispatch(*args, **kwargs)
    
    # private = لا تخزين CDN، لكن المتصفح نعم