djangofloor.decorators

Decorators to declare signals and remote functions

ALso define common functions for allowing (or not) signal calls to user, and several tools for checking arguments provided to the signal (or function).

Decorators

Three decorators are provided, for creating signals, websocket functions or special signals for validating forms. Original functions are left unmodified by the decorators. These decorators instantiate a djangofloor.decorators.Connection object and stores it in the corresponding dict (REGISTERED_FUNCTIONS or REGISTERED_SIGNALS).

Restrict signal/function use

When creating a connection, you provide a callable that checks if the browser is allowed to call this code. By default, the registered code can only be called from Python code. The callable takes three arguments:

Argument validators

The registered Python code can use py3 annotation for specifying data types.

from djangofloor.decorators import Choice, RE, SerializedForm
from django import forms

class MyForm(forms.Form):
    test = forms.CharField()

@signal(path='demo.signal')
def my_signal(window_info, kwarg1: Choice([1, 2], int)=1, kwarg2: Re('^\d+$', int)=2,
        kwarg3: SerializedForm(MyForm)):
   assert isinstance(kwarg1, int)
   assert isinstance(kwarg2, int)
   assert isinstance(kwarg3, MyForm)

scall(window_info, 'demo.signal', to=[SERVER], kwarg1="1", kwarg2="12312", kwarg3=[{'value': '12', 'name': 'test'}])
class djangofloor.decorators.Choice(values, caster=None)[source]

used to check if a value is among some valid choices.

Example (requires Python 3.2+), for a function that can only two values:

@signal(path='myproject.signals.test')
def test(window_info, value: Choice([True, False])):
    pass

Your code wan’t be called if value is not True or False.

Parameters:caster – callable to convert the provided deserialized JSON data before checking its validity.
class djangofloor.decorators.Connection(fn, path=None, is_allowed_to=<function server_side>, queue=None)[source]

Parent class of a registered signal or remote function. Do not use it directly.

check(kwargs)[source]

Check the provided kwargs and apply provided annotations to it. Return None if something is invalid (like an error raised by an annotation or a missing argument).

get_queue(window_info, original_kwargs)[source]

Provide the Celery queue name as a string.

register()[source]

Register the Python code to the right dict.

required_function_arg = 'window_info'
signature_check(fn)[source]

Analyze the signature of the registered Python code, and store the annotations. Check if the first argument is window_info.

class djangofloor.decorators.DynamicQueueName[source]

Allow to dynamically select a Celery queue when the signal is called. You can use it if all signals of a user must be processed by the same worker, but you still

want to dispatch signals to several workers.
get_available_queues()[source]

return the set of all queues that can be returned by the __call__ method. However, if this method is not implemented, the impact is currently limited:

  • the monitoring view will not display all required queues,
  • the systemd service files (provided by the packaging command) will not create all required workers.
class djangofloor.decorators.FormValidator(fn, path=None, is_allowed_to=<function server_side>, queue=None)[source]

Special signal, dedicated to dynamically validate a HTML form.

However, files cannot be sent in the validation process.

signature_check(fn)[source]

override the default method for checking the arguments, since they are independent from the Django Form.

class djangofloor.decorators.FunctionConnection(fn, path=None, is_allowed_to=<function server_side>, queue=None)[source]

represent a WS function

register()[source]

register the WS function into the REGISTERED_FUNCTIONS dict

class djangofloor.decorators.LegacySignalConnection(fn, path=None, is_allowed_to=<function server_side>, queue=None)[source]

Deprecated since version 1.0: do not use it

class djangofloor.decorators.RE(value, caster=None, flags=0)[source]

used to check if a string value matches a given regexp.

Example (requires Python 3.2+), for a function that can only handle a string of the form 123a456:

@signal(path='myproject.signals.test')
def test(window_info, value: RE('\d{3}a\d{3}')):
    pass

Your code won’t be called for values like “abc”.

Parameters:
  • value (str) – regexp pattern
  • caster (callable or None) – if not None, any callable applied to the value (if valid)
  • flags (int) – regexp flags passed to re.compile
class djangofloor.decorators.RandomDynamicQueueName(prefix: str, size: int)[source]

Return a random queue on each signal call.

This class is somewhat useless since you could just run more workers on the same queue.

>>> q = RandomDynamicQueueName('prefix-', 2)
>>> q.get_available_queues() == {'prefix-0', 'prefix-1'}
True
>>> q(None, None, None) in {'prefix-0', 'prefix-1'}
True
get_available_queues()[source]
class djangofloor.decorators.SerializedForm(form_cls)[source]

Transform values sent by JS to a Django form.

Given a form and a list of dict, transforms the list into a django.http.QueryDict and initialize the form with it.

>>> from django import forms
>>> class SimpleForm(forms.Form):
...    field = forms.CharField()
...
>>> x = SerializedForm(SimpleForm)
>>> form = x([{'name': 'field', 'value': 'object'}])
>>> form.is_valid()
True

How to use it with Python3:

@signal(path='myproject.signals.test')
def test(window_info, value: SerializedForm(SimpleForm), other: int):
    print(value.is_valid())

How to use it with Python2:

@signal(path='myproject.signals.test')
def test(window_info, value, other):
    value = SerializedForm(SimpleForm)(value)
    print(value.is_valid())

On the JS side, you can serialize the form with JQuery:

<form onsubmit="return $.df.call('myproject.signals.test', {value: $(this).serializeArray(), other: 42})">
    <input name='field' value='test' type='text'>
</form>
class djangofloor.decorators.SignalConnection(fn, path=None, is_allowed_to=<function server_side>, queue=None)[source]

represents a connected signal.

call(window_info, **kwargs)[source]
register()[source]

register the signal into the REGISTERED_SIGNALS dict

djangofloor.decorators.connect(fn=None, path=None, delayed=False, allow_from_client=True, auth_required=True)[source]

Deprecated since version 1.0: do not use it

djangofloor.decorators.everyone(connection, window_info, kwargs)[source]

allow everyone to call a Python WS signal or remote function

>>> @signal(is_allowed_to=everyone)
>>> def my_signal(request, arg1=None):
>>>     print(request, arg1)
djangofloor.decorators.function(fn=None, path=None, is_allowed_to=<function server_side>, queue=None)[source]

Allow the following Python code to be called from the JavaScript code. The result of this function is serialized (with JSON and settings.WEBSOCKET_SIGNAL_ENCODER) before being sent to the JavaScript part.

from djangofloor.decorators import function, everyone

@function(path='myproject.myfunc', is_allowed_to=everyone)
def myfunc(window_info, arg=None)
    print(arg)
    return 42

The this function can be called from your JavaScript code:

$.dfws.myproject.myfunc({arg: 3123}).then(function(result) { alert(result); });
class djangofloor.decorators.has_perm(perm)[source]

restrict a WS signal or a WS function to users with permission “perm”

>>> @signal(is_allowed_to=has_perm('app_label.codename'))
>>> def my_signal(request, arg1=None):
>>>     print(request, arg1)
djangofloor.decorators.is_anonymous(connection, window_info, kwargs)[source]

restrict a WS signal or a WS function to anonymous users

>>> @signal(is_allowed_to=is_anonymous)
>>> def my_signal(request, arg1=None):
>>>     print(request, arg1)
djangofloor.decorators.is_authenticated(connection, window_info, kwargs)[source]

restrict a WS signal or a WS function to authenticated users

>>> @signal(is_allowed_to=is_authenticated)
>>> def my_signal(request, arg1=None):
>>>     print(request, arg1)
djangofloor.decorators.is_staff(connection, window_info, kwargs)[source]

restrict a WS signal or a WS function to staff users

>>> @signal(is_allowed_to=is_staff)
>>> def my_signal(request, arg1=None):
>>>     print(request, arg1)
djangofloor.decorators.is_superuser(connection, window_info, kwargs)[source]

restrict a WS signal or a WS function to superusers

>>> @signal(is_allowed_to=is_superuser)
>>> def my_signal(request, arg1=None):
>>>     print(request, arg1)
djangofloor.decorators.server_side(connection, window_info, kwargs)[source]

never allows a signal to be called from WebSockets; this signal can only be called from Python code. This is the default choice.

>>> @signal(is_allowed_to=server_side)
>>> def my_signal(window_info, arg1=None):
>>>     print(window_info, arg1)
djangofloor.decorators.signal(fn=None, path=None, is_allowed_to=<function server_side>, queue=None, cls=<class 'djangofloor.decorators.SignalConnection'>)[source]

Decorator to use for registering a new signal. This decorator returns the original callable as-is.

djangofloor.decorators.validate_form(form_cls=None, path=None, is_allowed_to=<function server_side>, queue=None)[source]

Decorator for automatically validating HTML forms. Just add it to your Python code and set the ‘onchange’ attribute to your HTML code. The path argument should be unique to your form class.

param form_cls:any subclass of django.forms.Form
param path:unique name of your form
param is_allowed_to:
 callable for restricting the use of the form validation
param queue:name (or callable) for ensuring small response times
from djangofloor.decorators import everyone, validate_form

@validate_form(path='djangofloor.validate.search', is_allowed_to=everyone, queue='fast')
class MyForm(forms.Form):
    name = forms.CharField()
    ...
<form onchange="$.df.validateForm(this, 'djangofloor.validate.search');"  action="?" method="post">
    {% csrf_token %}
    {% bootstrap_form form %}
    <input type="submit" class="btn btn-primary" value="{% trans 'Search' %}">
</form>