Page MenuHome
No OneTemporary

File Metadata

Sun, Jan 26, 10:24 AM

import logging
import flask
from werkzeug.local import LocalProxy
from pillar.extension import PillarExtension
import pillar.web.subquery
from pillar.web.system_util import pillar_api
import pillarsdk
import attract.tasks
import attract.shots
EXTENSION_NAME = 'attract'
# Roles required to view task or shot details.
ROLES_REQUIRED_TO_VIEW_ITEMS = {u'demo', u'subscriber', u'admin'}
class AttractExtension(PillarExtension):
def __init__(self):
self._log = logging.getLogger('%s.AttractExtension' % __name__)
self.task_manager = attract.tasks.TaskManager()
self.shot_manager = attract.shots.ShotManager()
def name(self):
def flask_config(self):
"""Returns extension-specific defaults for the Flask configuration.
Use this to set sensible default values for configuration settings
introduced by the extension.
:rtype: dict
# Just so that it registers the management commands.
from . import cli
return {}
def eve_settings(self):
"""Returns extensions to the Eve settings.
Currently only the DOMAIN key is used to insert new resources into
Eve's configuration.
:rtype: dict
return {}
def blueprints(self):
"""Returns the list of top-level blueprints for the extension.
These blueprints will be mounted at the url prefix given to
:rtype: list of flask.Blueprint objects.
from . import routes
import attract.tasks.routes
import attract.shots.routes
return [
def template_path(self):
import os.path
return os.path.join(os.path.dirname(__file__), 'templates')
def static_path(self):
import os.path
return os.path.join(os.path.dirname(__file__), 'static')
def setup_app(self, app):
"""Connects Blinker signals and sets up other app-dependent stuff in submodules."""
from . import subversion, tasks, eve_hooks, shots
# Imports for side-effects
from . import node_url_finders
def attract_projects(self):
"""Returns projects set up for Attract.
:returns: {'_items': [proj, proj, ...], '_meta': Eve metadata}
import pillarsdk
from pillar.web.system_util import pillar_api
from .node_types.shot import node_type_shot
api = pillar_api()
# Find projects that are set up for Attract.
projects = pillarsdk.Project.all({
'where': {
'extension_props.attract': {'$exists': 1},
'': node_type_shot['name'],
}}, api=api)
return projects
def is_attract_project(self, project, test_extension_props=True):
"""Returns whether the project is set up for Attract.
Requires the task node type and Attract extension properties.
Testing the latter can be skipped with test_extension_props=False.
from .node_types.task import node_type_task
node_type_name = node_type_task['name']
node_type = project.get_node_type(node_type_name)
if not node_type:
return False
if not test_extension_props:
return True
pprops = project.extension_props[EXTENSION_NAME]
except AttributeError:
self._log.warning("is_attract_project: Project url=%r doesn't have any "
"extension properties.", project['url'])
if self._log.isEnabledFor(logging.DEBUG):
import pprint
self._log.debug('Project: %s', pprint.pformat(project.to_dict()))
return False
if pprops is None:
self._log.warning("is_attract_project: Project url=%r doesn't have Attract"
" extension properties.", project['url'])
return False
return True
def sidebar_links(self, project):
if not self.is_attract_project(project):
return ''
# Temporarily disabled until Attract is nicer to look at.
return ''
# return flask.render_template('attract/sidebar.html',
# project=project)
def activities_for_node(self, node_id, max_results=20, page=1):
"""Returns a page of activities for the given task or shot.
:returns: {'_items': [task, task, ...], '_meta': {Eve metadata}}
api = pillar_api()
activities = pillarsdk.Activity.all({
'where': {
'object_type': 'node',
'object': node_id,
'sort': [('_created', -1)],
'max_results': max_results,
'page': page,
}, api=api)
# Fetch more info for each activity.
for act in activities['_items']:
act.actor_user = pillar.web.subquery.get_user_info(act.actor_user)
return activities
def _get_current_attract():
"""Returns the Attract extension of the current application."""
return flask.current_app.pillar_extensions[EXTENSION_NAME]
current_attract = LocalProxy(_get_current_attract)
"""Attract extension of the current app."""

Event Timeline