Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@view_config(route_name='iam_policy_create', renderer=TEMPLATE, request_method='POST')
def iam_policy_create(self):
"""Handles the POST from the Create IAM Policy wizard"""
target_route = '{0}_view'.format(self.target_type) # 'user_view' or 'group_view'
location = self.request.route_url(target_route, name=self.target_name) # redirect to detail page after submit
if self.create_form.validate():
policy_name = self.request.params.get('name')
policy_json = self.request.params.get('policy', '{}')
try:
if self.target_type == 'user':
caller = self.iam_conn.put_user_policy
else:
caller = self.iam_conn.put_group_policy
caller(self.target_name, policy_name, policy_json)
prefix = _(u'Successfully created IAM policy')
msg = '{0} {1}'.format(prefix, policy_name)
queue = Notification.SUCCESS
@view_config(name='upload', context=File, request_method='GET',
permission='edit')
def get_upload(context, request):
external = context.propsheets.get('external', {})
upload_credentials = external.get('upload_credentials')
# Show s3 location info for files originally submitted to EDW.
if upload_credentials is None and external.get('service') == 's3':
upload_credentials = {
'upload_url': 's3://{bucket}/{key}'.format(**external),
}
return {
'@graph': [{
'@id': request.resource_path(context),
'upload_credentials': upload_credentials,
}],
@view_config(name="user", subpath=("username/*"), request_method="DELETE")
def delete_user(self):
""" Delete a user """
username = self.request.named_subpaths["username"]
self.request.access.delete_user(username)
return self.request.response
@view_config(name='add_document', context=Folder)
def add_document(self):
# Make a new Document
title = self.request.POST['document_title']
name = str(randint(0, 999999))
new_document = Document(name, self.context, title)
self.context[name] = new_document
# Redirect to the new document
url = self.request.resource_url(new_document)
return HTTPFound(location=url)
@view_config(route_name='tree_invalidate', renderer='json', accept='application/json', permission='edit')
def invalidate_tree(self):
invalidate_cache()
return Response(status_int=200)
@view_config(route_name='admin_data_histogram_dayssincepurchase',
renderer='json',
permission='manage')
def admin_data_histogram_dayssincepurchase(request):
return user_dayssincepurchase_histogram()
@view_config(route_name='scalinggroup_history', renderer=TEMPLATE, request_method='GET')
def scalinggroup_history(self):
return self.render_dict
@view_config(
route_name='view_tasks',
renderer='templates/task/view_tasks.jinja2',
permission='View_Task'
)
def view_tasks(request):
"""runs when viewing tasks of a TaskableEntity
"""
login = authenticated_userid(request)
logged_in_user = User.query.filter_by(login=login).first()
entity_id = request.matchdict['entity_id']
entity = Entity.query.filter(Entity.id==entity_id).first()
return {
'entity': entity
}
@view_config(name="add_folder", context=Folder)
def add_folder(self):
# Make a new Folder
title = self.request.POST['folder_title']
name = str(randint(0, 999999))
new_folder = Folder(title=title)
DBSession.add(new_folder)
self.context[name] = new_folder
# Redirect to the new folder
url = self.request.resource_url(new_folder)
return HTTPFound(location=url)
@view_config(
name='details',
request_method='GET',
renderer='travelcrm:templates/calculations/details.mako',
permission='view'
)
def details(self):
calculation = Calculation.get(self.request.params.get('id'))
contract_resource = None
if calculation.contract:
resource_cls = get_resource_class(
calculation.contract.resource.resource_type.name
)
contract_resource = resource_cls(self.request)
return {
'item': calculation,
'contract_resource': contract_resource