236 lines
7.9 KiB
Python
236 lines
7.9 KiB
Python
# Copyright (c) 2023-present Plane Software, Inc. and contributors
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
# See the LICENSE file for details.
|
|
|
|
# Python imports
|
|
import json
|
|
from typing import Optional, List, Dict
|
|
from uuid import UUID
|
|
from itertools import groupby
|
|
import logging
|
|
|
|
# Django imports
|
|
from django.utils import timezone
|
|
from django.db import transaction
|
|
|
|
# Third party imports
|
|
from celery import shared_task
|
|
|
|
# Module imports
|
|
from plane.db.models import (
|
|
Issue,
|
|
IssueVersion,
|
|
ProjectMember,
|
|
CycleIssue,
|
|
ModuleIssue,
|
|
IssueActivity,
|
|
IssueAssignee,
|
|
IssueLabel,
|
|
)
|
|
from plane.utils.exception_logger import log_exception
|
|
|
|
|
|
@shared_task
|
|
def issue_task(updated_issue, issue_id, user_id):
|
|
try:
|
|
current_issue = json.loads(updated_issue) if updated_issue else {}
|
|
issue = Issue.objects.get(id=issue_id)
|
|
|
|
updated_current_issue = {}
|
|
for key, value in current_issue.items():
|
|
if getattr(issue, key) != value:
|
|
updated_current_issue[key] = value
|
|
|
|
if updated_current_issue:
|
|
issue_version = IssueVersion.objects.filter(issue_id=issue_id).order_by("-last_saved_at").first()
|
|
|
|
if (
|
|
issue_version
|
|
and str(issue_version.owned_by) == str(user_id)
|
|
and (timezone.now() - issue_version.last_saved_at).total_seconds() <= 600
|
|
):
|
|
for key, value in updated_current_issue.items():
|
|
setattr(issue_version, key, value)
|
|
issue_version.last_saved_at = timezone.now()
|
|
issue_version.save(update_fields=list(updated_current_issue.keys()) + ["last_saved_at"])
|
|
else:
|
|
IssueVersion.log_issue_version(issue, user_id)
|
|
|
|
return
|
|
except Issue.DoesNotExist:
|
|
return
|
|
except Exception as e:
|
|
log_exception(e)
|
|
return
|
|
|
|
|
|
def get_owner_id(issue: Issue) -> Optional[int]:
|
|
"""Get the owner ID of the issue"""
|
|
|
|
if issue.updated_by_id:
|
|
return issue.updated_by_id
|
|
|
|
if issue.created_by_id:
|
|
return issue.created_by_id
|
|
|
|
# Find project admin as fallback
|
|
project_member = ProjectMember.objects.filter(
|
|
project_id=issue.project_id,
|
|
role=20, # Admin role
|
|
).first()
|
|
|
|
return project_member.member_id if project_member else None
|
|
|
|
|
|
def get_related_data(issue_ids: List[UUID]) -> Dict:
|
|
"""Get related data for the given issue IDs"""
|
|
|
|
cycle_issues = {ci.issue_id: ci.cycle_id for ci in CycleIssue.objects.filter(issue_id__in=issue_ids)}
|
|
|
|
# Get assignees with proper grouping
|
|
assignee_records = list(
|
|
IssueAssignee.objects.filter(issue_id__in=issue_ids).values_list("issue_id", "assignee_id").order_by("issue_id")
|
|
)
|
|
assignees = {}
|
|
for issue_id, group in groupby(assignee_records, key=lambda x: x[0]):
|
|
assignees[issue_id] = [str(g[1]) for g in group]
|
|
|
|
# Get labels with proper grouping
|
|
label_records = list(
|
|
IssueLabel.objects.filter(issue_id__in=issue_ids).values_list("issue_id", "label_id").order_by("issue_id")
|
|
)
|
|
labels = {}
|
|
for issue_id, group in groupby(label_records, key=lambda x: x[0]):
|
|
labels[issue_id] = [str(g[1]) for g in group]
|
|
|
|
# Get modules with proper grouping
|
|
module_records = list(
|
|
ModuleIssue.objects.filter(issue_id__in=issue_ids).values_list("issue_id", "module_id").order_by("issue_id")
|
|
)
|
|
modules = {}
|
|
for issue_id, group in groupby(module_records, key=lambda x: x[0]):
|
|
modules[issue_id] = [str(g[1]) for g in group]
|
|
|
|
# Get latest activities
|
|
latest_activities = {}
|
|
activities = IssueActivity.objects.filter(issue_id__in=issue_ids).order_by("issue_id", "-created_at")
|
|
for issue_id, activities_group in groupby(activities, key=lambda x: x.issue_id):
|
|
first_activity = next(activities_group, None)
|
|
if first_activity:
|
|
latest_activities[issue_id] = first_activity.id
|
|
|
|
return {
|
|
"cycle_issues": cycle_issues,
|
|
"assignees": assignees,
|
|
"labels": labels,
|
|
"modules": modules,
|
|
"activities": latest_activities,
|
|
}
|
|
|
|
|
|
def create_issue_version(issue: Issue, related_data: Dict) -> Optional[IssueVersion]:
|
|
"""Create IssueVersion object from the given issue and related data"""
|
|
|
|
try:
|
|
if not issue.workspace_id or not issue.project_id:
|
|
logging.warning(f"Skipping issue {issue.id} - missing workspace_id or project_id")
|
|
return None
|
|
|
|
owned_by_id = get_owner_id(issue)
|
|
if owned_by_id is None:
|
|
logging.warning(f"Skipping issue {issue.id} - missing owned_by")
|
|
return None
|
|
|
|
return IssueVersion(
|
|
workspace_id=issue.workspace_id,
|
|
project_id=issue.project_id,
|
|
created_by_id=issue.created_by_id,
|
|
updated_by_id=issue.updated_by_id,
|
|
owned_by_id=owned_by_id,
|
|
last_saved_at=timezone.now(),
|
|
activity_id=related_data["activities"].get(issue.id),
|
|
properties=getattr(issue, "properties", {}),
|
|
meta=getattr(issue, "meta", {}),
|
|
issue_id=issue.id,
|
|
parent=issue.parent_id,
|
|
state=issue.state_id,
|
|
estimate_point=issue.estimate_point_id,
|
|
name=issue.name,
|
|
priority=issue.priority,
|
|
start_date=issue.start_date,
|
|
target_date=issue.target_date,
|
|
assignees=related_data["assignees"].get(issue.id, []),
|
|
sequence_id=issue.sequence_id,
|
|
labels=related_data["labels"].get(issue.id, []),
|
|
sort_order=issue.sort_order,
|
|
completed_at=issue.completed_at,
|
|
archived_at=issue.archived_at,
|
|
is_draft=issue.is_draft,
|
|
external_source=issue.external_source,
|
|
external_id=issue.external_id,
|
|
type=issue.type_id,
|
|
cycle=related_data["cycle_issues"].get(issue.id),
|
|
modules=related_data["modules"].get(issue.id, []),
|
|
)
|
|
except Exception as e:
|
|
log_exception(e)
|
|
return None
|
|
|
|
|
|
@shared_task
|
|
def sync_issue_version(batch_size=5000, offset=0, countdown=300):
|
|
"""Task to create IssueVersion records for existing Issues in batches"""
|
|
|
|
try:
|
|
with transaction.atomic():
|
|
base_query = Issue.objects
|
|
total_issues_count = base_query.count()
|
|
|
|
if total_issues_count == 0:
|
|
return
|
|
|
|
end_offset = min(offset + batch_size, total_issues_count)
|
|
|
|
# Get issues batch with optimized queries
|
|
issues_batch = list(
|
|
base_query.order_by("created_at").select_related("workspace", "project").all()[offset:end_offset]
|
|
)
|
|
|
|
if not issues_batch:
|
|
return
|
|
|
|
# Get all related data in bulk
|
|
issue_ids = [issue.id for issue in issues_batch]
|
|
related_data = get_related_data(issue_ids)
|
|
|
|
issue_versions = []
|
|
for issue in issues_batch:
|
|
version = create_issue_version(issue, related_data)
|
|
if version:
|
|
issue_versions.append(version)
|
|
|
|
# Bulk create versions
|
|
if issue_versions:
|
|
IssueVersion.objects.bulk_create(issue_versions, batch_size=1000)
|
|
|
|
# Schedule the next batch if there are more workspaces to process
|
|
if end_offset < total_issues_count:
|
|
sync_issue_version.apply_async(
|
|
kwargs={
|
|
"batch_size": batch_size,
|
|
"offset": end_offset,
|
|
"countdown": countdown,
|
|
},
|
|
countdown=countdown,
|
|
)
|
|
|
|
logging.info(f"Processed Issues: {end_offset}")
|
|
return
|
|
except Exception as e:
|
|
log_exception(e)
|
|
return
|
|
|
|
|
|
@shared_task
|
|
def schedule_issue_version(batch_size=5000, countdown=300):
|
|
sync_issue_version.delay(batch_size=int(batch_size), countdown=countdown)
|