Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def post_submit(self, proj, asset_id, comment=None):
"""Submit an asset to supervisors for approval."""
session = session_get()
user = tmpl_context.user
asset = asset_get(proj, asset_id)
if not asset.submitted and not asset.approved:
asset.submit(user)
text = u'[%s v%03d]\n%s' % (_('submitted'), asset.current.ver,
comment or '')
asset.current.notes.append(Note(user, text))
session.refresh(asset.current.annotable)
# log into Journal
journal.add(user, 'submitted %s' % asset)
# send a stomp message to notify clients
notify.send(asset)
notify.ancestors(asset)
return dict(msg='submitted asset "%s"' % asset.path,
result='success')
def get_delete(self, proj, asset_id, **kwargs):
"""Display a DELETE confirmation form."""
asset = asset_get(proj, asset_id)
f_confirm.custom_method = 'DELETE'
f_confirm.value = dict(proj=asset.project.id,
project_name_=asset.project.name,
container_=asset.parent.owner.path,
category_id_=asset.category.id,
asset_id=asset.id,
asset_name_=asset.name,
)
warning = ('This will only delete the asset entry in the database. '
'The data must be deleted manually if needed.')
tmpl_context.form = f_confirm
return dict(title='%s %s?' % (_('Are you sure you want to delete'),
asset.path), warning=warning)
def get_sendback(self, proj, asset_id, **kwargs):
"""Display a SENDBACK form."""
asset = asset_get(proj, asset_id)
f_status.custom_method = 'SENDBACK'
f_status.value = dict(proj=asset.project.id,
asset_id=asset.id,
project_name_=asset.project.name,
container_=asset.parent.owner.path,
category_id_=asset.category.id,
asset_name_=asset.name,
)
tmpl_context.form = f_status
return dict(title='%s: %s' % (_('Send back for revisions'), asset.path))
def post_approve(self, proj, asset_id, comment=None):
"""Approve an asset submitted for approval."""
session = session_get()
user = tmpl_context.user
asset = asset_get(proj, asset_id)
if asset.submitted and not asset.approved:
asset.approve(user)
text = u'[%s v%03d]\n%s' % (_('approved'), asset.current.ver,
comment or '')
asset.current.notes.append(Note(user, text))
session.refresh(asset.current.annotable)
# log into Journal
journal.add(user, 'approved %s' % asset)
# send a stomp message to notify clients
notify.send(asset)
notify.ancestors(asset)
return dict(msg='approved asset "%s"' % asset.path,
result='success')
def post_delete(self, proj, asset_id):
"""Delete an asset.
Only delete the asset record from the db, the asset file(s) must be
removed manually.
(This should help prevent awful accidents) ;)
"""
session = session_get()
user = tmpl_context.user
asset = asset_get(proj, asset_id)
session.delete(asset)
# delete association objects or they will be orphaned
session.flush()
for ver in asset.versions:
session.delete(ver.annotable)
session.delete(asset.taggable)
# log into Journal
journal.add(user, 'deleted %s' % asset)
# send a stomp message to notify clients
notify.send(asset, update_type='deleted')
return dict(msg='deleted asset "%s"' % asset.path, result='success')
def get_approve(self, proj, asset_id, **kwargs):
"""Display a APPROVE form."""
asset = asset_get(proj, asset_id)
f_status.custom_method = 'APPROVE'
f_status.value = dict(proj=asset.project.id,
asset_id=asset.id,
project_name_=asset.project.name,
container_=asset.parent.owner.path,
category_id_=asset.category.id,
asset_name_=asset.name,
)
tmpl_context.form = f_status
return dict(title='%s: %s' % (_('Approve'), asset.path))
def post_recall(self, proj, asset_id, comment=None):
"""Recall an asset submitted for approval."""
session = session_get()
user = tmpl_context.user
asset = asset_get(proj, asset_id)
if asset.submitted and not asset.approved:
asset.recall(user)
text = u'[%s v%03d]\n%s' % (_('recalled'), asset.current.ver,
comment or '')
asset.current.notes.append(Note(user, text))
session.refresh(asset.current.annotable)
# log into Journal
journal.add(user, 'recall submission for %s' % asset)
# send a stomp message to notify clients
notify.send(asset)
notify.ancestors(asset)
return dict(msg='recalled submission for asset "%s"' % asset.path,
result='success')
def asset_set_active(func, *args, **kwargs):
"""Extract the current asset id from the args passed to the function
and puts the corresponding asset in the template context.
If the asset id is not valid raise an error."""
if 'asset_id' in kwargs:
asset_id = kwargs['asset_id']
elif len(args)>2:
asset_id = args[2]
else:
raise SPAMError('No asset_id defined')
project = tmpl_context.project
tmpl_context.asset = asset_get(project.id, asset_id)
return func(*args, **kwargs)
def get_one(self, proj, asset_id):
"""Return a `standalone` page with the asset history"""
tmpl_context.t_history = t_history
asset = asset_get(proj, asset_id)
# thumb, ver, note
history = []
for ver in asset.versions:
if ver.notes:
for note in ver.notes:
history.append(dict(id=None, proj_id=None, thumbnail=None,
ver=None, fmtver=None, header=note.header,
text=note.text, lines=note.lines))
else:
history.append(dict(id=None, proj_id=None, thumbnail=None,
ver=None, fmtver=None, header='',
text='', lines=[]))
history[-1]['id'] = ver.id
history[-1]['proj_id'] = ver.asset.proj_id
def post_publish(self, proj, asset_id, uploaded, comment=None,
uploader=None):
"""Publish a new version of an asset.
This will commit to the repo the file(s) already uploaded in a temporary
storage area, and create a thumbnail and preview if required.
"""
session = session_get()
asset = asset_get(proj, asset_id)
user = tmpl_context.user
if not asset.checkedout or user != asset.owner:
return dict(msg='cannot publish asset "%s"' % asset_id,
result='failed')
if isinstance(uploaded, list):
# the form might send empty strings, so we strip them
uploaded = [uf for uf in uploaded if uf]
else:
uploaded = [uploaded]
# check that uploaded file extension matches asset name
name, ext = os.path.splitext(asset.name)
for uf in uploaded:
uf_name, uf_ext = os.path.splitext(uf)