fix: jms upgrade_db bug

pull/8928/head
feng 2022-11-21 19:54:00 +08:00
parent f39a3a34e4
commit 4b26fb3e6e
2 changed files with 16 additions and 13 deletions

View File

@ -1,14 +1,16 @@
import ast import ast
from celery import signals
from django.db import transaction from django.db import transaction
from django.core.cache import cache
from django.dispatch import receiver from django.dispatch import receiver
from django.db.utils import ProgrammingError
from django.utils import translation, timezone from django.utils import translation, timezone
from django.utils.translation import gettext as _ from django.utils.translation import gettext as _
from django.core.cache import cache
from celery import signals, current_app
from common.db.utils import close_old_connections, get_logger
from common.signals import django_ready from common.signals import django_ready
from common.db.utils import close_old_connections, get_logger
from .celery import app from .celery import app
from .models import CeleryTaskExecution, CeleryTask from .models import CeleryTaskExecution, CeleryTask
@ -23,15 +25,15 @@ def sync_registered_tasks(*args, **kwargs):
with transaction.atomic(): with transaction.atomic():
try: try:
db_tasks = CeleryTask.objects.all() db_tasks = CeleryTask.objects.all()
except Exception as e: celery_task_names = [key for key in app.tasks]
return db_task_names = db_tasks.values_list('name', flat=True)
celery_task_names = [key for key in app.tasks]
db_task_names = db_tasks.values_list('name', flat=True)
db_tasks.exclude(name__in=celery_task_names).delete() db_tasks.exclude(name__in=celery_task_names).delete()
not_in_db_tasks = set(celery_task_names) - set(db_task_names) not_in_db_tasks = set(celery_task_names) - set(db_task_names)
tasks_to_create = [CeleryTask(name=name) for name in not_in_db_tasks] tasks_to_create = [CeleryTask(name=name) for name in not_in_db_tasks]
CeleryTask.objects.bulk_create(tasks_to_create) CeleryTask.objects.bulk_create(tasks_to_create)
except ProgrammingError:
pass
@signals.before_task_publish.connect @signals.before_task_publish.connect
@ -45,7 +47,7 @@ def before_task_publish(headers=None, **kwargs):
@signals.task_prerun.connect @signals.task_prerun.connect
def on_celery_task_pre_run(task_id='', **kwargs): def on_celery_task_pre_run(task_id='', **kwargs):
# 更新状态 # 更新状态
CeleryTaskExecution.objects.filter(id=task_id)\ CeleryTaskExecution.objects.filter(id=task_id) \
.update(state='RUNNING', date_start=timezone.now()) .update(state='RUNNING', date_start=timezone.now())
# 关闭之前的数据库连接 # 关闭之前的数据库连接
close_old_connections() close_old_connections()

View File

@ -7,6 +7,7 @@ from functools import partial
import django.db.utils import django.db.utils
from django.dispatch import receiver from django.dispatch import receiver
from django.conf import settings from django.conf import settings
from django.db.utils import ProgrammingError, OperationalError
from django.utils.functional import LazyObject from django.utils.functional import LazyObject
from django.db.models.signals import post_save, pre_delete, m2m_changed from django.db.models.signals import post_save, pre_delete, m2m_changed
@ -48,7 +49,7 @@ def subscribe_orgs_mapping_expire(sender, **kwargs):
if settings.DEBUG: if settings.DEBUG:
try: try:
set_to_default_org() set_to_default_org()
except django.db.utils.OperationalError: except (ProgrammingError, OperationalError):
pass pass
def keep_subscribe_org_mapping(): def keep_subscribe_org_mapping():