Page MenuHome
No OneTemporary

File Metadata

Mon, Nov 18, 11:46 PM

"""Task management."""
import attr
import flask
import flask_login
from dateutil import parser
import bson
import pillarsdk
from pillar import attrs_extra
from pillar.api.activities import register_activity
from pillar.web.system_util import pillar_api
from pillar.api.utils import authentication
from attract.node_types.task import node_type_task
class TaskManager(object):
_log = attrs_extra.log('%s.TaskManager' % __name__)
def create_task(self, project, task_type=None, parent=None):
"""Creates a new task, owned by the current user.
:rtype: pillarsdk.Node
from pillar.web.jinja import format_undertitle
api = pillar_api()
node_type = project.get_node_type(node_type_task['name'])
if not node_type:
raise ValueError('Project %s not set up for Attract' % project._id)
node_props = dict(
name='New task',
'status': node_type['dyn_schema']['status']['default'],
if task_type:
node_props['name'] = format_undertitle(task_type)
node_props['properties']['task_type'] = task_type
if parent:
node_props['parent'] = parent
task = pillarsdk.Node(node_props)
return pillarsdk.Node.find(task._id, api=api)
def edit_task(self, task_id, **fields):
"""Edits a task.
:type task_id: str
:type fields: dict
:rtype: pillarsdk.Node
api = pillar_api()
task = pillarsdk.Node.find(task_id, api=api)
task._etag = fields.pop('_etag') = fields.pop('name')
task.description = fields.pop('description') = fields.pop('status') = fields.pop('task_type', None)
if isinstance(, str): = or None
due_date = fields.pop('due_date', None)
if due_date:
# TODO: configure per-project timezone, or per-user timezone
due_date = parser.parse('%s 00:00:00 UTC' % due_date) = due_date
else: = None
users = fields.pop('users', None) = {'users': users or []}'Saving task %s', task.to_dict())
if fields:
self._log.warning('edit_task(%r, ...) called with unknown fields %r; ignoring them.',
task_id, fields)
return task
def delete_task(self, task_id, etag):
api = pillar_api()'Deleting task %s', task_id)
task = pillarsdk.Node({'_id': task_id, '_etag': etag})
def tasks_for_user(self, user_id):
"""Returns the tasks for the given user.
:returns: {'_items': [task, task, ...], '_meta': {Eve metadata}}
api = pillar_api()
# TODO: also include tasks assigned to any of the user's groups.
tasks = pillarsdk.Node.all({
'where': {
'properties.assigned_to.users': user_id,
'node_type': node_type_task['name'],
}, api=api)
return tasks
def tasks_for_project(self, project_id):
"""Returns the tasks for the given project.
:returns: {'_items': [task, task, ...], '_meta': {Eve metadata}}
api = pillar_api()
tasks = pillarsdk.Node.all({
'where': {
'project': project_id,
'node_type': node_type_task['name'],
}}, api=api)
return tasks
def api_task_for_shortcode(self, project_id: bson.ObjectId, shortcode: str) -> dict:
"""Returns the task for the given shortcode.
:returns: the task Node, or None if not found.
db = flask.current_app.db()
task = db['nodes'].find_one({
'properties.shortcode': shortcode,
'project': project_id,
'node_type': node_type_task['name'],
return task
def api_task_logged_in_svn(self, sender, shortcode, log_entry):
"""Blinker signal receiver; connects the logged commit with the task.
:param sender: sender of the signal
:type sender: attract_server.subversion.CommitLogObserver
:type log_entry: attract.subversion.LogEntry
""""Project %s, task '%s' logged in SVN by %s: %s...",
log_entry.project_id, shortcode,, log_entry.msg[:30].replace('\n', ' // '))
# Find the task
task = self.api_task_for_shortcode(log_entry.project_id, shortcode)
if not task:
self._log.warning('Task %s not found, ignoring SVN commit.', shortcode)
# Find the author
db = flask.current_app.db()
proj = db['projects'].find_one({'_id': task['project']},
projection={'extension_props.attract': 1})
if not proj:
self._log.warning('Project %s for task %s not found, ignoring SVN commit.',
task['project'], task['_id'])
# We have to have a user ID to register an activity, which is why we fall back
# to the current user (the SVNer service account) if there is no mapping.
usermap = proj['extension_props'].get('attract', {}).get('svn_usermap', {})
user_id = usermap.get(, None)
if user_id:
msg = 'committed SVN revision %s: %s' % (log_entry.revision, log_entry.msg)
self._log.warning('No Pillar user mapped for SVN user %s, using SVNer account.',
user_id = authentication.current_user_id()
msg = 'committed SVN revision %s authored by SVN user %s: %s' % (
log_entry.revision,, log_entry.msg)
user_id, msg,
'node', task['_id'],
'node', task.get('parent', None) or task['_id'],
def setup_app(app):
from . import eve_hooks

Event Timeline