pangea/django-orchestra
7
0
Fork 0
Code Issues 1 Pull requests 2 Packages Projects Releases Wiki Activity

Import ContextDecorator from std library

Browse source 
https://docs.djangoproject.com/en/4.2/releases/3.0/#removed-private-python-2-compatibility-apis
This commit is contained in:
Santiago L 2023-11-29 10:38:54 +01:00
parent 3ab341a473
commit 049ff451ec
1 changed files with 7 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
orchestra/contrib/orchestration/managers.py
View file

@ -1,14 +1,14 @@
import sys
from contextlib import ContextDecorator
from threading import local
from django.contrib.admin.models import LogEntry
from django.db.models.signals import pre_delete, post_save, m2m_changed
from django.db.models.signals import m2m_changed, post_save, pre_delete
from django.dispatch import receiver
from django.utils.decorators import ContextDecorator
from orchestra.utils.python import OrderedSet
from . import manager, Operation, helpers
from . import Operation, helpers, manager
from .middlewares import OperationsMiddleware
from .models import BackendLog, BackendOperation
@ -37,7 +37,7 @@ def m2m_collector(sender, *args, **kwargs):
class orchestrate(ContextDecorator):
"""
Context manager for triggering backend operations out of request-response cycle, e.g. shell
with orchestrate():
user = SystemUser.objects.get(username='rata')
user.shell = '/dev/null'
@ -46,7 +46,7 @@ class orchestrate(ContextDecorator):
thread_locals = local()
thread_locals.pending_operations = None
thread_locals.route_cache = None
@classmethod
def collect(cls, action, **kwargs):
""" Collects all pending operations derived from model signals """
@ -57,14 +57,14 @@ class orchestrate(ContextDecorator):
kwargs['route_cache'] = cls.thread_locals.route_cache
instance = kwargs.pop('instance')
manager.collect(instance, action, **kwargs)
def __enter__(self):
cls = type(self)
self.old_pending_operations = cls.thread_locals.pending_operations
cls.thread_locals.pending_operations = OrderedSet()
self.old_route_cache = cls.thread_locals.route_cache
cls.thread_locals.route_cache = {}
def __exit__(self, exc_type, exc_value, traceback):
cls = type(self)
if not exc_type: