Page MenuHome

__init__.py
No OneTemporary

File Metadata

Created
Tue, Nov 19, 4:59 AM

__init__.py

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
from pillar.web.nodes.routes import url_for_node
import pillarsdk
import attract.auth
import attract.tasks
import attract.shots_and_assets
EXTENSION_NAME = 'attract'
class AttractExtension(PillarExtension):
has_project_settings = True
user_roles = {'org-attract'}
user_roles_indexable = {'org-attract'}
user_caps = {
'org-attract': {'attract-view', 'attract-use'},
'subscriber': {'attract-view', 'attract-use'},
'demo': {'attract-view', 'attract-use'},
'admin': {'attract-view', 'attract-use'},
}
def __init__(self):
self._log = logging.getLogger('%s.AttractExtension' % __name__)
self.task_manager = attract.tasks.TaskManager()
self.shot_manager = attract.shots_and_assets.ShotAssetManager()
self.auth = attract.auth.Auth()
@property
def name(self):
return EXTENSION_NAME
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
app.load_extension().
:rtype: list of flask.Blueprint objects.
"""
from . import routes
import attract.tasks.routes
import attract.shots_and_assets.routes_assets
import attract.shots_and_assets.routes_shots
import attract.subversion.routes
return [
routes.blueprint,
attract.tasks.routes.blueprint,
attract.tasks.routes.perproject_blueprint,
attract.shots_and_assets.routes_assets.perproject_blueprint,
attract.shots_and_assets.routes_shots.perproject_blueprint,
attract.subversion.routes.blueprint,
attract.subversion.routes.api_blueprint,
]
@property
def template_path(self):
import os.path
return os.path.join(os.path.dirname(__file__), 'templates')
@property
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 comments, subversion, tasks, eve_hooks, shots_and_assets
subversion.task_logged.connect(self.task_manager.api_task_logged_in_svn)
comments.setup_app(app)
tasks.setup_app(app)
shots_and_assets.setup_app(app)
eve_hooks.setup_app(app)
# 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_types.name': 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
try:
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
except KeyError:
# Not set up for Attract
return False
if pprops is None:
self._log.debug("is_attract_project: Project url=%r doesn't have Attract"
" extension properties.", project['url'])
return False
return True
def sidebar_links(self, project):
from pillar.api.utils import str2id
if not self.is_attract_project(project):
return ''
if not self.auth.current_user_may(auth.Actions.VIEW, str2id(project['_id'])):
return ''
return flask.render_template('attract/sidebar.html',
project=project)
@property
def has_project_settings(self) -> bool:
return self.auth.current_user_is_attract_user()
def project_settings(self, project: pillarsdk.Project, **template_args: dict) -> flask.Response:
"""Renders the project settings page for this extension.
Set YourExtension.has_project_settings = True and Pillar will call this function.
:param project: the project for which to render the settings.
:param template_args: additional template arguments.
:returns: a Flask HTTP response
"""
from attract.routes import project_settings
return project_settings(project, **template_args)
def activities_for_node(self, node_id, max_results=20, page=1):
"""Returns a page of activities for the given task or shot.
Activities that are either on this node or have this node as context
are returned.
:returns: {'_items': [task, task, ...], '_meta': {Eve metadata}}
"""
api = pillar_api()
activities = pillarsdk.Activity.all({
'where': {
'$or': [
{'object_type': 'node',
'object': node_id},
{'context_object_type': 'node',
'context_object': node_id},
],
},
'sort': [('_id', 1)], # Sort by creation, _id is incremental.
'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 link_for_activity(self, act):
"""Returns the URL for the activity.
:type act: pillarsdk.Activity
"""
from .node_types.task import node_type_task
from .node_types.shot import node_type_shot
if act.node_type == node_type_task['name']:
if act.context_object:
return flask.url_for('attract.shots.perproject.with_task',
project_url=act.project.url,
task_id=act.object)
return flask.url_for('attract.tasks.perproject.view_task',
project_url=act.project.url,
task_id=act.object)
elif act.node_type == node_type_shot['name']:
return flask.url_for('attract.shots.perproject.view_shot',
project_url=act.project.url,
shot_id=act.object)
return url_for_node(node_id=act.object)
def _get_current_attract() -> AttractExtension:
"""Returns the Attract extension of the current application."""
return flask.current_app.pillar_extensions[EXTENSION_NAME]
current_attract: AttractExtension = LocalProxy(_get_current_attract)
"""Attract extension of the current app."""

Event Timeline