← 블로그 홈

Django Ninja로 테스트 참여자 관리 시스템 구축하기: GA 연동으로 참여자 수 자동 집계

12분 읽기
Django Python API Analytics
#Django #Django-Ninja #FastAPI #Google Analytics #REST API #Testing #User Management #Analytics
현대 웹 서비스에서 A/B 테스트나 사용자 피드백 수집은 필수적인 요소입니다. Django Ninja를 활용하여 테스트 참여자를 효율적으로 관리하고, Google Analytics를 연동해 실시간으로 참여자 수를 집계하는 시스템을 구축해보겠습니다.

현대 웹 서비스에서 A/B 테스트나 사용자 피드백 수집은 필수적인 요소입니다. Django Ninja를 활용하여 테스트 참여자를 효율적으로 관리하고, Google Analytics를 연동해 실시간으로 참여자 수를 집계하는 시스템을 구축해보겠습니다.

🎯 시스템 개요

이번 프로젝트에서 구축할 시스템의 핵심 기능은 다음과 같습니다:

  • Test 모델: 다양한 테스트 케이스 관리
  • Tester 모델: 테스트 참여자 정보 관리
  • Django Ninja API: 빠르고 타입 안전한 API 구축
  • Google Analytics 연동: 실시간 참여자 수 집계

📊 모델 설계

Test 모델

from django.db import models
from django.utils import timezone
import uuid

class Test(models.Model):
    """테스트 케이스를 관리하는 모델"""
    
    # 기본 정보
    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
    title = models.CharField(max_length=200, verbose_name="테스트 제목")
    description = models.TextField(verbose_name="테스트 설명")
    
    # 테스트 설정
    test_type = models.CharField(
        max_length=50,
        choices=[
            ('AB_TEST', 'A/B 테스트'),
            ('USABILITY', '사용성 테스트'),
            ('FEATURE', '기능 테스트'),
            ('SURVEY', '설문조사'),
        ],
        default='AB_TEST'
    )
    
    # 기간 관리
    start_date = models.DateTimeField(verbose_name="시작일")
    end_date = models.DateTimeField(verbose_name="종료일")
    
    # 참여자 제한
    max_participants = models.PositiveIntegerField(
        default=1000, 
        verbose_name="최대 참여자 수"
    )
    
    # GA 연동을 위한 키
    ga_event_name = models.CharField(
        max_length=100, 
        verbose_name="GA 이벤트 이름",
        help_text="Google Analytics에서 추적할 이벤트명"
    )
    
    # 상태 관리
    is_active = models.BooleanField(default=True, verbose_name="활성 상태")
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)
    
    class Meta:
        db_table = 'test'
        verbose_name = '테스트'
        verbose_name_plural = '테스트 목록'
    
    def __str__(self):
        return f"{self.title} ({self.get_test_type_display()})"
    
    @property
    def is_ongoing(self):
        """현재 진행 중인 테스트인지 확인"""
        now = timezone.now()
        return self.start_date <= now <= self.end_date and self.is_active
    
    @property
    def current_participants_count(self):
        """현재 참여자 수"""
        return self.testers.filter(participated_at__isnull=False).count()
    
    @property
    def can_participate(self):
        """참여 가능 여부"""
        return (
            self.is_ongoing and 
            self.current_participants_count < self.max_participants
        )

Tester 모델

from django.db import models
from django.contrib.auth.models import User
import hashlib
import uuid

class Tester(models.Model):
    """테스트 참여자 정보를 관리하는 모델"""
    
    # 기본 식별자
    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
    
    # 테스트 관계
    test = models.ForeignKey(
        Test, 
        on_delete=models.CASCADE, 
        related_name='testers',
        verbose_name="테스트"
    )
    
    # 사용자 정보 (익명 참여 지원)
    user = models.ForeignKey(
        User, 
        on_delete=models.CASCADE, 
        null=True, 
        blank=True,
        verbose_name="사용자"
    )
    
    # 익명 사용자를 위한 고유 식별자
    anonymous_id = models.CharField(
        max_length=64, 
        null=True, 
        blank=True,
        verbose_name="익명 ID"
    )
    
    # 참여자 메타데이터
    user_agent = models.TextField(verbose_name="User Agent")
    ip_address = models.GenericIPAddressField(verbose_name="IP 주소")
    
    # 참여 정보
    participated_at = models.DateTimeField(
        null=True, 
        blank=True, 
        verbose_name="참여 시간"
    )
    completed_at = models.DateTimeField(
        null=True, 
        blank=True, 
        verbose_name="완료 시간"
    )
    
    # GA 추적을 위한 클라이언트 ID
    ga_client_id = models.CharField(
        max_length=100, 
        null=True, 
        blank=True,
        verbose_name="GA 클라이언트 ID"
    )
    
    # 추가 데이터
    metadata = models.JSONField(
        default=dict, 
        verbose_name="추가 메타데이터"
    )
    
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)
    
    class Meta:
        db_table = 'tester'
        verbose_name = '테스트 참여자'
        verbose_name_plural = '테스트 참여자 목록'
        unique_together = [
            ('test', 'user'),  # 로그인 사용자는 테스트당 한 번만 참여
            ('test', 'anonymous_id'),  # 익명 사용자 중복 방지
        ]
    
    def __str__(self):
        if self.user:
            return f"{self.user.username} - {self.test.title}"
        return f"익명({self.anonymous_id[:8]}) - {self.test.title}"
    
    def save(self, *args, **kwargs):
        """익명 사용자 ID 자동 생성"""
        if not self.user and not self.anonymous_id:
            # IP + User-Agent 기반으로 고유 ID 생성
            unique_string = f"{self.ip_address}_{self.user_agent}"
            self.anonymous_id = hashlib.sha256(
                unique_string.encode()
            ).hexdigest()
        
        super().save(*args, **kwargs)
    
    @property
    def is_completed(self):
        """테스트 완료 여부"""
        return self.completed_at is not None
    
    @property
    def display_name(self):
        """표시용 이름"""
        if self.user:
            return self.user.username
        return f"익명_{self.anonymous_id[:8]}"

🚀 Django Ninja API 구축

API 스키마 정의

from ninja import Schema
from datetime import datetime
from typing import Optional, List
from uuid import UUID

class TestListSchema(Schema):
    """테스트 목록 응답 스키마"""
    id: UUID
    title: str
    description: str
    test_type: str
    start_date: datetime
    end_date: datetime
    max_participants: int
    current_participants_count: int
    can_participate: bool
    is_ongoing: bool

class TestDetailSchema(TestListSchema):
    """테스트 상세 응답 스키마"""
    ga_event_name: str
    created_at: datetime
    updated_at: datetime

class ParticipationRequestSchema(Schema):
    """테스트 참여 요청 스키마"""
    test_id: UUID
    ga_client_id: Optional[str] = None
    metadata: Optional[dict] = {}

class ParticipationResponseSchema(Schema):
    """테스트 참여 응답 스키마"""
    success: bool
    message: str
    tester_id: Optional[UUID] = None
    participation_token: Optional[str] = None

class TesterStatsSchema(Schema):
    """참여자 통계 스키마"""
    total_participants: int
    completed_participants: int
    completion_rate: float
    daily_participants: List[dict]

API 엔드포인트 구현

from ninja import NinjaAPI, Router
from django.shortcuts import get_object_or_404
from django.http import HttpRequest
from django.utils import timezone
from django.db.models import Count, Q
from django.db.models.functions import TruncDate
from typing import List
import secrets

api = NinjaAPI(title="Test Participant Management API", version="1.0.0")

# 테스트 관련 API
test_router = Router(tags=["Tests"])

@test_router.get("/", response=List[TestListSchema])
def list_active_tests(request: HttpRequest):
    """활성화된 테스트 목록 조회"""
    tests = Test.objects.filter(
        is_active=True,
        start_date__lte=timezone.now(),
        end_date__gte=timezone.now()
    ).prefetch_related('testers')
    
    return [
        TestListSchema(
            id=test.id,
            title=test.title,
            description=test.description,
            test_type=test.get_test_type_display(),
            start_date=test.start_date,
            end_date=test.end_date,
            max_participants=test.max_participants,
            current_participants_count=test.current_participants_count,
            can_participate=test.can_participate,
            is_ongoing=test.is_ongoing
        )
        for test in tests
    ]

@test_router.get("/{test_id}", response=TestDetailSchema)
def get_test_detail(request: HttpRequest, test_id: str):
    """테스트 상세 정보 조회"""
    test = get_object_or_404(Test, id=test_id, is_active=True)
    
    return TestDetailSchema(
        id=test.id,
        title=test.title,
        description=test.description,
        test_type=test.get_test_type_display(),
        start_date=test.start_date,
        end_date=test.end_date,
        max_participants=test.max_participants,
        current_participants_count=test.current_participants_count,
        can_participate=test.can_participate,
        is_ongoing=test.is_ongoing,
        ga_event_name=test.ga_event_name,
        created_at=test.created_at,
        updated_at=test.updated_at
    )

# 참여자 관련 API
participant_router = Router(tags=["Participants"])

@participant_router.post("/participate", response=ParticipationResponseSchema)
def participate_in_test(request: HttpRequest, data: ParticipationRequestSchema):
    """테스트 참여 신청"""
    test = get_object_or_404(Test, id=data.test_id)
    
    # 참여 가능 여부 확인
    if not test.can_participate:
        return ParticipationResponseSchema(
            success=False,
            message="참여할 수 없는 테스트입니다."
        )
    
    # 클라이언트 정보 수집
    ip_address = get_client_ip(request)
    user_agent = request.META.get('HTTP_USER_AGENT', '')
    
    try:
        # 기존 참여자 확인
        tester = None
        if request.user.is_authenticated:
            tester, created = Tester.objects.get_or_create(
                test=test,
                user=request.user,
                defaults={
                    'user_agent': user_agent,
                    'ip_address': ip_address,
                    'ga_client_id': data.ga_client_id,
                    'metadata': data.metadata,
                    'participated_at': timezone.now(),
                }
            )
        else:
            # 익명 사용자 처리
            anonymous_id = generate_anonymous_id(ip_address, user_agent)
            tester, created = Tester.objects.get_or_create(
                test=test,
                anonymous_id=anonymous_id,
                defaults={
                    'user_agent': user_agent,
                    'ip_address': ip_address,
                    'ga_client_id': data.ga_client_id,
                    'metadata': data.metadata,
                    'participated_at': timezone.now(),
                }
            )
        
        if created:
            # GA 이벤트 전송
            send_ga_event(test, tester, 'test_participation')
            
            # 참여 토큰 생성
            participation_token = secrets.token_urlsafe(32)
            
            return ParticipationResponseSchema(
                success=True,
                message="테스트 참여가 완료되었습니다.",
                tester_id=tester.id,
                participation_token=participation_token
            )
        else:
            return ParticipationResponseSchema(
                success=False,
                message="이미 참여한 테스트입니다."
            )
    
    except Exception as e:
        return ParticipationResponseSchema(
            success=False,
            message=f"참여 처리 중 오류가 발생했습니다: {str(e)}"
        )

@participant_router.post("/{tester_id}/complete")
def complete_test(request: HttpRequest, tester_id: str):
    """테스트 완료 처리"""
    tester = get_object_or_404(Tester, id=tester_id)
    
    if not tester.completed_at:
        tester.completed_at = timezone.now()
        tester.save()
        
        # GA 완료 이벤트 전송
        send_ga_event(tester.test, tester, 'test_completion')
    
    return {"success": True, "message": "테스트가 완료되었습니다."}

@participant_router.get("/{test_id}/stats", response=TesterStatsSchema)
def get_test_statistics(request: HttpRequest, test_id: str):
    """테스트 참여자 통계"""
    test = get_object_or_404(Test, id=test_id)
    
    # 기본 통계
    total_participants = test.testers.count()
    completed_participants = test.testers.filter(
        completed_at__isnull=False
    ).count()
    completion_rate = (
        completed_participants / total_participants * 100 
        if total_participants > 0 else 0
    )
    
    # 일별 참여자 수
    daily_stats = (
        test.testers
        .annotate(date=TruncDate('participated_at'))
        .values('date')
        .annotate(count=Count('id'))
        .order_by('date')
    )
    
    daily_participants = [
        {
            'date': stat['date'].isoformat() if stat['date'] else None,
            'count': stat['count']
        }
        for stat in daily_stats
    ]
    
    return TesterStatsSchema(
        total_participants=total_participants,
        completed_participants=completed_participants,
        completion_rate=round(completion_rate, 2),
        daily_participants=daily_participants
    )

# 라우터 등록
api.add_router("/tests", test_router)
api.add_router("/participants", participant_router)

📈 Google Analytics 연동

GA4 이벤트 전송 함수

import requests
import json
from django.conf import settings
from typing import Optional

def send_ga_event(
    test: Test, 
    tester: Tester, 
    event_type: str,
    custom_parameters: Optional[dict] = None
):
    """Google Analytics 4에 이벤트 전송"""
    
    if not hasattr(settings, 'GA4_MEASUREMENT_ID') or not hasattr(settings, 'GA4_API_SECRET'):
        return False
    
    # GA4 Measurement Protocol URL
    url = f"https://www.google-analytics.com/mp/collect"
    
    # 기본 이벤트 매개변수
    event_params = {
        'test_id': str(test.id),
        'test_title': test.title,
        'test_type': test.test_type,
        'participant_type': 'authenticated' if tester.user else 'anonymous',
    }
    
    # 사용자 정의 매개변수 추가
    if custom_parameters:
        event_params.update(custom_parameters)
    
    # GA4 이벤트 페이로드
    payload = {
        'client_id': tester.ga_client_id or str(tester.id),
        'events': [
            {
                'name': f"{test.ga_event_name}_{event_type}",
                'params': event_params
            }
        ]
    }
    
    # API 요청 파라미터
    params = {
        'measurement_id': settings.GA4_MEASUREMENT_ID,
        'api_secret': settings.GA4_API_SECRET,
    }
    
    try:
        response = requests.post(
            url,
            params=params,
            json=payload,
            timeout=5
        )
        return response.status_code == 204
    except requests.RequestException as e:
        print(f"GA 이벤트 전송 실패: {e}")
        return False

# 헬퍼 함수들
def get_client_ip(request):
    """클라이언트 실제 IP 주소 추출"""
    x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
    if x_forwarded_for:
        ip = x_forwarded_for.split(',')[0]
    else:
        ip = request.META.get('REMOTE_ADDR')
    return ip

def generate_anonymous_id(ip_address, user_agent):
    """익명 사용자 고유 ID 생성"""
    import hashlib
    unique_string = f"{ip_address}_{user_agent}"
    return hashlib.sha256(unique_string.encode()).hexdigest()

⚙️ 설정 및 환경 구성

settings.py 설정

# Django Ninja 설정
INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'ninja',  # Django Ninja 추가
    'your_app',  # 앱 이름
]

# Google Analytics 설정
GA4_MEASUREMENT_ID = os.getenv('GA4_MEASUREMENT_ID')
GA4_API_SECRET = os.getenv('GA4_API_SECRET')

# CORS 설정 (필요한 경우)
CORS_ALLOW_ALL_ORIGINS = True  # 개발 환경에서만
CORS_ALLOWED_ORIGINS = [
    "https://yourdomain.com",
]

# API 설정
NINJA_PAGINATION_CLASS = 'ninja.pagination.LimitOffsetPagination'
NINJA_PAGINATION_PER_PAGE = 20

URL 설정

from django.contrib import admin
from django.urls import path
from .api import api

urlpatterns = [
    path('admin/', admin.site.urls),
    path('api/', api.urls),  # API 엔드포인트 등록
]

🎨 프론트엔드 연동

JavaScript 클라이언트 예제

class TestParticipantManager {
    constructor(apiBaseUrl) {
        this.apiBaseUrl = apiBaseUrl;
        this.gaClientId = this.getGAClientId();
    }
    
    // Google Analytics Client ID 추출
    getGAClientId() {
        if (typeof gtag !== 'undefined') {
            gtag('get', 'GA_MEASUREMENT_ID', 'client_id', (clientId) => {
                return clientId;
            });
        }
        return null;
    }
    
    // 활성 테스트 목록 조회
    async getActiveTests() {
        try {
            const response = await fetch(`${this.apiBaseUrl}/tests/`);
            return await response.json();
        } catch (error) {
            console.error('테스트 목록 조회 실패:', error);
            return [];
        }
    }
    
    // 테스트 참여
    async participateInTest(testId, metadata = {}) {
        try {
            const response = await fetch(`${this.apiBaseUrl}/participants/participate`, {
                method: 'POST',
                headers: {
                    'Content-Type': 'application/json',
                    'X-CSRFToken': this.getCsrfToken(),
                },
                body: JSON.stringify({
                    test_id: testId,
                    ga_client_id: this.gaClientId,
                    metadata: metadata
                })
            });
            
            const result = await response.json();
            
            if (result.success) {
                // GA 이벤트 수동 전송 (백업)
                if (typeof gtag !== 'undefined') {
                    gtag('event', 'test_participation', {
                        'test_id': testId,
                        'participation_method': 'api'
                    });
                }
                
                // 로컬 스토리지에 참여 정보 저장
                localStorage.setItem(`test_${testId}_token`, result.participation_token);
            }
            
            return result;
        } catch (error) {
            console.error('테스트 참여 실패:', error);
            return { success: false, message: '네트워크 오류가 발생했습니다.' };
        }
    }
    
    // 테스트 완료
    async completeTest(testerId) {
        try {
            const response = await fetch(
                `${this.apiBaseUrl}/participants/${testerId}/complete`,
                {
                    method: 'POST',
                    headers: {
                        'X-CSRFToken': this.getCsrfToken(),
                    }
                }
            );
            
            const result = await response.json();
            
            if (result.success && typeof gtag !== 'undefined') {
                gtag('event', 'test_completion', {
                    'tester_id': testerId
                });
            }
            
            return result;
        } catch (error) {
            console.error('테스트 완료 처리 실패:', error);
            return { success: false, message: '완료 처리 중 오류가 발생했습니다.' };
        }
    }
    
    // 테스트 통계 조회
    async getTestStats(testId) {
        try {
            const response = await fetch(`${this.apiBaseUrl}/participants/${testId}/stats`);
            return await response.json();
        } catch (error) {
            console.error('통계 조회 실패:', error);
            return null;
        }
    }
    
    // CSRF 토큰 추출
    getCsrfToken() {
        return document.querySelector('[name=csrfmiddlewaretoken]')?.value || '';
    }
}

// 사용 예제
const testManager = new TestParticipantManager('/api');

// 테스트 참여 버튼 이벤트
document.getElementById('participate-btn').addEventListener('click', async () => {
    const testId = document.getElementById('test-select').value;
    const result = await testManager.participateInTest(testId, {
        source: 'web_form',
        browser: navigator.userAgent
    });
    
    if (result.success) {
        alert('테스트 참여가 완료되었습니다!');
        // UI 업데이트 로직
    } else {
        alert(result.message);
    }
});

📊 관리자 인터페이스 구성

Django Admin 커스터마이징

from django.contrib import admin
from django.db.models import Count
from django.utils.html import format_html
from django.urls import reverse
from django.utils.safestring import mark_safe

@admin.register(Test)
class TestAdmin(admin.ModelAdmin):
    list_display = [
        'title', 
        'test_type', 
        'is_active', 
        'participant_count',
        'completion_rate',
        'start_date', 
        'end_date',
        'status_badge'
    ]
    
    list_filter = [
        'test_type', 
        'is_active', 
        'start_date', 
        'end_date'
    ]
    
    search_fields = ['title', 'description']
    
    readonly_fields = [
        'participant_count', 
        'completion_rate', 
        'created_at', 
        'updated_at'
    ]
    
    fieldsets = (
        ('기본 정보', {
            'fields': ('title', 'description', 'test_type')
        }),
        ('기간 설정', {
            'fields': ('start_date', 'end_date', 'is_active')
        }),
        ('참여자 설정', {
            'fields': ('max_participants', 'participant_count', 'completion_rate')
        }),
        ('Analytics 설정', {
            'fields': ('ga_event_name',)
        }),
        ('메타 정보', {
            'fields': ('created_at', 'updated_at'),
            'classes': ('collapse',)
        })
    )
    
    def get_queryset(self, request):
        return super().get_queryset(request).annotate(
            participant_count=Count('testers'),
            completed_count=Count('testers', filter=models.Q(testers__completed_at__isnull=False))
        )
    
    def participant_count(self, obj):
        count = getattr(obj, 'participant_count', 0)
        url = reverse('admin:your_app_tester_changelist') + f'?test__id__exact={obj.id}'
        return format_html('<a href="{}">{} 명</a>', url, count)
    participant_count.short_description = '참여자 수'
    participant_count.admin_order_field = 'participant_count'
    
    def completion_rate(self, obj):
        total = getattr(obj, 'participant_count', 0)
        completed = getattr(obj, 'completed_count', 0)
        if total == 0:
            return "0%"
        rate = (completed / total) * 100
        return f"{rate:.1f}%"
    completion_rate.short_description = '완료율'
    
    def status_badge(self, obj):
        if obj.is_ongoing:
            return mark_safe('<span style="color: green;">●</span> 진행중')
        elif obj.is_active:
            return mark_safe('<span style="color: orange;">●</span> 대기중')
        else:
            return mark_safe('<span style="color: red;">●</span> 비활성')
    status_badge.short_description = '상태'

@admin.register(Tester)
class TesterAdmin(admin.ModelAdmin):
    list_display = [
        'display_name',
        'test',
        'participant_type',
        'participated_at',
        'completed_at',
        'is_completed'
    ]
    
    list_filter = [
        'test',
        'participated_at',
        'completed_at'
    ]
    
    search_fields = [
        'user__username',
        'anonymous_id',
        'test__title'
    ]
    
    readonly_fields = [
        'anonymous_id',
        'user_agent',
        'ip_address',
        'created_at',
        'updated_at'
    ]
    
    def participant_type(self, obj):
        return "로그인 사용자" if obj.user else "익명 사용자"
    participant_type.short_description = '참여자 유형'
    
    def display_name(self, obj):
        return obj.display_name
    display_name.short_description = '참여자'

🔍 성능 최적화 및 모니터링

데이터베이스 인덱스 최적화

# models.py에 추가할 인덱스들

class Test(models.Model):
    # ... 기존 필드들 ...
    
    class Meta:
        db_table = 'test'
        indexes = [
            models.Index(fields=['is_active', 'start_date', 'end_date']),
            models.Index(fields=['ga_event_name']),
            models.Index(fields=['created_at']),
        ]

class Tester(models.Model):
    # ... 기존 필드들 ...
    
    class Meta:
        db_table = 'tester'
        indexes = [
            models.Index(fields=['test', 'participated_at']),
            models.Index(fields=['test', 'completed_at']),
            models.Index(fields=['anonymous_id']),
            models.Index(fields=['ga_client_id']),
        ]

API 성능 모니터링

from ninja import NinjaAPI
from django.middleware.decorators import decorator_from_middleware
from django.utils.decorators import method_decorator
import time
import logging

# 성능 로깅 미들웨어
class APIPerformanceMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
        self.logger = logging.getLogger('api_performance')

    def __call__(self, request):
        start_time = time.time()
        response = self.get_response(request)
        duration = time.time() - start_time
        
        if request.path.startswith('/api/'):
            self.logger.info(
                f"API Call: {request.method} {request.path} "
                f"- Duration: {duration:.3f}s - Status: {response.status_code}"
            )
        
        return response

# 캐싱을 활용한 통계 조회 최적화
from django.core.cache import cache

@participant_router.get("/{test_id}/stats", response=TesterStatsSchema)
def get_test_statistics(request: HttpRequest, test_id: str):
    """캐시를 활용한 테스트 참여자 통계"""
    cache_key = f"test_stats_{test_id}"
    cached_stats = cache.get(cache_key)
    
    if cached_stats:
        return TesterStatsSchema(**cached_stats)
    
    test = get_object_or_404(Test, id=test_id)
    
    # 통계 계산 (기존 로직)
    stats_data = {
        'total_participants': test.testers.count(),
        'completed_participants': test.testers.filter(completed_at__isnull=False).count(),
        # ... 기타 통계
    }
    
    # 5분간 캐시 저장
    cache.set(cache_key, stats_data, 300)
    
    return TesterStatsSchema(**stats_data)

🚀 배포 및 운영 가이드

Docker를 활용한 배포

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# 의존성 설치
COPY requirements.txt .
RUN pip install -r requirements.txt

# 애플리케이션 코드 복사
COPY . .

# 정적 파일 수집
RUN python manage.py collectstatic --noinput

# 서버 실행
EXPOSE 8000
CMD ["gunicorn", "--bind", "0.0.0.0:8000", "project.wsgi:application"]

# docker-compose.yml
version: '3.8'

services:
  web:
    build: .
    ports:
      - "8000:8000"
    depends_on:
      - db
      - redis
    environment:
      - DEBUG=False
      - GA4_MEASUREMENT_ID=${GA4_MEASUREMENT_ID}
      - GA4_API_SECRET=${GA4_API_SECRET}
    
  db:
    image: postgres:14
    environment:
      - POSTGRES_DB=testdb
      - POSTGRES_USER=testuser
      - POSTGRES_PASSWORD=testpass
    volumes:
      - postgres_data:/var/lib/postgresql/data
  
  redis:
    image: redis:7-alpine
    
volumes:
  postgres_data:

📈 결론 및 활용 방안

이번 프로젝트에서 구축한 Django Ninja 기반 테스트 참여자 관리 시스템은 다음과 같은 장점을 제공합니다:

✅ 주요 장점

  1. 타입 안전성: Django Ninja의 Pydantic 기반 스키마로 API 안정성 보장
  2. 실시간 분석: Google Analytics 연동으로 즉시 데이터 확인 가능
  3. 유연한 참여자 관리: 로그인/익명 사용자 모두 지원
  4. 확장성: 다양한 테스트 유형과 메타데이터 지원

🎯 활용 사례

  • A/B 테스트: 새로운 기능이나 디자인의 효과 측정
  • 사용성 테스트: UI/UX 개선을 위한 사용자 피드백 수집
  • 베타 테스트: 신규 서비스 출시 전 사용자 검증
  • 설문조사: 대규모 사용자 의견 수집

🔮 향후 개선 방안

  • 실시간 대시보드: WebSocket을 활용한 실시간 통계 업데이트
  • A/B 테스트 자동화: 트래픽 분할 및 결과 분석 자동화
  • 머신러닝 연동: 참여자 행동 패턴 분석 및 예측
  • 다국어 지원: 글로벌 서비스를 위한 국제화

Django Ninja의 현대적인 API 개발 방식과 Google Analytics의 강력한 분석 기능을 결합하면, 효율적이고 인사이트가 풍부한 테스트 환경을 구축할 수 있습니다. 이 시스템을 기반으로 데이터 기반의 의사결정을 내려보세요!

이 글을 공유해보세요!