<s> import datetime <EOL> import functools <EOL> import json <EOL> from django import http <EOL> from django . conf import settings <EOL> from django . core . exceptions import PermissionDenied <EOL> from django . db import connection , transaction <EOL> import commonware . log <EOL> from olympia . amo import get_user , set_user <EOL> from olympia . users . utils import get_task_user <EOL> from . import models as context <EOL> from . utils import JSONEncoder , redirect_for_login <EOL> task_log = commonware . log . getLogger ( '<STR_LIT>' ) <EOL> def login_required ( f = None , redirect = True ) : <EOL> """<STR_LIT>""" <EOL> def decorator ( func ) : <EOL> @ functools . wraps ( func ) <EOL> def wrapper ( request , * args , ** kw ) : <EOL> if request . user . is_authenticated ( ) : <EOL> return func ( request , * args , ** kw ) <EOL> else : <EOL> if redirect : <EOL> return redirect_for_login ( request ) <EOL> else : <EOL> return http . HttpResponse ( status = <NUM_LIT> ) <EOL> return wrapper <EOL> if f : <EOL> return decorator ( f ) <EOL> else : <EOL> return decorator <EOL> def post_required ( f ) : <EOL> @ functools . wraps ( f ) <EOL> def wrapper ( request , * args , ** kw ) : <EOL> if request . method != '<STR_LIT:POST>' : <EOL> return http . HttpResponseNotAllowed ( [ '<STR_LIT:POST>' ] ) <EOL> else : <EOL> return f ( request , * args , ** kw ) <EOL> return wrapper <EOL> def permission_required ( app , action ) : <EOL> def decorator ( f ) : <EOL> @ functools . wraps ( f ) <EOL> @ login_required <EOL> def wrapper ( request , * args , ** kw ) : <EOL> from olympia . access import acl <EOL> if acl . action_allowed ( request , app , action ) : <EOL> return f ( request , * args , ** kw ) <EOL> else : <EOL> raise PermissionDenied <EOL> return wrapper <EOL> return decorator <EOL> def any_permission_required ( pairs ) : <EOL> """<STR_LIT>""" <EOL> def decorator ( f ) : <EOL> @ functools . wraps ( f ) <EOL> @ login_required <EOL> def wrapper ( request , * args , ** kw ) : <EOL> from olympia . access import acl <EOL> for app , action in pairs : <EOL> if acl . action_allowed ( request , app , action ) : <EOL> return f ( request , * args , ** kw ) <EOL> raise PermissionDenied <EOL> return wrapper <EOL> return decorator <EOL> def restricted_content ( f ) : <EOL> """<STR_LIT>""" <EOL> @ functools . wraps ( f ) <EOL> def wrapper ( request , * args , ** kw ) : <EOL> from olympia . access import acl <EOL> if ( acl . action_allowed ( request , '<STR_LIT:*>' , '<STR_LIT:*>' ) <EOL> or not acl . action_allowed ( request , '<STR_LIT>' , '<STR_LIT>' ) ) : <EOL> return f ( request , * args , ** kw ) <EOL> else : <EOL> raise PermissionDenied <EOL> return wrapper <EOL> def modal_view ( f ) : <EOL> @ functools . wraps ( f ) <EOL> def wrapper ( * args , ** kw ) : <EOL> response = f ( * args , modal = True , ** kw ) <EOL> return response <EOL> return wrapper <EOL> def json_response ( response , has_trans = False , status_code = <NUM_LIT:200> ) : <EOL> """<STR_LIT>""" <EOL> if has_trans : <EOL> response = json . dumps ( response , cls = JSONEncoder ) <EOL> else : <EOL> response = json . dumps ( response ) <EOL> return http . HttpResponse ( response , <EOL> content_type = '<STR_LIT:application/json>' , <EOL> status = status_code ) <EOL> def json_view ( f = None , has_trans = False , status_code = <NUM_LIT:200> ) : <EOL> def decorator ( func ) : <EOL> @ functools . wraps ( func ) <EOL> def wrapper ( * args , ** kw ) : <EOL> response = func ( * args , ** kw ) <EOL> if isinstance ( response , http . HttpResponse ) : <EOL> return response <EOL> else : <EOL> return json_response ( response , has_trans = has_trans , <EOL> status_code = status_code ) <EOL> return wrapper <EOL> if f : <EOL> return decorator ( f ) <EOL> else : <EOL> return decorator <EOL> json_view . error = lambda s : http . HttpResponseBadRequest ( <EOL> json . dumps ( s ) , content_type = '<STR_LIT:application/json>' ) <EOL> class skip_cache ( object ) : <EOL> def __init__ ( self , f ) : <EOL> functools . update_wrapper ( self , f ) <EOL> self . f = f <EOL> def __call__ ( self , * args , ** kw ) : <EOL> with context . skip_cache ( ) : <EOL> return self . f ( * args , ** kw ) <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % self . f <EOL> def __get__ ( self , obj , typ = None ) : <EOL> return skip_cache ( self . f . __get__ ( obj , typ ) ) <EOL> def use_master ( f ) : <EOL> @ functools . wraps ( f ) <EOL> def wrapper ( * args , ** kw ) : <EOL> with context . use_master ( ) : <EOL> return f ( * args , ** kw ) <EOL> return wrapper <EOL> def write ( f ) : <EOL> return use_master ( skip_cache ( f ) ) <EOL> def set_modified_on ( f ) : <EOL> """<STR_LIT>""" <EOL> from olympia . amo . tasks import set_modified_on_object <EOL> @ functools . wraps ( f ) <EOL> def wrapper ( * args , ** kw ) : <EOL> objs = kw . pop ( '<STR_LIT>' , None ) <EOL> result = f ( * args , ** kw ) <EOL> if objs and result : <EOL> for obj in objs : <EOL> task_log . info ( '<STR_LIT>' % <EOL> ( obj . __class__ . __name__ , obj . pk ) ) <EOL> set_modified_on_object . apply_async ( <EOL> args = [ obj ] , kwargs = None , <EOL> eta = ( datetime . datetime . now ( ) + <EOL> datetime . timedelta ( seconds = settings . NFS_LAG_DELAY ) ) ) <EOL> return result <EOL> return wrapper <EOL> def allow_cross_site_request ( f ) : <EOL> """<STR_LIT>""" <EOL> @ functools . wraps ( f ) <EOL> def wrapper ( request , * args , ** kw ) : <EOL> response = f ( request , * args , ** kw ) <EOL> """<STR_LIT>""" <EOL> response [ '<STR_LIT>' ] = '<STR_LIT:*>' <EOL> response [ '<STR_LIT>' ] = '<STR_LIT:GET>' <EOL> return response <EOL> return wrapper <EOL> def set_task_user ( f ) : <EOL> """<STR_LIT>""" <EOL> @ functools . wraps ( f ) <EOL> def wrapper ( * args , ** kw ) : <EOL> old_user = get_user ( ) <EOL> set_user ( get_task_user ( ) ) <EOL> try : <EOL> result = f ( * args , ** kw ) <EOL> finally : <EOL> set_user ( old_user ) <EOL> return result <EOL> return wrapper <EOL> def allow_mine ( f ) : <EOL> @ functools . wraps ( f ) <EOL> def wrapper ( request , username , * args , ** kw ) : <EOL> """<STR_LIT>""" <EOL> if username == '<STR_LIT>' : <EOL> if not request . user . is_authenticated ( ) : <EOL> return redirect_for_login ( request ) <EOL> username = request . user . username <EOL> return f ( request , username , * args , ** kw ) <EOL> return wrapper <EOL> def atomic ( fn ) : <EOL> """<STR_LIT>""" <EOL> @ functools . wraps ( fn ) <EOL> @ write <EOL> def inner ( * args , ** kwargs ) : <EOL> cursor = connection . cursor ( ) <EOL> cursor . execute ( '<STR_LIT>' ) <EOL> with transaction . atomic ( ) : <EOL> return fn ( * args , ** kwargs ) <EOL> inner . non_atomic = fn <EOL> return inner </s>
