Mirrors/plane
1
0
Fork 0
mirror of https://github.com/makeplane/plane.git synced 2026-07-12 21:40:18 +02:00
Code Issues Packages Projects Releases Wiki Activity

chore: updated the logic and added sync tasks in celery imports

Browse Source
This commit is contained in:
gurusainath
2024-12-10 17:59:44 +05:30
parent daf0d9d3a5
commit b4d008d4bb
7 changed files with 519 additions and 395 deletions
Download Patch File Download Diff File Expand all files Collapse all files

115
apiserver/plane/bgtasks/issue_description_version_sync.py Normal file
View File

@@ -0,0 +1,115 @@
# Python imports
from typing import Optional
# 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, IssueDescriptionVersion, ProjectMember
from plane.utils.exception_logger import log_exception
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
@shared_task
def sync_issue_description_version(batch_size=5000, offset=0, countdown=300):
"""Task to create IssueDescriptionVersion 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
# Calculate batch range
end_offset = min(offset + batch_size, total_issues_count)
# Fetch issues with related data
issues_batch = (
base_query.order_by("created_at")
.select_related("workspace", "project")
.only(
"id",
"workspace_id",
"project_id",
"created_by_id",
"updated_by_id",
"description_binary",
"description_html",
"description_stripped",
"description",
)[offset:end_offset]
)
if not issues_batch:
return
version_objects = []
for issue in issues_batch:
# Validate required fields
if not issue.workspace_id or not issue.project_id:
print(f"Skipping {issue.id} - missing workspace_id or project_id")
continue
# Determine owned_by_id
owned_by_id = get_owner_id(issue)
if owned_by_id is None:
print(f"Skipping issue {issue.id} - missing owned_by")
continue
# Create version object
version_objects.append(
IssueDescriptionVersion(
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(),
issue_id=issue.id,
description_binary=issue.description_binary,
description_html=issue.description_html,
description_stripped=issue.description_stripped,
description_json=issue.description,
)
)
# Bulk create version objects
if version_objects:
IssueDescriptionVersion.objects.bulk_create(version_objects)
# Schedule next batch if needed
if end_offset < total_issues_count:
sync_issue_description_version.apply_async(
args=[batch_size, end_offset], countdown=countdown
)
return
except Exception as e:
log_exception(e)
return
@shared_task
def schedule_issue_description_version(batch_size=5000, countdown=300):
sync_issue_description_version.delay(batch_size=batch_size, countdown=countdown)

85
apiserver/plane/bgtasks/issue_description_version_task.py Normal file
View File

@@ -0,0 +1,85 @@
from celery import shared_task
from django.db import transaction
from django.utils import timezone
from typing import Optional, Dict, Any
import json
from plane.db.models import Issue, IssueDescriptionVersion
from plane.utils.logging import log_exception
def should_update_existing_version(
version: IssueDescriptionVersion, user_id: str, max_time_difference: int = 600
) -> bool:
if not version:
return
time_difference = (timezone.now() - version.last_saved_at).total_seconds()
return (
str(version.owned_by_id) == str(user_id)
and time_difference <= max_time_difference
)
def update_existing_version(
version: IssueDescriptionVersion, description_data: Dict[str, Any]
) -> None:
version.description_json = description_data.get("description")
version.description_html = description_data.get("description_html")
version.description_binary = description_data.get("description_binary")
version.description_stripped = description_data.get("description_stripped")
version.last_saved_at = timezone.now()
version.save(
update_fields=[
"description_json",
"description_html",
"description_binary",
"description_stripped",
"last_saved_at",
]
)
@shared_task
def issue_description_version_task(
updated_issue: Optional[str], issue_id: str, user_id: str
) -> Optional[bool]:
try:
# Parse updated issue data
current_issue: Dict = json.loads(updated_issue) if updated_issue else {}
# Get current issue
issue = Issue.objects.get(id=issue_id)
# Check if description has changed
if current_issue.get("description_html") == issue.description_html:
return
with transaction.atomic():
# Get latest version
latest_version = (
IssueDescriptionVersion.objects.filter(issue_id=issue_id)
.order_by("-last_saved_at")
.first()
)
# Determine whether to update existing or create new version
if should_update_existing_version(latest_version, user_id):
update_existing_version(latest_version, current_issue)
else:
IssueDescriptionVersion.log_issue_description_version(
current_issue, user_id
)
return
except Issue.DoesNotExist:
# Issue no longer exists, skip processing
return
except json.JSONDecodeError as e:
log_exception(f"Invalid JSON for updated_issue: {e}")
return
except Exception as e:
log_exception(f"Error processing issue description version: {e}")
return

250
apiserver/plane/bgtasks/issue_version_sync.py Normal file
View File

@@ -0,0 +1,250 @@
# Python imports
import json
from typing import Optional, List, Dict
from uuid import UUID
from itertools import groupby
# 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:
print(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:
print(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,
point=issue.point,
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
print(f"Offset: {offset}")
print(f"Total Issues: {total_issues_count}")
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(
args=[batch_size, end_offset], countdown=countdown
)
print(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=batch_size, countdown=countdown)

451
apiserver/plane/bgtasks/issue_version_task.py
View File

@@ -1,6 +1,6 @@
# Python imports
import json
from typing import Optional, List, Dict
from typing import Dict, Any, Optional, List
# Django imports
from django.utils import timezone
@@ -10,410 +10,81 @@ from django.db import transaction
from celery import shared_task
# Module imports
from plane.db.models import (
Issue,
IssueVersion,
IssueDescriptionVersion,
ProjectMember,
CycleIssue,
ModuleIssue,
IssueActivity,
IssueAssignee,
IssueLabel,
)
from plane.db.models import Issue, IssueVersion
from plane.utils.exception_logger import log_exception
def get_changed_fields(current_issue: Dict[str, Any], issue: Issue) -> Dict[str, Any]:
return {
key: value
for key, value in current_issue.items()
if getattr(issue, key) != value
}
def should_update_existing_version(
version: Optional[IssueVersion], user_id: str, max_time_difference: int = 600
) -> bool:
if not version:
return False
time_difference = (timezone.now() - version.last_saved_at).total_seconds()
return (
str(version.owned_by_id) == str(user_id)
and time_difference <= max_time_difference
)
def update_version_fields(
version: IssueVersion, changed_fields: Dict[str, Any]
) -> List[str]:
for key, value in changed_fields.items():
setattr(version, key, value)
version.last_saved_at = timezone.now()
update_fields = list(changed_fields.keys()) + ["last_saved_at"]
return update_fields
@shared_task
def issue_task(updated_issue, issue_id, user_id):
def issue_version_task(
updated_issue: Optional[str], issue_id: str, user_id: str
) -> Optional[bool]:
try:
print("====== Issue detail update ======")
current_issue = json.loads(updated_issue) if updated_issue else {}
issue = Issue.objects.get(id=issue_id)
# Parse updated issue data
current_issue: Dict = json.loads(updated_issue) if updated_issue else {}
updated_current_issue = {}
for key, value in current_issue.items():
if getattr(issue, key) != value:
updated_current_issue[key] = value
with transaction.atomic():
# Get current issue
issue = Issue.objects.get(id=issue_id)
if updated_current_issue:
issue_version = (
# Get changed fields
changed_fields = get_changed_fields(current_issue, issue)
if not changed_fields:
return True
# Get latest version
latest_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"]
)
# Update existing or create new version
if should_update_existing_version(latest_version, user_id):
update_fields = update_version_fields(latest_version, changed_fields)
latest_version.save(update_fields=update_fields)
else:
IssueVersion.log_issue_version(issue, user_id)
return
return True
except Issue.DoesNotExist:
return
except Exception as e:
log_exception(e)
return
@shared_task
def issue_description_task(updated_issue, issue_id, user_id):
try:
print("====== Issue description update ======")
current_issue = json.loads(updated_issue) if updated_issue else {}
issue = Issue.objects.get(id=issue_id)
current_issue_description_html = current_issue.get("description_html")
issue_description_html = issue.description_html
if current_issue_description_html != issue_description_html:
issue_description_version = (
IssueVersion.objects.filter(issue_id=issue_id)
.order_by("-last_saved_at")
.first()
)
current_issue_description_json = current_issue.get("description")
current_issue_description_html = current_issue.get("description_html")
current_issue_description_binary = current_issue.get("description_binary")
current_issue_description_stripped = current_issue.get(
"description_stripped"
)
if (
issue_description_version
and str(issue_description_version.owned_by) == str(user_id)
and (
timezone.now() - issue_description_version.last_saved_at
).total_seconds()
<= 600
):
issue_description_version.description_json = (
current_issue_description_json
)
issue_description_version.description_html = (
current_issue_description_html
)
issue_description_version.description_binary = (
current_issue_description_binary
)
issue_description_version.description_stripped = (
current_issue_description_stripped
)
issue_description_version.last_saved_at = timezone.now()
issue_description_version.save(
update_fields=[
"description_json",
"description_html",
"description_binary",
"description_stripped",
"last_saved_at",
]
)
else:
IssueDescriptionVersion.log_issue_description_version(
current_issue, user_id
)
return
except Issue.DoesNotExist:
return
except Exception as e:
log_exception(e)
return
@shared_task
def issue_versioning_task(updated_issue, issue_id, user_id):
try:
issue_task.delay(updated_issue, issue_id, user_id)
issue_description_task.delay(updated_issue, issue_id, user_id)
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
# ============ Issue versioning management task starts ============
def get_related_data(issue_ids: List[int]) -> 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)
}
assignees = {
issue_id: list(assignee_ids)
for issue_id, assignee_ids in (
IssueAssignee.objects.filter(issue_id__in=issue_ids)
.values_list("issue_id", "assignee_id")
.order_by("issue_id")
).groupby("issue_id")
}
labels = {
issue_id: list(label_ids)
for issue_id, label_ids in (
IssueLabel.objects.filter(issue_id__in=issue_ids)
.values_list("issue_id", "label_id")
.order_by("issue_id")
).groupby("issue_id")
}
modules = {
issue_id: list(module_ids)
for issue_id, module_ids in (
ModuleIssue.objects.filter(issue_id__in=issue_ids)
.values_list("issue_id", "module_id")
.order_by("issue_id")
).groupby("issue_id")
}
latest_activities = {
ia.issue_id: ia.id
for ia in IssueActivity.objects.filter(issue_id__in=issue_ids)
.order_by("issue_id", "-created_at")
.distinct("issue_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"""
if not issue.workspace_id or not issue.project_id:
print(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:
print(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,
point=issue.point,
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, []),
)
@shared_task
def issue_management_command_task(batch_size=5000, offset=0, countdown=300):
"""Task to create IssueVersion records for existing Issues in batches"""
try:
with transaction.atomic():
# Get processed issues and total count
processed_issue_ids = set(
IssueVersion.objects.values_list("issue_id", flat=True).distinct()
)
base_query = Issue.objects.exclude(id__in=processed_issue_ids)
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:
issue_management_command_task.apply_async(
args=[batch_size, end_offset], countdown=countdown
)
return
except json.JSONDecodeError as e:
log_exception(f"Invalid JSON for updated_issue: {e}")
return False
except Exception as e:
log_exception(e)
return
@shared_task
def schedule_issue_management_command_task(batch_size=5000, countdown=300):
issue_management_command_task.delay(batch_size=batch_size, countdown=countdown)
# ============ Issue versioning management task ends ============
# ============ Issue description versioning management task starts ============
@shared_task
def issue_description_management_command_task(batch_size=5000, offset=0, countdown=300):
"""Task to create IssueDescriptionVersion records for existing Issues in batches"""
try:
with transaction.atomic():
# Get processed issue IDs and total count
processed_ids = set(
IssueDescriptionVersion.objects.values_list(
"issue_id", flat=True
).distinct()
)
base_query = Issue.objects.exclude(id__in=processed_ids)
total_issues_count = base_query.count()
if total_issues_count == 0:
return
# Calculate batch range
end_offset = min(offset + batch_size, total_issues_count)
# Fetch issues with related data
issues_batch = (
base_query.order_by("created_at")
.select_related("workspace", "project")
.only(
"id",
"workspace_id",
"project_id",
"created_by_id",
"updated_by_id",
"description_binary",
"description_html",
"description_stripped",
"description",
)[offset:end_offset]
)
if not issues_batch:
return
version_objects = []
for issue in issues_batch:
# Validate required fields
if not issue.workspace_id or not issue.project_id:
print(f"Skipping {issue.id} - missing workspace_id or project_id")
continue
# Determine owned_by_id
owned_by_id = get_owner_id(issue)
if owned_by_id is None:
print(f"Skipping issue {issue.id} - missing owned_by")
continue
# Create version object
version_objects.append(
IssueDescriptionVersion(
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(),
issue_id=issue.id,
description_binary=issue.description_binary,
description_html=issue.description_html,
description_stripped=issue.description_stripped,
description_json=issue.description,
)
)
# Bulk create version objects
if version_objects:
IssueDescriptionVersion.objects.bulk_create(version_objects)
# Schedule next batch if needed
if end_offset < total_issues_count:
issue_description_management_command_task.apply_async(
args=[batch_size, end_offset], countdown=countdown
)
return
except Exception as e:
log_exception(e)
return
@shared_task
def schedule_issue_description_management_command_task(batch_size=5000, countdown=300):
issue_description_management_command_task.delay(
batch_size=batch_size, countdown=countdown
)
# ============ Issue description versioning management task nds ============
log_exception(f"Error processing issue version: {e}")
return False

6
apiserver/plane/db/management/commands/create_issue_description_version.py → apiserver/plane/db/management/commands/sync_issue_description_version.py
View File

@@ -2,8 +2,8 @@
from django.core.management.base import BaseCommand
# Module imports
from plane.bgtasks.issue_version_task import (
schedule_issue_description_management_command_task,
from plane.bgtasks.issue_description_version_sync import (
schedule_issue_description_version,
)
@@ -14,7 +14,7 @@ class Command(BaseCommand):
batch_size = input("Enter the batch size: ")
batch_countdown = input("Enter the batch countdown: ")
schedule_issue_description_management_command_task.delay(
schedule_issue_description_version.delay(
batch_size=int(batch_size), countdown=int(batch_countdown)
)

4
apiserver/plane/db/management/commands/create_issue_version.py → apiserver/plane/db/management/commands/sync_issue_version.py
View File

@@ -2,7 +2,7 @@
from django.core.management.base import BaseCommand
# Module imports
from plane.bgtasks.issue_version_task import schedule_issue_management_command_task
from plane.bgtasks.issue_version_sync import schedule_issue_version
class Command(BaseCommand):
@@ -12,7 +12,7 @@ class Command(BaseCommand):
batch_size = input("Enter the batch size: ")
batch_countdown = input("Enter the batch countdown: ")
schedule_issue_management_command_task.delay(
schedule_issue_version.delay(
batch_size=int(batch_size), countdown=int(batch_countdown)
)

3
apiserver/plane/settings/common.py
View File

@@ -262,6 +262,9 @@ CELERY_IMPORTS = (
"plane.license.bgtasks.tracer",
# management tasks
"plane.bgtasks.dummy_data_task",
# issue sync tasks
"plane.bgtasks.issue_version_sync",
"plane.bgtasks.issue_description_version_sync",
)
# Sentry Settings