/usr/lib/python3/dist-packages/doit/loader.py is in python3-doit 0.30.3-3.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 | """Loads dodo file (a python module) and convert them to 'tasks' """
import os
import sys
import inspect
import importlib
from collections import OrderedDict
from .exceptions import InvalidTask, InvalidCommand, InvalidDodoFile
from .task import DelayedLoader, Task, dict_to_task
# Directory path from where doit was executed.
# Set by loader, to be used on dodo.py by users.
initial_workdir = None
# TASK_STRING: (string) prefix used to identify python function
# that are task generators in a dodo file.
TASK_STRING = "task_"
def flat_generator(gen, gen_doc=''):
"""return only values from generators
if any generator yields another generator it is recursivelly called
"""
for item in gen:
if inspect.isgenerator(item):
item_doc = item.gi_code.co_consts[0]
for value, value_doc in flat_generator(item, item_doc):
yield value, value_doc
else:
yield item, gen_doc
def get_module(dodo_file, cwd=None, seek_parent=False):
"""
Find python module defining tasks, it is called "dodo" file.
@param dodo_file(str): path to file containing the tasks
@param cwd(str): path to be used cwd, if None use path from dodo_file
@param seek_parent(bool): search for dodo_file in parent paths if not found
@return (module) dodo module
"""
global initial_workdir
initial_workdir = os.getcwd()
def exist_or_raise(path):
"""raise exception if file on given path doesnt exist"""
if not os.path.exists(path):
msg = ("Could not find dodo file '%s'.\n" +
"Please use '-f' to specify file name.\n")
raise InvalidDodoFile(msg % path)
# get absolute path name
if os.path.isabs(dodo_file):
dodo_path = dodo_file
exist_or_raise(dodo_path)
else:
if not seek_parent:
dodo_path = os.path.abspath(dodo_file)
exist_or_raise(dodo_path)
else:
# try to find file in any folder above
current_dir = initial_workdir
dodo_path = os.path.join(current_dir, dodo_file)
file_name = os.path.basename(dodo_path)
parent = os.path.dirname(dodo_path)
while not os.path.exists(dodo_path):
new_parent = os.path.dirname(parent)
if new_parent == parent: # reached root path
exist_or_raise(dodo_file)
parent = new_parent
dodo_path = os.path.join(parent, file_name)
## load module dodo file and set environment
base_path, file_name = os.path.split(dodo_path)
# make sure dodo path is on sys.path so we can import it
sys.path.insert(0, base_path)
if cwd is None:
# by default cwd is same as dodo.py base path
full_cwd = base_path
else:
# insert specified cwd into sys.path
full_cwd = os.path.abspath(cwd)
if not os.path.isdir(full_cwd):
msg = "Specified 'dir' path must be a directory.\nGot '%s'(%s)."
raise InvalidCommand(msg % (cwd, full_cwd))
sys.path.insert(0, full_cwd)
# file specified on dodo file are relative to cwd
os.chdir(full_cwd)
# get module containing the tasks
return importlib.import_module(os.path.splitext(file_name)[0])
def create_after(executed=None, target_regex=None, creates=None):
"""Annotate a task-creator function with delayed loader info"""
def decorated(func):
func.doit_create_after = DelayedLoader(
func,
executed=executed,
target_regex=target_regex,
creates=creates
)
return func
return decorated
def load_tasks(namespace, command_names=(), allow_delayed=False):
"""Find task-creators and create tasks
@param namespace: (dict) containing the task creators, it might
contain other stuff
@param command_names: (list - str) blacklist for task names
@param load_all: (bool) if True ignore doit_crate_after['executed']
`load_all == False` is used by the runner to delay the creation of
tasks until a dependent task is executed. This is only used by the `run`
command, other commands should always load all tasks since it wont execute
any task.
@return task_list (list) of Tasks in the order they were defined on the file
"""
funcs = _get_task_creators(namespace, command_names)
# sort by the order functions were defined (line number)
# TODO: this ordering doesnt make sense when generators come
# from different modules
funcs.sort(key=lambda obj: obj[2])
task_list = []
def _process_gen():
task_list.extend(generate_tasks(name, ref(), ref.__doc__))
def _add_delayed(tname):
task_list.append(Task(tname, None, loader=delayed,
doc=delayed.creator.__doc__))
for name, ref, _ in funcs:
delayed = getattr(ref, 'doit_create_after', None)
if not delayed: # not a delayed task, just run creator
_process_gen()
elif delayed.creates: # delayed with explicit task basename
for tname in delayed.creates:
_add_delayed(tname)
elif allow_delayed: # delayed no explicit name, cmd run
_add_delayed(name)
else: # delayed no explicit name, cmd list (run creator)
_process_gen()
return task_list
def _get_task_creators(namespace, command_names):
"""get functions defined in the `namespace` and select the task-creators
A task-creator is a function that:
- name starts with the string TASK_STRING
- has the attribute `create_doit_tasks`
@return (list - func) task-creators
"""
funcs = []
prefix_len = len(TASK_STRING)
# get all functions that are task-creators
for name, ref in namespace.items():
# function is a task creator because of its name
if ((inspect.isfunction(ref) or inspect.ismethod(ref)) and
name.startswith(TASK_STRING)):
# remove TASK_STRING prefix from name
task_name = name[prefix_len:]
# object is a task creator because it contains the special method
elif hasattr(ref, 'create_doit_tasks'):
ref = ref.create_doit_tasks
# If create_doit_tasks is a method, it should be called only
# if it is bounded to an object.
# This avoids calling it for the class definition.
if inspect.signature(ref).parameters:
continue
task_name = name
# ignore functions that are not a task creator
else: # pragma: no cover
# coverage can't get "else: continue"
continue
# tasks can't have the same name of a commands
if task_name in command_names:
msg = ("Task can't be called '%s' because this is a command name."+
" Please choose another name.")
raise InvalidDodoFile(msg % task_name)
# get line number where function is defined
line = inspect.getsourcelines(ref)[1]
# add to list task generator functions
funcs.append((task_name, ref, line))
return funcs
def load_doit_config(dodo_module):
"""
@param dodo_module (dict) dict with module members
"""
doit_config = dodo_module.get('DOIT_CONFIG', {})
if not isinstance(doit_config, dict):
msg = ("DOIT_CONFIG must be a dict. got:'%s'%s")
raise InvalidDodoFile(msg % (repr(doit_config), type(doit_config)))
return doit_config
def _generate_task_from_return(func_name, task_dict, gen_doc):
"""generate a single task from a dict return'ed by a task generator"""
if 'name' in task_dict:
raise InvalidTask("Task '%s'. Only subtasks use field name." %
func_name)
task_dict['name'] = task_dict.pop('basename', func_name)
# Use task generator docstring
# if no doc present in task dict
if not 'doc' in task_dict:
task_dict['doc'] = gen_doc
return dict_to_task(task_dict)
def _generate_task_from_yield(tasks, func_name, task_dict, gen_doc):
"""generate a single task from a dict yield'ed by task generator
@param tasks: dictionary with created tasks
@return None: the created task is added to 'tasks' dict
"""
# check valid input
if not isinstance(task_dict, dict):
raise InvalidTask("Task '%s' must yield dictionaries" %
func_name)
msg_dup = "Task generation '%s' has duplicated definition of '%s'"
basename = task_dict.pop('basename', None)
# if has 'name' this is a sub-task
if 'name' in task_dict:
basename = basename or func_name
# if subname is None attributes from group task
if task_dict['name'] is None:
task_dict['name'] = basename
task_dict['actions'] = None
group_task = dict_to_task(task_dict)
group_task.has_subtask = True
tasks[basename] = group_task
return
# name is '<task>.<subtask>'
full_name = "%s:%s"% (basename, task_dict['name'])
if full_name in tasks:
raise InvalidTask(msg_dup % (func_name, full_name))
task_dict['name'] = full_name
sub_task = dict_to_task(task_dict)
sub_task.is_subtask = True
# get/create task group
group_task = tasks.get(basename)
if group_task:
if not group_task.has_subtask:
raise InvalidTask(msg_dup % (func_name, basename))
else:
group_task = Task(basename, None, doc=gen_doc, has_subtask=True)
tasks[basename] = group_task
group_task.task_dep.append(sub_task.name)
tasks[sub_task.name] = sub_task
# NOT a sub-task
else:
if not basename:
raise InvalidTask(
"Task '%s' must contain field 'name' or 'basename'. %s"%
(func_name, task_dict))
if basename in tasks:
raise InvalidTask(msg_dup % (func_name, basename))
task_dict['name'] = basename
# Use task generator docstring if no doc present in task dict
if not 'doc' in task_dict:
task_dict['doc'] = gen_doc
tasks[basename] = dict_to_task(task_dict)
def generate_tasks(func_name, gen_result, gen_doc=None):
"""Create tasks from a task generator result.
@param func_name: (string) name of taskgen function
@param gen_result: value returned by a task generator function
it can be a dict or generator (generating dicts)
@param gen_doc: (string/None) docstring from the task generator function
@return: (list - Task)
"""
# a task instance, just return it without any processing
if isinstance(gen_result, Task):
return (gen_result,)
# task described as a dictionary
if isinstance(gen_result, dict):
return [_generate_task_from_return(func_name, gen_result, gen_doc)]
# a generator
if inspect.isgenerator(gen_result):
tasks = OrderedDict() # task_name: task
# the generator return subtasks as dictionaries
for task_dict, x_doc in flat_generator(gen_result, gen_doc):
if isinstance(task_dict, Task):
tasks[task_dict.name] = task_dict
else:
_generate_task_from_yield(tasks, func_name, task_dict, x_doc)
if tasks:
return list(tasks.values())
else:
# special case task_generator did not generate any task
# create an empty group task
return [Task(func_name, None, doc=gen_doc, has_subtask=True)]
if gen_result is None:
return ()
raise InvalidTask(
"Task '%s'. Must return a dictionary or generator. Got %s" %
(func_name, type(gen_result)))
|