def special_access_required(func):
@login_required
@wraps(func)
def decorated_function(*args, **kwargs):
if current_user.type != UserType.ADMIN:
if 'club' in kwargs:
club = kwargs['club']
elif 'activity' in kwargs:
club = kwargs['activity'].club
else:
abort(403) # Admin-only page
if current_user.type == UserType.TEACHER:
if current_user != club.teacher:
abort(403)
else:
if current_user != club.leader:
abort(403)
return func(*args, **kwargs)
return decorated_function
python类args()的实例源码
def require_student_membership(func):
@login_required
@wraps(func)
def decorated_function(*args, **kwargs):
if 'club' in kwargs:
club = kwargs['club']
elif 'activity' in kwargs:
club = kwargs['activity'].club
else:
assert False
if current_user not in club.members:
abort(403)
return func(*args, **kwargs)
return decorated_function
def require_membership(func):
@login_required
@wraps(func)
def decorated_function(*args, **kwargs):
if current_user.type != UserType.ADMIN:
if 'club' in kwargs:
club = kwargs['club']
elif 'activity' in kwargs:
club = kwargs['activity'].club
else:
assert False
if current_user not in [club.teacher] + club.members:
abort(403)
return func(*args, **kwargs)
return decorated_function
def __call__(self, func):
@wraps(func)
def decorator(*args, **kwargs):
response = func(*args, **kwargs)
if not USE_CACHE:
return response
for path in self.routes:
route = '%s,%s,%s' % (path, json.dumps(self.args), None)
Cache.delete(unicode(route))
user = kwargs.get('user', None) if self.clear_for_user else None
Logger.debug(unicode('clear %s from cache' % (route)))
if user:
route = '%s,%s,%s' % (path, json.dumps(self.args), user)
Cache.delete(unicode(route))
Logger.debug(unicode('clear %s from cache' % (route)))
return response
return decorator
def perm_required(func):
""" Check if user can do actions
"""
@wraps(func)
def check_perm(*args, **kwargs):
if 'report' in kwargs:
GeneralController.check_perms(
method=request.method,
user=g.user,
report=kwargs['report']
)
if 'ticket' in kwargs:
GeneralController.check_perms(
method=request.method,
user=g.user,
ticket=kwargs['ticket']
)
if 'defendant' in kwargs and request.method != 'GET':
GeneralController.check_perms(
method=request.method,
user=g.user,
defendant=kwargs['defendant']
)
return func(*args, **kwargs)
return check_perm
def callback(self):
if 'code' not in request.args:
return None, None, None
oauth_session = self.service.get_auth_session(
data={'code': request.args['code'],
'grant_type': 'authorization_code',
'redirect_uri': self.get_callback_url()},
decoder=json.loads
)
me = oauth_session.get('/api/v1/user.json').json()
return (
me.get('data').get('id'),
me.get('data').get('first_name'),
me.get('data').get('email')
)
def __init__(self, *args, **kwargs):
print "init app"
super(App, self).__init__(*args, **kwargs)
self.config.from_object('config.Production')
self.config['ROOT_FOLDER'], _ = os.path.split(self.instance_path)
self.setup()
def set_request(cls, request, **kwargs):
uuid = uuid4().hex[:8]
cls.requests[uuid] = dict(request.args)
if kwargs:
cls.requests[uuid].update(kwargs)
return uuid
def collector(command):
command = command.replace('-', '_')
cmd = getattr(app._cr, 'do_%s' % command)
return cmd(**request.args)
def analyzer(command):
kwargs = dict(request.args)
uuid = parse_single_arg('request_id', kwargs)
if uuid:
kwargs = Request.get_request(uuid)
command = command.replace('-', '_')
cmd = getattr(app._ar, 'do_%s' % command)
return cmd(**kwargs)
def plot():
command = parse_single_arg('plot', request.args, default_val='raw-data')
uuid = Request.set_request(request, columns_data=True)
return render_template('chart.html', command=command, req_id=uuid)
def handle_verification():
"""
Handles Facebook verification call.
"""
return request.args['hub.challenge']
def showgauges():
return render_template("showgauges.html",
subjectivityavg=request.args["subjectivityavg"],
sentimentavg=request.args["sentimentavg"])
def telemetry(function):
def _wrapper(*args, **kwargs):
telemetry = Telemetry(referrer=request.referrer,
ip=md5(request.remote_addr).hexdigest(),
creation_date=datetime.now())
save(telemetry)
return function(*args, **kwargs)
return _wrapper
def index_page():
if request.args.get('epub'):
return create_epub()
else:
return create_page()
def create_page():
page = int(request.args.get("page", "0"))
limit = int(request.args.get("limit", "10"))
offset = page * limit
toots, count, count_all = get_toots(offset, limit)
accounts = Account.query.order_by(Account.username)
instances = Instance.query.order_by(Instance.domain)
blacklist_status = True if request.args.get('blacklisted', None) else False
if request.args.get('blacklisted') != 'ignore':
accounts = accounts.filter(Account.blacklisted == blacklist_status)
instances = instances.filter(Instance.blacklisted == blacklist_status)
pagination = {'page': page + 1,
'limit': limit}
pagination['next'] = "/?page=%s&" % (page + 1)
pagination['previous'] = "/?page=%s&" % (page - 1)
for key, value in request.args.iteritems():
if not key == "page":
pagination['next'] += "&%s=%s" % (key, value)
pagination['previous'] += "&%s=%s" % (key, value)
if count < limit:
pagination.pop('next')
if page == 0:
pagination.pop('previous')
pagination['page_count'] = int(count_all / limit) + 1
return render_template('index.html',
toots=toots,
accounts=accounts,
instances=instances,
pagination=pagination)
def authorize(self, *args, **kwargs):
"""Default authorization method."""
if self.api is not None:
return self.api.authorize(*args, **kwargs)
return current_user
def get_many(self, *args, **kwargs):
"""Get collection."""
return []
def get_one(self, *args, **kwargs):
"""Load resource."""
return kwargs.get(self.meta.name)
def filter(self, collection, *args, **kwargs):
"""Filter collection."""
return self.meta.filters.filter(collection, self, *args, **kwargs)
