yoyowallet / django-idempotency-key Goto Github PK

If the request has previously succeeded then the original data will be returned and
nothing new is created.

class CustomExemptIdempotencyKeyMiddleware(ExemptIdempotencyKeyMiddleware):
    def __init__(self, get_response):
        super().__init__(get_response)
        self.storage_lock_class = utils.get_lock_class()

    def reload_storage_lock(self, encoded_key):
        self.storage_lock = self.storage_lock_class(encoded_key)

    def release_lock(self):
        self.storage_lock.release()

    def acquire_lock(self):
        return self.storage_lock.acquire()

    def generate_response(self, request, encoded_key, lock=None):
        # The default storage lock is established with a common lock prefix shared among all locks.
        # To generate a lock with an encoded key, simply reload the storage lock.
        self.reload_storage_lock(encoded_key)

        if lock is None:
            lock = utils.get_lock_enable()

        if not lock:
            return self.perform_generate_response(request, encoded_key)

        # If there was a timeout for a lock on the storage object then return a
        # HTTP_423_LOCKED
        if not self.acquire_lock():
            return resource_locked(request, None)
        try:
            return self.perform_generate_response(request, encoded_key)
        finally:
            self.release_lock()

class CustomMultiProcessRedisLock(IdempotencyKeyLock):
    """
    Should be used if a lock is required across processes. Note that this class uses
    Redis in order to perform the lock.
    """

    def __init__(self, idempotency_key=None):
        location = utils.get_lock_location()
        if location is None or location == "":
            raise ValueError("Redis server location must be set in the settings file.")

        self.redis_obj = Redis.from_url(location)

        lock_name = (
            f"{utils.get_lock_name()}-{idempotency_key}"
            if idempotency_key
            else utils.get_lock_name()
        )
        self.storage_lock = self.redis_obj.lock(
            name=lock_name,
            # Time before lock is forcefully released.
            timeout=utils.get_lock_time_to_live(),
            blocking_timeout=utils.get_lock_timeout(),
        )

    def acquire(self, *args, **kwargs) -> bool:
        return self.storage_lock.acquire(blocking=False)

    def release(self):
        # Just incase if Middleware releases without acquiring. Less likely. Ignoring it instead of popping up an error.
        with suppress(Exception):
            self.storage_lock.release()

class DebugMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        if request.method == 'POST':
            print('Headers', request.headers)
            print('META', request.META)
            idempotency_key = request.META.get("HTTP_IDEMPOTENCY_KEY")
            print(f'idempotency_key: {idempotency_key}')
        response = self.get_response(request)
        return response

class OrdersCreateViews(CreateAPIView):
    """
    View to create new Orders.
    """
    serializer_class = OrderSerializer
    authentication_classes = (TokenAuthentication,)

    @idempotency_key
    def create(self, request, *args, **kwargs):
        serializer = self.get_serializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        self.perform_create(serializer)
        headers = self.get_success_headers(serializer.data)
        return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers)