Django Ninja와 스케줄링으로 프로젝트 마감일 관리하기 - 실전 가이드
Django Ninja와 다양한 스케줄링 도구를 활용해 프로젝트 마감일을 효율적으로 관리하는 방법을 알아봅니다. Celery Beat, APScheduler, Crontab 등 다양한 접근법을 비교분석합니다.
Django Ninja와 스케줄링으로 프로젝트 마감일 관리하기
프로젝트 개발에서 마감일 관리는 성공의 핵심 요소 중 하나입니다. Django Ninja를 활용한 API와 함께 스케줄링 시스템을 구축하여 효율적인 프로젝트 관리 시스템을 만드는 방법을 알아보겠습니다.
📋 목차
- Django Ninja 기본 설정
- 프로젝트 모델 설계
- 스케줄링 방법 비교
- Celery Beat를 이용한 구현
- APScheduler를 이용한 구현
- Django-cron을 이용한 구현
- 알림 시스템 구축
- 실전 예제
Django Ninja 기본 설정
먼저 Django Ninja를 설치하고 기본 설정을 진행합니다.
pip install django-ninja
pip install celery[redis]
pip install APScheduler
pip install django-cron
# settings.py
INSTALLED_APPS = [
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
'ninja',
'django_cron',
'project_manager',
]
# Celery 설정
CELERY_BROKER_URL = 'redis://localhost:6379'
CELERY_RESULT_BACKEND = 'redis://localhost:6379'
CELERY_TIMEZONE = 'Asia/Seoul'
CELERY_ENABLE_UTC = True
프로젝트 모델 설계
프로젝트 마감일 관리를 위한 모델을 설계해보겠습니다.
# models.py
from django.db import models
from django.contrib.auth.models import User
from django.utils import timezone
from datetime import timedelta
class Project(models.Model):
PRIORITY_CHOICES = [
('low', '낮음'),
('medium', '보통'),
('high', '높음'),
('urgent', '긴급'),
]
STATUS_CHOICES = [
('planning', '기획'),
('development', '개발'),
('testing', '테스트'),
('deployment', '배포'),
('completed', '완료'),
('cancelled', '취소'),
]
name = models.CharField(max_length=200, verbose_name="프로젝트명")
description = models.TextField(blank=True, verbose_name="설명")
owner = models.ForeignKey(User, on_delete=models.CASCADE, verbose_name="담당자")
start_date = models.DateTimeField(verbose_name="시작일")
deadline = models.DateTimeField(verbose_name="마감일")
status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='planning')
priority = models.CharField(max_length=10, choices=PRIORITY_CHOICES, default='medium')
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
ordering = ['-priority', 'deadline']
@property
def days_until_deadline(self):
"""마감일까지 남은 일수"""
return (self.deadline - timezone.now()).days
@property
def is_overdue(self):
"""마감일 초과 여부"""
return timezone.now() > self.deadline
@property
def is_approaching_deadline(self):
"""마감일 임박 여부 (3일 이내)"""
return 0 <= self.days_until_deadline <= 3
class Task(models.Model):
project = models.ForeignKey(Project, on_delete=models.CASCADE, related_name='tasks')
title = models.CharField(max_length=200)
description = models.TextField(blank=True)
assignee = models.ForeignKey(User, on_delete=models.CASCADE)
due_date = models.DateTimeField()
completed = models.BooleanField(default=False)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
ordering = ['due_date']
class Notification(models.Model):
NOTIFICATION_TYPES = [
('deadline_warning', '마감일 경고'),
('overdue', '마감일 초과'),
('task_completed', '작업 완료'),
('project_updated', '프로젝트 업데이트'),
]
user = models.ForeignKey(User, on_delete=models.CASCADE)
project = models.ForeignKey(Project, on_delete=models.CASCADE, null=True, blank=True)
notification_type = models.CharField(max_length=20, choices=NOTIFICATION_TYPES)
title = models.CharField(max_length=200)
message = models.TextField()
is_read = models.BooleanField(default=False)
created_at = models.DateTimeField(auto_now_add=True)
스케줄링 방법 비교
프로젝트 마감일 관리를 위한 다양한 스케줄링 방법을 비교해보겠습니다.
| 방법 | 장점 | 단점 | 사용 사례 |
|---|---|---|---|
| Celery Beat | - 분산 처리 가능 - Redis/RabbitMQ 지원 - 강력한 모니터링 |
- 설정 복잡 - 인프라 비용 |
대규모 서비스, 복잡한 작업 |
| APScheduler | - 설정 간단 - 다양한 스케줄러 - 메모리 기반 |
- 서버 재시작시 초기화 - 단일 서버만 지원 |
중소규모, 단순한 작업 |
| Django-cron | - Django 친화적 - 관리자에서 관리 - 간단한 설정 |
- 기능 제한적 - 정밀도 낮음 |
일간/주간 정기 작업 |
| 시스템 Cron | - 시스템 레벨 - 안정성 높음 - 자원 효율적 |
- Django 통합 어려움 - 로깅 복잡 |
시스템 관리 작업 |
Celery Beat를 이용한 구현
가장 강력하고 확장 가능한 Celery Beat를 사용한 구현입니다.
# celery.py
from celery import Celery
from django.conf import settings
import os
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
app = Celery('myproject')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()
# 스케줄링 설정
app.conf.beat_schedule = {
'check-deadlines': {
'task': 'project_manager.tasks.check_project_deadlines',
'schedule': 60.0, # 1분마다 실행
},
'send-daily-report': {
'task': 'project_manager.tasks.send_daily_report',
'schedule': crontab(hour=9, minute=0), # 매일 오전 9시
},
'cleanup-completed-projects': {
'task': 'project_manager.tasks.cleanup_completed_projects',
'schedule': crontab(hour=2, minute=0, day_of_week=1), # 매주 월요일 오전 2시
},
}
# tasks.py
from celery import shared_task
from django.utils import timezone
from django.contrib.auth.models import User
from .models import Project, Notification
from .notifications import send_notification, send_email
from datetime import timedelta
@shared_task
def check_project_deadlines():
"""프로젝트 마감일을 체크하고 알림을 발송합니다."""
now = timezone.now()
# 마감일이 임박한 프로젝트 (3일, 1일, 당일)
warning_periods = [
(timedelta(days=3), "3일 후 마감일입니다."),
(timedelta(days=1), "내일 마감일입니다!"),
(timedelta(hours=6), "6시간 후 마감일입니다!!"),
]
for period, message in warning_periods:
deadline_time = now + period
projects = Project.objects.filter(
deadline__date=deadline_time.date(),
status__in=['planning', 'development', 'testing']
)
for project in projects:
# 중복 알림 방지
existing_notification = Notification.objects.filter(
user=project.owner,
project=project,
notification_type='deadline_warning',
created_at__date=now.date()
).exists()
if not existing_notification:
Notification.objects.create(
user=project.owner,
project=project,
notification_type='deadline_warning',
title=f"프로젝트 '{project.name}' 마감일 임박",
message=f"{message}\n프로젝트: {project.name}\n마감일: {project.deadline.strftime('%Y-%m-%d %H:%M')}"
)
# 이메일 알림 발송
send_deadline_notification.delay(project.id, message)
# 마감일을 초과한 프로젝트
overdue_projects = Project.objects.filter(
deadline__lt=now,
status__in=['planning', 'development', 'testing']
)
for project in overdue_projects:
# 일일 한번만 알림
today_overdue_notification = Notification.objects.filter(
user=project.owner,
project=project,
notification_type='overdue',
created_at__date=now.date()
).exists()
if not today_overdue_notification:
days_overdue = (now - project.deadline).days
Notification.objects.create(
user=project.owner,
project=project,
notification_type='overdue',
title=f"프로젝트 '{project.name}' 마감일 초과",
message=f"마감일을 {days_overdue}일 초과했습니다.\n즉시 조치가 필요합니다."
)
send_overdue_notification.delay(project.id, days_overdue)
@shared_task
def send_deadline_notification(project_id, message):
"""마감일 알림 이메일을 발송합니다."""
try:
project = Project.objects.get(id=project_id)
send_email(
to=project.owner.email,
subject=f"[프로젝트 알림] {project.name} 마감일 임박",
message=message,
project=project
)
except Project.DoesNotExist:
pass
@shared_task
def send_overdue_notification(project_id, days_overdue):
"""마감일 초과 알림을 발송합니다."""
try:
project = Project.objects.get(id=project_id)
send_email(
to=project.owner.email,
subject=f"[긴급] {project.name} 마감일 초과",
message=f"프로젝트가 마감일을 {days_overdue}일 초과했습니다.",
project=project,
urgent=True
)
except Project.DoesNotExist:
pass
@shared_task
def send_daily_report():
"""일일 프로젝트 현황 보고서를 발송합니다."""
users = User.objects.filter(is_active=True)
for user in users:
user_projects = Project.objects.filter(
owner=user,
status__in=['planning', 'development', 'testing']
)
if user_projects.exists():
report_data = {
'total_projects': user_projects.count(),
'approaching_deadline': user_projects.filter(
deadline__lte=timezone.now() + timedelta(days=3)
).count(),
'overdue': user_projects.filter(
deadline__lt=timezone.now()
).count(),
}
send_daily_report_email.delay(user.id, report_data)
@shared_task
def send_daily_report_email(user_id, report_data):
"""일일 보고서 이메일을 발송합니다."""
try:
user = User.objects.get(id=user_id)
send_email(
to=user.email,
subject="일일 프로젝트 현황 보고서",
template="daily_report",
context=report_data
)
except User.DoesNotExist:
pass
APScheduler를 이용한 구현
더 간단한 설정으로 중소규모 프로젝트에 적합한 APScheduler 구현입니다.
# scheduler.py
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
from django.conf import settings
import atexit
import logging
logger = logging.getLogger(__name__)
def start_scheduler():
"""스케줄러를 시작합니다."""
scheduler = BackgroundScheduler(timezone=settings.TIME_ZONE)
# 프로젝트 마감일 체크 (5분마다)
scheduler.add_job(
func=check_deadlines_job,
trigger='interval',
minutes=5,
id='check_deadlines',
name='Check project deadlines',
replace_existing=True
)
# 일일 보고서 (매일 오전 9시)
scheduler.add_job(
func=daily_report_job,
trigger=CronTrigger(hour=9, minute=0),
id='daily_report',
name='Send daily report',
replace_existing=True
)
# 주간 정리 (매주 일요일 오후 6시)
scheduler.add_job(
func=weekly_cleanup_job,
trigger=CronTrigger(day_of_week=6, hour=18, minute=0),
id='weekly_cleanup',
name='Weekly cleanup',
replace_existing=True
)
scheduler.start()
logger.info("Scheduler started")
# 애플리케이션 종료시 스케줄러 정지
atexit.register(lambda: scheduler.shutdown())
def check_deadlines_job():
"""마감일 체크 작업"""
from .models import Project, Notification
from .notifications import send_notification
from django.utils import timezone
from datetime import timedelta
now = timezone.now()
# 임박한 마감일 체크
for project in Project.objects.filter(status__in=['planning', 'development', 'testing']):
time_until_deadline = project.deadline - now
# 알림 조건 확인
if time_until_deadline <= timedelta(days=3) and time_until_deadline > timedelta(0):
create_deadline_notification(project, time_until_deadline)
elif time_until_deadline <= timedelta(0):
create_overdue_notification(project, abs(time_until_deadline))
def create_deadline_notification(project, time_until_deadline):
"""마감일 임박 알림 생성"""
from .models import Notification
from django.utils import timezone
# 중복 알림 방지
today = timezone.now().date()
if Notification.objects.filter(
project=project,
notification_type='deadline_warning',
created_at__date=today
).exists():
return
days = time_until_deadline.days
hours = time_until_deadline.seconds // 3600
if days > 0:
message = f"{days}일 후 마감일입니다."
else:
message = f"{hours}시간 후 마감일입니다!"
Notification.objects.create(
user=project.owner,
project=project,
notification_type='deadline_warning',
title=f"프로젝트 '{project.name}' 마감일 임박",
message=message
)
def create_overdue_notification(project, time_overdue):
"""마감일 초과 알림 생성"""
from .models import Notification
from django.utils import timezone
today = timezone.now().date()
if Notification.objects.filter(
project=project,
notification_type='overdue',
created_at__date=today
).exists():
return
days_overdue = time_overdue.days
Notification.objects.create(
user=project.owner,
project=project,
notification_type='overdue',
title=f"프로젝트 '{project.name}' 마감일 초과",
message=f"마감일을 {days_overdue}일 초과했습니다."
)
# apps.py
from django.apps import AppConfig
class ProjectManagerConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'project_manager'
def ready(self):
from . import scheduler
if os.environ.get('RUN_MAIN'): # 개발 서버에서 중복 실행 방지
scheduler.start_scheduler()
Django Ninja API 구현
프로젝트 관리를 위한 REST API를 구현해보겠습니다.
# schemas.py
from ninja import Schema
from datetime import datetime
from typing import Optional, List
class ProjectCreateSchema(Schema):
name: str
description: Optional[str] = None
start_date: datetime
deadline: datetime
priority: str = 'medium'
class ProjectUpdateSchema(Schema):
name: Optional[str] = None
description: Optional[str] = None
deadline: Optional[datetime] = None
status: Optional[str] = None
priority: Optional[str] = None
class ProjectSchema(Schema):
id: int
name: str
description: Optional[str]
owner: str
start_date: datetime
deadline: datetime
status: str
priority: str
days_until_deadline: int
is_overdue: bool
is_approaching_deadline: bool
created_at: datetime
updated_at: datetime
class TaskCreateSchema(Schema):
title: str
description: Optional[str] = None
assignee_id: int
due_date: datetime
class TaskSchema(Schema):
id: int
title: str
description: Optional[str]
assignee: str
due_date: datetime
completed: bool
created_at: datetime
class NotificationSchema(Schema):
id: int
notification_type: str
title: str
message: str
is_read: bool
created_at: datetime
# api.py
from ninja import NinjaAPI, Query
from ninja.security import django_auth
from django.shortcuts import get_object_or_404
from django.contrib.auth.models import User
from django.utils import timezone
from django.db.models import Q, Count
from typing import List, Optional
from .models import Project, Task, Notification
from .schemas import (
ProjectCreateSchema, ProjectUpdateSchema, ProjectSchema,
TaskCreateSchema, TaskSchema, NotificationSchema
)
api = NinjaAPI(title="Project Management API", auth=django_auth)
# 프로젝트 관리 API
@api.post("/projects", response=ProjectSchema)
def create_project(request, payload: ProjectCreateSchema):
"""프로젝트를 생성합니다."""
project = Project.objects.create(
owner=request.user,
**payload.dict()
)
return project
@api.get("/projects", response=List[ProjectSchema])
def list_projects(
request,
status: Optional[str] = Query(None),
priority: Optional[str] = Query(None),
approaching_deadline: Optional[bool] = Query(None),
overdue: Optional[bool] = Query(None)
):
"""프로젝트 목록을 조회합니다."""
projects = Project.objects.filter(owner=request.user)
if status:
projects = projects.filter(status=status)
if priority:
projects = projects.filter(priority=priority)
if approaching_deadline:
deadline_filter = timezone.now() + timezone.timedelta(days=3)
projects = projects.filter(
deadline__lte=deadline_filter,
deadline__gt=timezone.now()
)
if overdue:
projects = projects.filter(deadline__lt=timezone.now())
return projects
@api.get("/projects/{project_id}", response=ProjectSchema)
def get_project(request, project_id: int):
"""특정 프로젝트를 조회합니다."""
project = get_object_or_404(Project, id=project_id, owner=request.user)
return project
@api.put("/projects/{project_id}", response=ProjectSchema)
def update_project(request, project_id: int, payload: ProjectUpdateSchema):
"""프로젝트를 수정합니다."""
project = get_object_or_404(Project, id=project_id, owner=request.user)
for attr, value in payload.dict(exclude_unset=True).items():
setattr(project, attr, value)
project.save()
# 상태 변경시 알림 생성
if 'status' in payload.dict(exclude_unset=True):
Notification.objects.create(
user=request.user,
project=project,
notification_type='project_updated',
title=f"프로젝트 '{project.name}' 상태 변경",
message=f"상태가 '{project.get_status_display()}'로 변경되었습니다."
)
return project
@api.delete("/projects/{project_id}")
def delete_project(request, project_id: int):
"""프로젝트를 삭제합니다."""
project = get_object_or_404(Project, id=project_id, owner=request.user)
project.delete()
return {"message": "프로젝트가 삭제되었습니다."}
# 작업 관리 API
@api.post("/projects/{project_id}/tasks", response=TaskSchema)
def create_task(request, project_id: int, payload: TaskCreateSchema):
"""프로젝트에 작업을 추가합니다."""
project = get_object_or_404(Project, id=project_id, owner=request.user)
assignee = get_object_or_404(User, id=payload.assignee_id)
task = Task.objects.create(
project=project,
assignee=assignee,
**payload.dict(exclude={'assignee_id'})
)
return task
@api.get("/projects/{project_id}/tasks", response=List[TaskSchema])
def list_tasks(request, project_id: int):
"""프로젝트의 작업 목록을 조회합니다."""
project = get_object_or_404(Project, id=project_id, owner=request.user)
return project.tasks.all()
@api.put("/tasks/{task_id}/complete")
def complete_task(request, task_id: int):
"""작업을 완료 처리합니다."""
task = get_object_or_404(Task, id=task_id)
# 권한 확인
if task.assignee != request.user and task.project.owner != request.user:
return {"error": "권한이 없습니다."}, 403
task.completed = True
task.save()
# 완료 알림 생성
Notification.objects.create(
user=task.project.owner,
project=task.project,
notification_type='task_completed',
title=f"작업 '{task.title}' 완료",
message=f"{task.assignee.username}님이 작업을 완료했습니다."
)
return {"message": "작업이 완료되었습니다."}
# 알림 관리 API
@api.get("/notifications", response=List[NotificationSchema])
def list_notifications(request, unread_only: bool = Query(False)):
"""알림 목록을 조회합니다."""
notifications = Notification.objects.filter(user=request.user)
if unread_only:
notifications = notifications.filter(is_read=False)
return notifications.order_by('-created_at')
@api.put("/notifications/{notification_id}/read")
def mark_notification_read(request, notification_id: int):
"""알림을 읽음 처리합니다."""
notification = get_object_or_404(
Notification,
id=notification_id,
user=request.user
)
notification.is_read = True
notification.save()
return {"message": "알림을 읽음 처리했습니다."}
# 대시보드 API
@api.get("/dashboard")
def get_dashboard(request):
"""대시보드 데이터를 조회합니다."""
user_projects = Project.objects.filter(owner=request.user)
now = timezone.now()
return {
"total_projects": user_projects.count(),
"active_projects": user_projects.filter(
status__in=['planning', 'development', 'testing']
).count(),
"completed_projects": user_projects.filter(status='completed').count(),
"overdue_projects": user_projects.filter(
deadline__lt=now,
status__in=['planning', 'development', 'testing']
).count(),
"approaching_deadline": user_projects.filter(
deadline__lte=now + timezone.timedelta(days=3),
deadline__gt=now,
status__in=['planning', 'development', 'testing']
).count(),
"unread_notifications": Notification.objects.filter(
user=request.user,
is_read=False
).count(),
"recent_projects": user_projects.order_by('-created_at')[:5],
}
# URL 설정
# urls.py
from django.urls import path
from .api import api
urlpatterns = [
path("api/", api.urls),
]
알림 시스템 구축
다양한 채널을 통한 알림 시스템을 구축해보겠습니다.
# notifications.py
from django.core.mail import send_mail
from django.conf import settings
from django.template.loader import render_to_string
import requests
import json
import logging
logger = logging.getLogger(__name__)
def send_email(to, subject, message=None, template=None, context=None, project=None, urgent=False):
"""이메일 알림을 발송합니다."""
try:
if template:
html_message = render_to_string(f'emails/{template}.html', context)
message = render_to_string(f'emails/{template}.txt', context)
# 긴급한 경우 제목에 표시
if urgent:
subject = f"🚨 [긴급] {subject}"
send_mail(
subject=subject,
message=message,
from_email=settings.DEFAULT_FROM_EMAIL,
recipient_list=[to],
html_message=html_message if template else None,
fail_silently=False
)
logger.info(f"Email sent to {to}: {subject}")
except Exception as e:
logger.error(f"Failed to send email to {to}: {str(e)}")
def send_slack_notification(channel, message, urgent=False):
"""Slack 알림을 발송합니다."""
if not settings.SLACK_WEBHOOK_URL:
return
try:
payload = {
"channel": channel,
"text": message,
"icon_emoji": ":warning:" if urgent else ":information_source:",
"username": "Project Manager Bot"
}
response = requests.post(
settings.SLACK_WEBHOOK_URL,
data=json.dumps(payload),
headers={'Content-Type': 'application/json'}
)
response.raise_for_status()
logger.info(f"Slack notification sent to {channel}")
except Exception as e:
logger.error(f"Failed to send Slack notification: {str(e)}")
def send_discord_notification(webhook_url, message, urgent=False):
"""Discord 알림을 발송합니다."""
try:
payload = {
"content": message,
"embeds": [{
"color": 16711680 if urgent else 3447003, # 빨간색 or 파란색
"title": "프로젝트 알림" if not urgent else "🚨 긴급 알림",
"description": message,
"timestamp": timezone.now().isoformat()
}]
}
response = requests.post(webhook_url, json=payload)
response.raise_for_status()
logger.info("Discord notification sent")
except Exception as e:
logger.error(f"Failed to send Discord notification: {str(e)}")
# 통합 알림 함수
def send_notification(user, message, notification_type='info', urgent=False):
"""사용자 설정에 따라 다양한 채널로 알림을 발송합니다."""
# 이메일 알림 (기본)
if user.email:
send_email(
to=user.email,
subject="프로젝트 알림",
message=message,
urgent=urgent
)
# 사용자 프로필에 따른 추가 알림
try:
profile = user.userprofile
if profile.slack_channel:
send_slack_notification(profile.slack_channel, message, urgent)
if profile.discord_webhook:
send_discord_notification(profile.discord_webhook, message, urgent)
except:
pass # UserProfile이 없는 경우 무시
실전 예제: 프론트엔드 연동
React를 사용한 프론트엔드 예제입니다.
// ProjectDashboard.jsx
import React, { useState, useEffect } from 'react';
import axios from 'axios';
const ProjectDashboard = () => {
const [dashboardData, setDashboardData] = useState(null);
const [projects, setProjects] = useState([]);
const [notifications, setNotifications] = useState([]);
const [loading, setLoading] = useState(true);
useEffect(() => {
fetchDashboardData();
fetchProjects();
fetchNotifications();
// 5분마다 데이터 새로고침
const interval = setInterval(() => {
fetchNotifications();
fetchDashboardData();
}, 5 * 60 * 1000);
return () => clearInterval(interval);
}, []);
const fetchDashboardData = async () => {
try {
const response = await axios.get('/api/dashboard');
setDashboardData(response.data);
} catch (error) {
console.error('Failed to fetch dashboard data:', error);
}
};
const fetchProjects = async () => {
try {
const response = await axios.get('/api/projects');
setProjects(response.data);
} catch (error) {
console.error('Failed to fetch projects:', error);
} finally {
setLoading(false);
}
};
const fetchNotifications = async () => {
try {
const response = await axios.get('/api/notifications?unread_only=true');
setNotifications(response.data);
} catch (error) {
console.error('Failed to fetch notifications:', error);
}
};
const markNotificationRead = async (notificationId) => {
try {
await axios.put(`/api/notifications/${notificationId}/read`);
fetchNotifications(); // 알림 목록 새로고침
} catch (error) {
console.error('Failed to mark notification as read:', error);
}
};
const getProjectStatusColor = (project) => {
if (project.is_overdue) return 'bg-red-500';
if (project.is_approaching_deadline) return 'bg-yellow-500';
return 'bg-green-500';
};
const getPriorityIcon = (priority) => {
const icons = {
urgent: '🚨',
high: '⚡',
medium: '📋',
low: '📝'
};
return icons[priority] || '📋';
};
if (loading) {
return <div className="flex justify-center items-center h-screen">로딩중...</div>;
}
return (
<div className="container mx-auto p-6">
{/* 대시보드 요약 */}
{dashboardData && (
<div className="grid grid-cols-1 md:grid-cols-2 lg:grid-cols-4 gap-6 mb-8">
<div className="bg-white rounded-lg shadow p-6">
<h3 className="text-lg font-semibold text-gray-700">전체 프로젝트</h3>
<p className="text-3xl font-bold text-blue-600">{dashboardData.total_projects}</p>
</div>
<div className="bg-white rounded-lg shadow p-6">
<h3 className="text-lg font-semibold text-gray-700">진행중</h3>
<p className="text-3xl font-bold text-green-600">{dashboardData.active_projects}</p>
</div>
<div className="bg-white rounded-lg shadow p-6">
<h3 className="text-lg font-semibold text-gray-700">마감임박</h3>
<p className="text-3xl font-bold text-yellow-600">{dashboardData.approaching_deadline}</p>
</div>
<div className="bg-white rounded-lg shadow p-6">
<h3 className="text-lg font-semibold text-gray-700">지연</h3>
<p className="text-3xl font-bold text-red-600">{dashboardData.overdue_projects}</p>
</div>
</div>
)}
{/* 알림 섹션 */}
{notifications.length > 0 && (
<div className="bg-white rounded-lg shadow mb-8 p-6">
<h2 className="text-xl font-semibold mb-4">🔔 새로운 알림</h2>
<div className="space-y-3">
{notifications.map(notification => (
<div
key={notification.id}
className="flex items-start justify-between p-3 bg-gray-50 rounded-lg hover:bg-gray-100 cursor-pointer"
onClick={() => markNotificationRead(notification.id)}
>
<div className="flex-1">
<h3 className="font-medium text-gray-900">{notification.title}</h3>
<p className="text-sm text-gray-600 mt-1">{notification.message}</p>
<p className="text-xs text-gray-400 mt-1">
{new Date(notification.created_at).toLocaleString()}
</p>
</div>
<button
className="text-sm text-blue-600 hover:text-blue-800"
onClick={(e) => {
e.stopPropagation();
markNotificationRead(notification.id);
}}
>
읽음
</button>
</div>
))}
</div>
</div>
)}
{/* 프로젝트 목록 */}
<div className="bg-white rounded-lg shadow">
<div className="p-6 border-b border-gray-200">
<h2 className="text-xl font-semibold">프로젝트 목록</h2>
</div>
<div className="divide-y divide-gray-200">
{projects.map(project => (
<div key={project.id} className="p-6 hover:bg-gray-50">
<div className="flex items-center justify-between">
<div className="flex items-center space-x-3">
<div className={`w-3 h-3 rounded-full ${getProjectStatusColor(project)}`}></div>
<span className="text-lg">{getPriorityIcon(project.priority)}</span>
<div>
<h3 className="text-lg font-medium text-gray-900">{project.name}</h3>
<p className="text-sm text-gray-500">{project.description}</p>
</div>
</div>
<div className="text-right">
<p className="text-sm font-medium text-gray-900">
마감: {new Date(project.deadline).toLocaleDateString()}
</p>
<p className={`text-sm ${project.is_overdue ? 'text-red-600' : project.is_approaching_deadline ? 'text-yellow-600' : 'text-gray-500'}`}>
{project.is_overdue ?
`${Math.abs(project.days_until_deadline)}일 지연` :
`${project.days_until_deadline}일 남음`
}
</p>
<span className={`inline-block px-2 py-1 text-xs rounded-full ${
project.status === 'completed' ? 'bg-green-100 text-green-800' :
project.status === 'development' ? 'bg-blue-100 text-blue-800' :
'bg-gray-100 text-gray-800'
}`}>
{project.status}
</span>
</div>
</div>
</div>
))}
</div>
</div>
</div>
);
};
export default ProjectDashboard;
성능 최적화 및 모니터링
# monitoring.py
import logging
from django.core.management.base import BaseCommand
from django.db.models import Count, Q
from django.utils import timezone
from datetime import timedelta
import redis
logger = logging.getLogger(__name__)
class Command(BaseCommand):
help = 'Monitor project scheduling system performance'
def handle(self, *args, **options):
"""시스템 상태를 모니터링합니다."""
# Redis 연결 상태 확인
try:
r = redis.Redis(host='localhost', port=6379, db=0)
r.ping()
self.stdout.write("✅ Redis connection: OK")
except:
self.stdout.write("❌ Redis connection: Failed")
# Celery 작업 큐 상태 확인
from celery import current_app
inspect = current_app.control.inspect()
try:
active_tasks = inspect.active()
scheduled_tasks = inspect.scheduled()
total_active = sum(len(tasks) for tasks in active_tasks.values()) if active_tasks else 0
total_scheduled = sum(len(tasks) for tasks in scheduled_tasks.values()) if scheduled_tasks else 0
self.stdout.write(f"📋 Active Celery tasks: {total_active}")
self.stdout.write(f"⏰ Scheduled Celery tasks: {total_scheduled}")
except:
self.stdout.write("❌ Celery inspection failed")
# 프로젝트 통계
from project_manager.models import Project, Notification
now = timezone.now()
week_ago = now - timedelta(days=7)
stats = {
'total_projects': Project.objects.count(),
'active_projects': Project.objects.filter(
status__in=['planning', 'development', 'testing']
).count(),
'overdue_projects': Project.objects.filter(
deadline__lt=now,
status__in=['planning', 'development', 'testing']
).count(),
'notifications_sent_week': Notification.objects.filter(
created_at__gte=week_ago
).count(),
}
self.stdout.write("\n📊 프로젝트 통계:")
for key, value in stats.items():
self.stdout.write(f" {key}: {value}")
# 성능 최적화 제안
self.check_performance_issues()
def check_performance_issues(self):
"""성능 문제를 확인하고 제안합니다."""
from project_manager.models import Project, Notification
# 오래된 완료 프로젝트 확인
old_completed = Project.objects.filter(
status='completed',
updated_at__lt=timezone.now() - timedelta(days=90)
).count()
if old_completed > 100:
self.stdout.write(f"⚠️ {old_completed}개의 오래된 완료 프로젝트가 있습니다. 아카이브를 고려하세요.")
# 읽지 않은 알림 확인
old_notifications = Notification.objects.filter(
is_read=False,
created_at__lt=timezone.now() - timedelta(days=30)
).count()
if old_notifications > 1000:
self.stdout.write(f"⚠️ {old_notifications}개의 오래된 알림이 있습니다. 정리가 필요합니다.")
# 성능 최적화를 위한 데이터베이스 인덱스
# models.py에 추가
class Project(models.Model):
# ... 기존 필드들 ...
class Meta:
ordering = ['-priority', 'deadline']
indexes = [
models.Index(fields=['owner', 'status']),
models.Index(fields=['deadline', 'status']),
models.Index(fields=['priority', 'deadline']),
models.Index(fields=['created_at']),
]
class Notification(models.Model):
# ... 기존 필드들 ...
class Meta:
indexes = [
models.Index(fields=['user', 'is_read']),
models.Index(fields=['created_at']),
models.Index(fields=['notification_type', 'created_at']),
]
배포 및 운영 가이드
# docker-compose.yml
version: '3.8'
services:
web:
build: .
ports:
- "8000:8000"
environment:
- DEBUG=False
- DATABASE_URL=postgresql://user:password@db:5432/projectdb
- REDIS_URL=redis://redis:6379
depends_on:
- db
- redis
volumes:
- ./logs:/app/logs
db:
image: postgres:13
environment:
POSTGRES_DB: projectdb
POSTGRES_USER: user
POSTGRES_PASSWORD: password
volumes:
- postgres_data:/var/lib/postgresql/data
redis:
image: redis:6-alpine
ports:
- "6379:6379"
celery:
build: .
command: celery -A myproject worker -l info
environment:
- DATABASE_URL=postgresql://user:password@db:5432/projectdb
- REDIS_URL=redis://redis:6379
depends_on:
- db
- redis
celery-beat:
build: .
command: celery -A myproject beat -l info --scheduler django_celery_beat.schedulers:DatabaseScheduler
environment:
- DATABASE_URL=postgresql://user:password@db:5432/projectdb
- REDIS_URL=redis://redis:6379
depends_on:
- db
- redis
nginx:
image: nginx:alpine
ports:
- "80:80"
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf
- ./static:/static
depends_on:
- web
volumes:
postgres_data:
마무리
이번 포스트에서는 Django Ninja와 다양한 스케줄링 도구를 활용해 프로젝트 마감일을 효율적으로 관리하는 시스템을 구축하는 방법을 알아봤습니다.
핵심 포인트
- 적절한 스케줄링 도구 선택: 프로젝트 규모와 요구사항에 맞는 도구 선택이 중요
- 체계적인 모델 설계: 확장 가능한 데이터베이스 구조 설계
- 효과적인 알림 시스템: 다양한 채널을 통한 적시 알림
- 성능 최적화: 인덱스와 모니터링을 통한 성능 관리
개발 난이도 순위 (쉬운 순)
- Django-cron: 가장 간단하지만 기능 제한적
- APScheduler: 중간 난이도, 중소규모에 적합
- Celery Beat: 높은 난이도, 대규모 서비스에 최적
이 가이드를 참고하여 프로젝트 규모와 요구사항에 맞는 마감일 관리 시스템을 구축해보시기 바랍니다. 궁금한 점이 있으시면 언제든 댓글로 질문해주세요!
관련 포스트
참고 자료