Source code for plainbox.impl.job
# This file is part of Checkbox.
#
# Copyright 2012, 2013 Canonical Ltd.
# Written by:
# Zygmunt Krynicki <zygmunt.krynicki@canonical.com>
#
# Checkbox is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 3,
# as published by the Free Software Foundation.
#
# Checkbox is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Checkbox. If not, see <http://www.gnu.org/licenses/>.
"""
:mod:`plainbox.impl.job` -- job definition
==========================================
.. warning::
THIS MODULE DOES NOT HAVE STABLE PUBLIC API
"""
import logging
from plainbox.impl.unit.job import JobDefinition
__all__ = ['JobDefinition', 'JobTreeNode']
logger = logging.getLogger("plainbox.job")
[docs]class JobTreeNode:
"""
JobTreeNode class is used to store a tree structure. A tree consists of
a collection of JobTreeNode instances connected in a hierarchical way
where nodes are used as categories, jobs belonging to a category are
listed in the node leaves.
**Example:**
::
/ Job A
Root-|
| / Job B
\--- Category X |
\ Job C
"""
def __init__(self, name=None):
self._name = name if name else 'Root'
self._parent = None
self._categories = []
self._jobs = []
@property
[docs] def name(self):
"""
node name
"""
return self._name
@property
[docs] def parent(self):
"""
parent node for this node
"""
return self._parent
@property
[docs] def categories(self):
"""
list of sub categories
"""
return self._categories
@property
[docs] def jobs(self):
"""
job(s) belonging to this node/category
"""
return self._jobs
@property
[docs] def depth(self):
"""
level of depth for this node
"""
return (self._parent.depth + 1) if self._parent else 0
def __str__(self):
return self.name
def __repr__(self):
return "<JobTreeNode name:{!r}>".format(self.name)
[docs] def add_category(self, category):
"""
Adds a new category to this node.
:argument category: the node instance to be added as a category.
"""
self._categories.append(category)
# Always keep this list sorted to easily find a given child by index
self._categories.sort(key=lambda item: item.name)
category._parent = self
[docs] def add_job(self, job):
"""
Adds a new job to this node.
:argument job: the job instance to be added to this node.
"""
self._jobs.append(job)
# Always keep this list sorted to easily find a given leaf by index
# Note bisect.insort(a, x) cannot be used here as JobDefinition are
# not sortable
self._jobs.sort(key=lambda item: item.id)
[docs] def get_ancestors(self):
"""
Returns the list of all ancestor nodes from current node to the
current tree root.
"""
ancestors = []
node = self
while node.parent is not None:
ancestors.append(node.parent)
node = node.parent
return ancestors
[docs] def get_descendants(self):
"""
Returns a list of all descendant category nodes.
"""
descendants = []
for category in self.categories:
descendants.append(category)
descendants.extend(category.get_descendants())
return descendants
@classmethod
[docs] def create_tree(cls, session_state, job_list):
"""
Build a rooted JobTreeNode from a job list
:argument session_state:
A session state object
:argument job_list:
List of jobs to consider for building the tree.
"""
builder = TreeBuilder(session_state, cls)
for job in job_list:
builder.auto_add_job(job)
return builder.root_node
class TreeBuilder:
"""
Helper class that assists in building a tree of :class:`JobTreeNode`
objects out of job definitions and their associations, as expressed by
:attr:`JobState.via_job` associated with each job.
The builder is a single-use object and should be re-created for each new
construct. Internally it stores the job_state_map of the
:class:`SessionState` it was created with as well as additional helper
state.
"""
def __init__(self, session_state: "SessionState", node_cls):
self._job_state_map = session_state.job_state_map
self._node_cls = node_cls
self._root_node = node_cls()
self._category_node_map = {} # id -> node
@property
def root_node(self):
return self._root_node
def auto_add_job(self, job):
"""
Add a job to the tree, automatically creating category nodes as needed.
:param job:
The job definition to add.
"""
if job.plugin == 'local':
# For local jobs, just create the category node but don't add the
# local job itself there.
self.get_or_create_category_node(job)
else:
# For all other jobs, look at the parent job (if any) and create
# the category node out of that node. This never fails as "None" is
# the root_node object.
state = self._job_state_map[job.id]
node = self.get_or_create_category_node(state.via_job)
# Then add that job to the category node
node.add_job(job)
def get_or_create_category_node(self, category_job):
"""
Get or create a :class:`JobTreeNode` that corresponds to the
category defined (somehow) by the job ``category_job``.
:param category_job:
The job that describes the category. This is either a
plugin="local" job or a plugin="resource" job. This can also be
None, which is a shorthand to say "root node".
:returns:
The ``root_node`` if ``category_job`` is None. A freshly
created node, created with :func:`create_category_node()` if
the category_job was never seen before (as recorded by the
category_node_map).
"""
if category_job is None:
return self._root_node
if category_job.id not in self._category_node_map:
category_node = self.create_category_node(category_job)
# The category is added to its parent, that's either the root
# (if we're standalone) or the non-root category this one
# belongs to.
category_state = self._job_state_map[category_job.id]
if category_state.via_job is not None:
parent_category_node = self.get_or_create_category_node(
category_state.via_job)
else:
parent_category_node = self._root_node
parent_category_node.add_category(category_node)
self._category_node_map[category_job.id] = category_node
else:
category_node = self._category_node_map[category_job.id]
return category_node
def create_category_node(self, category_job):
"""
Create a :class:`JobTreeNode` that corresponds to the category defined
(somehow) by the job ``category_job``.
:param category_job:
The job that describes the node to create.
:returns:
A fresh node with appropriate data.
"""
if category_job.summary == category_job.partial_id:
return self._node_cls(category_job.description)
else:
return self._node_cls(category_job.summary)