hishnash

class TabSection(Directive):
    has_content = True  # this directive can have children
    required_arguments = 1  # has 1 arg
    optional_arguments = 0
    
     # add a kwarg `keyworks` these are used for search
    option_spec = {'keywords': directives.unchanged} 
    
    # we need a new line after the args before the child content
    final_argument_whitespace = True

hishnash

URL Encoding

Visual

A Percent-encoding converter conforming to RFC3986.

Code

hishnash

HMAC Signature :keywords hmac

Visual

A Dynamic Value providing HMAC keyed hash function (Message authentication code (MAC)).

Code

hishnash

from collections import deque

import psutil

from settings import settings
from collections import AsyncIterable


class AsyncIteratorWrapper(AsyncIterable):
    def __init__(self, obj):

hishnash

_type: request
_uuid: 0848dc66-833e-4d98-8362-dda98ee319cd
bodyString: null
followRedirects: false
headers: []
method:
  ref: d289f99d-8886-443c-b7e8-1f55934837f0
redirectAuthorization: false
redirectMethod: false
sendCookies: true

hishnash

def exclusive_lock(method):
    @wraps(method)
    def wrapped_method(self, request, *args, document_pk=None, **kwargs):
        document = self.assert_user_has_write_permission(document_pk)
        document_state = self.get_document_state(document=document)
        # currently there is no transaction,
        # the context managers will create/release as needed
        with document_state.get_exclusive_lock():
            # we have exclusive access
            try:

hishnash

from contextlib import contextmanager
from django.db import transaction
from rest_framework.exceptions import APIException

class LockTakenException(APIException, OperationalError):
    status_code = 423
    default_detail = 'This item is currently locked by another request.'
    client_code = 'ObjectLocked'

def ensure_atomic(func):

hishnash

class LockableModel(TimeStampedModel):
    class Meta:
        abstract = True

    @ensure_atomic
    @contextmanager
    def get_exclusive_lock(self,
                           _connection=connection,
                           timeout=settings.PG_LOCKING_TIMEOUT_MS,
                           wait=True):