Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
Tree: 9ec50ac442
Fetching contributors…

Cannot retrieve contributors at this time

220 lines (171 sloc) 6.372 kB
"""
The :mod:`permissions` module bundles a set of permission classes that are used
for checking if a request passes a certain set of constraints. You can assign a permission
class to your view by setting your View's :attr:`permissions` class attribute.
"""
from django.core.cache import cache
from djangorestframework import status
from djangorestframework.response import ErrorResponse
import time
__all__ = (
'BasePermission',
'FullAnonAccess',
'IsAuthenticated',
'IsAdminUser',
'IsUserOrIsAnonReadOnly',
'PerUserThrottling',
'PerViewThrottling',
'PerResourceThrottling'
)
_403_FORBIDDEN_RESPONSE = ErrorResponse(
status.HTTP_403_FORBIDDEN,
{'detail': 'You do not have permission to access this resource. ' +
'You may need to login or otherwise authenticate the request.'})
_503_SERVICE_UNAVAILABLE = ErrorResponse(
status.HTTP_503_SERVICE_UNAVAILABLE,
{'detail': 'request was throttled'})
class BasePermission(object):
"""
A base class from which all permission classes should inherit.
"""
def __init__(self, view):
"""
Permission classes are always passed the current view on creation.
"""
self.view = view
def check_permission(self, auth):
"""
Should simply return, or raise an :exc:`response.ErrorResponse`.
"""
pass
class FullAnonAccess(BasePermission):
"""
Allows full access.
"""
def check_permission(self, user):
pass
class IsAuthenticated(BasePermission):
"""
Allows access only to authenticated users.
"""
def check_permission(self, user):
if not user.is_authenticated():
raise _403_FORBIDDEN_RESPONSE
class IsAdminUser(BasePermission):
"""
Allows access only to admin users.
"""
def check_permission(self, user):
if not user.is_admin():
raise _403_FORBIDDEN_RESPONSE
class IsUserOrIsAnonReadOnly(BasePermission):
"""
The request is authenticated as a user, or is a read-only request.
"""
def check_permission(self, user):
if (not user.is_authenticated() and
self.view.method != 'GET' and
self.view.method != 'HEAD'):
raise _403_FORBIDDEN_RESPONSE
class BaseThrottle(BasePermission):
"""
Rate throttling of requests.
The rate (requests / seconds) is set by a :attr:`throttle` attribute
on the :class:`.View` class. The attribute is a string of the form 'number of
requests/period'.
Period should be one of: ('s', 'sec', 'm', 'min', 'h', 'hour', 'd', 'day')
Previous request information used for throttling is stored in the cache.
"""
attr_name = 'throttle'
default = '0/sec'
timer = time.time
def get_cache_key(self):
"""
Should return a unique cache-key which can be used for throttling.
Muse be overridden.
"""
pass
def check_permission(self, auth):
"""
Check the throttling.
Return `None` or raise an :exc:`.ErrorResponse`.
"""
num, period = getattr(self.view, self.attr_name, self.default).split('/')
self.num_requests = int(num)
self.duration = {'s': 1, 'm': 60, 'h': 3600, 'd': 86400}[period[0]]
self.auth = auth
self.check_throttle()
def check_throttle(self):
"""
Implement the check to see if the request should be throttled.
On success calls :meth:`throttle_success`.
On failure calls :meth:`throttle_failure`.
"""
self.key = self.get_cache_key()
self.history = cache.get(self.key, [])
self.now = self.timer()
# Drop any requests from the history which have now passed the
# throttle duration
while self.history and self.history[-1] <= self.now - self.duration:
self.history.pop()
if len(self.history) >= self.num_requests:
self.throttle_failure()
else:
self.throttle_success()
def throttle_success(self):
"""
Inserts the current request's timestamp along with the key
into the cache.
"""
self.history.insert(0, self.now)
cache.set(self.key, self.history, self.duration)
header = 'status=SUCCESS; next=%s sec' % self.next()
self.view.add_header('X-Throttle', header)
def throttle_failure(self):
"""
Called when a request to the API has failed due to throttling.
Raises a '503 service unavailable' response.
"""
header = 'status=FAILURE; next=%s sec' % self.next()
self.view.add_header('X-Throttle', header)
raise _503_SERVICE_UNAVAILABLE
def next(self):
"""
Returns the recommended next request time in seconds.
"""
if self.history:
remaining_duration = self.duration - (self.now - self.history[-1])
else:
remaining_duration = self.duration
available_requests = self.num_requests - len(self.history) + 1
return '%.2f' % (remaining_duration / float(available_requests))
class PerUserThrottling(BaseThrottle):
"""
Limits the rate of API calls that may be made by a given user.
The user id will be used as a unique identifier if the user is
authenticated. For anonymous requests, the IP address of the client will
be used.
"""
def get_cache_key(self):
if self.auth.is_authenticated():
ident = str(self.auth)
else:
ident = self.view.request.META.get('REMOTE_ADDR', None)
return 'throttle_user_%s' % ident
class PerViewThrottling(BaseThrottle):
"""
Limits the rate of API calls that may be used on a given view.
The class name of the view is used as a unique identifier to
throttle against.
"""
def get_cache_key(self):
return 'throttle_view_%s' % self.view.__class__.__name__
class PerResourceThrottling(BaseThrottle):
"""
Limits the rate of API calls that may be used against all views on
a given resource.
The class name of the resource is used as a unique identifier to
throttle against.
"""
def get_cache_key(self):
return 'throttle_resource_%s' % self.view.resource.__class__.__name__
Jump to Line
Something went wrong with that request. Please try again.