Módulos Django
5
leads, pipelines, activities, users, workspaces
Endpoints DRF
23
CRUD + custom actions
Cache TTL
15m
Dashboard aggregations en Redis
Entornos
3
dev / staging / prod
Flujo de una Request
React Client
→
RequestLogging Middleware
→
TokenAuth + RolePermission
→
LeadViewSet
→
LeadService
→
ORM + PostgreSQL
...
←
LeadSerializer
←
+ Redis cache check
Estructura del Proyecto
leadflow/
config/
__init__.py
settings/
base.py # SECRET_KEY, INSTALLED_APPS, DATABASES, REST_FRAMEWORK
development.py # DEBUG=True, debug_toolbar, email console
production.py # SSL, HSTS, logging a fichero, Sentry DSN
staging.py # igual que prod pero con DEBUG=False y BD staging
urls.py # router DRF + health check
wsgi.py asgi.py
apps/
leads/ # módulo principal
models.py serializers.py views.py
services.py permissions.py filters.py
signals.py urls.py
tests/ # test_models.py, test_views.py, test_services.py
pipelines/ activities/ users/ workspaces/
middleware/
request_logging.py workspace_context.py
manage.py requirements/base.txt dev.txt prod.txt
Modelos — apps/leads/models.py
apps/leads/models.py
· Lead + LeadQuerySet · N+1 safe
from django.db import models
from django.core.validators import MinValueValidator
class LeadQuerySet(models.QuerySet):
"""QuerySet reutilizable — evita N+1 y centraliza filtros frecuentes."""
def active(self):
return self.filter(status__in=['new', 'contacted', 'qualified'])
def for_workspace(self, workspace_id):
return self.filter(workspace_id=workspace_id)
def with_assignee(self):
# select_related evita query extra por cada lead
return self.select_related('assignee', 'pipeline')
def with_activities(self):
return self.prefetch_related('activities', 'tags')
def high_value(self, min_value=5000):
return self.filter(deal_value__gte=min_value)
def search(self, q):
return self.filter(
models.Q(name__icontains=q) |
models.Q(company__icontains=q) |
models.Q(email__icontains=q)
)
class Lead(models.Model):
class Status(models.TextChoices):
NEW = 'new', 'Nuevo'
CONTACTED = 'contacted', 'Contactado'
QUALIFIED = 'qualified', 'Cualificado'
CLOSED_WON = 'won', 'Cerrado Ganado'
CLOSED_LOST= 'lost', 'Cerrado Perdido'
workspace = models.ForeignKey('workspaces.Workspace', on_delete=models.CASCADE, related_name='leads')
assignee = models.ForeignKey('users.User', null=True, blank=True, on_delete=models.SET_NULL)
pipeline = models.ForeignKey('pipelines.Pipeline', on_delete=models.CASCADE)
name = models.CharField(max_length=200)
email = models.EmailField()
company = models.CharField(max_length=200, blank=True)
phone = models.CharField(max_length=30, blank=True)
status = models.CharField(max_length=20, choices=Status.choices, default=Status.NEW)
deal_value = models.DecimalField(max_digits=12, decimal_places=2, default=0,
validators=[MinValueValidator(0)])
score = models.PositiveSmallIntegerField(default=0) # 0-100 calculado por LeadService
tags = models.ManyToManyField('Tag', blank=True)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
objects = LeadQuerySet.as_manager() # QuerySet como manager directo
class Meta:
db_table = 'leads'
ordering = ['-created_at']
indexes = [
models.Index(fields=['workspace', 'status']),
models.Index(fields=['assignee', '-created_at']),
models.Index(fields=['-deal_value']),
]
Service Layer — apps/leads/services.py
apps/leads/services.py
· Lógica de negocio separada de la vista
from django.db import transaction
from django.core.cache import cache
from .models import Lead
class LeadService:
"""Toda la lógica de negocio de leads aquí, no en la vista."""
@staticmethod
def calculate_score(lead: Lead) -> int:
"""Score basado en deal_value, completitud y actividad reciente."""
score = 0
if lead.deal_value > 10000: score += 40
elif lead.deal_value > 3000: score += 25
if lead.company: score += 10
if lead.phone: score += 10
if lead.activities.filter(type='call').exists(): score += 20
if lead.activities.filter(type='demo').exists(): score += 20
return min(score, 100)
@staticmethod
@transaction.atomic
def convert_to_won(lead: Lead, closed_by) -> Lead:
"""Cierra el deal: actualiza estado, recalcula, invalida cache."""
lead.status = Lead.Status.CLOSED_WON
lead.assignee = closed_by
lead.score = LeadService.calculate_score(lead)
lead.save(update_fields=['status', 'assignee', 'score', 'updated_at'])
# Invalida dashboard cache del workspace
cache.delete(f'dashboard_{lead.workspace_id}')
# Signal post_save dispara el webhook (ver signals.py)
return lead
@staticmethod
def get_dashboard_stats(workspace_id: int) -> dict:
"""Cache-first: costoso en BD, TTL 15 min."""
cache_key = f'dashboard_{workspace_id}'
stats = cache.get(cache_key)
if stats is None:
from django.db.models import Count, Sum
qs = Lead.objects.for_workspace(workspace_id)
stats = {
'total': qs.count(),
'won': qs.filter(status='won').count(),
'value': qs.filter(status='won').aggregate(t=Sum('deal_value'))['t'] or 0,
'by_status': list(qs.values('status').annotate(n=Count('id'))),
}
cache.set(cache_key, stats, timeout=60 * 15)
return stats
ViewSet DRF + Permisos por Rol
apps/leads/views.py
· LeadViewSet
class LeadViewSet(viewsets.ModelViewSet):
permission_classes = [IsWorkspaceMember]
filter_backends = [DjangoFilterBackend,
filters.SearchFilter,
filters.OrderingFilter]
filterset_class = LeadFilter
search_fields = ['name', 'company', 'email']
ordering_fields = ['deal_value', 'created_at', 'score']
def get_queryset(self):
# multi-tenant: solo leads del workspace
return Lead.objects\
.for_workspace(self.request.user.workspace_id)\
.with_assignee()
def get_serializer_class(self):
if self.action in ['create', 'update']:
return LeadWriteSerializer
return LeadReadSerializer
@action(detail=True, methods=['post'])
def close_won(self, request, pk=None):
lead = self.get_object()
lead = LeadService.convert_to_won(lead, request.user)
return Response(LeadReadSerializer(lead).data)
@action(detail=False, methods=['get'])
def dashboard(self, request):
stats = LeadService.get_dashboard_stats(
request.user.workspace_id
)
return Response(stats)
apps/leads/permissions.py
· Rol-based
class IsWorkspaceMember(BasePermission):
"""Solo miembros del mismo workspace."""
def has_permission(self, request, view):
return request.user.is_authenticated
def has_object_permission(self, request, view, obj):
return obj.workspace_id == request.user.workspace_id
class IsAdminOrReadOnly(BasePermission):
"""Admin escribe, el resto solo lee."""
def has_permission(self, request, view):
if request.method in SAFE_METHODS:
return request.user.is_authenticated
return request.user.role == 'admin'
class CanDeleteLead(BasePermission):
"""Solo admin o el asignado puede borrar."""
def has_object_permission(self, request, view, obj):
return (request.user.role == 'admin' or
obj.assignee == request.user)
Signals (Webhooks) + Middleware de Logging
apps/leads/signals.py
· Webhook al cerrar deal
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import Lead
from .tasks import send_webhook_task # Celery
@receiver(post_save, sender=Lead)
def on_lead_won(sender, instance, created, **kwargs):
"""Dispara webhook async al cerrar un deal ganado."""
if not created and instance.status == Lead.Status.CLOSED_WON:
payload = {
'event': 'lead.won',
'lead_id': instance.id,
'deal_value': str(instance.deal_value),
'workspace': instance.workspace_id,
}
# Tarea Celery — no bloquea la respuesta HTTP
send_webhook_task.delay(
workspace_id=instance.workspace_id,
payload=payload
)
# apps/leads/apps.py
class LeadsConfig(AppConfig):
name = 'apps.leads'
def ready(self):
import apps.leads.signals
middleware/request_logging.py
· Timing + workspace context
import time, logging
from django.utils.deprecation import MiddlewareMixin
logger = logging.getLogger('leadflow.api')
class RequestLoggingMiddleware(MiddlewareMixin):
"""Loguea método, path, status y duración de cada request."""
def process_request(self, request):
request._start = time.time()
def process_response(self, request, response):
duration = time.time() - getattr(request, '_start', time.time())
logger.info(
'%s %s — %d — %.3fs — ws:%s',
request.method, request.path,
response.status_code, duration,
getattr(request.user, 'workspace_id', '-')
)
return response
class WorkspaceContextMiddleware(MiddlewareMixin):
"""Inyecta workspace_id en cada request autenticado."""
def process_request(self, request):
if request.user.is_authenticated:
request.workspace_id = request.user.workspace_id
Mapa de Patrones Aplicados — LeadFlow
Multi-tenant seguro
Cada QuerySet filtra por workspace_id desde el método get_queryset(), nunca expone datos cruzados.
LeadQuerySet.for_workspace()
IsWorkspaceMember permission
WorkspaceContextMiddleware
Zero N+1 queries
select_related en FKs (assignee, pipeline) y prefetch_related en M2M (tags, activities) configurados en el QuerySet.
with_assignee() → select_related
with_activities() → prefetch_related
3 índices compuestos en Meta
Async por defecto
Webhooks y emails delegados a tareas Celery con Redis como broker; la respuesta HTTP no espera operaciones lentas.
Signal → send_webhook_task.delay()
send_confirmation_email.delay()
Celery + Redis como broker