enum34 be gone

doc/parameters.rst

@@ -88,6 +88,26 @@ are not the same instance:
  >>> hash(c) == hash(d)
  True
 
++Parameter visibility
+^^^^^^^^^^^^^^^^^^^^
+
+Using :class:`~luigi.parameter.ParameterVisibility` you can configure parameter visibility. By default, all
+parameters are public, but you can also set them hidden or private.
+
+.. code:: python
+
+ >>> import luigi
+ >>> from luigi.parameter import ParameterVisibility
+
+ >>> luigi.Parameter(visibility=ParameterVisibility.PRIVATE)
+
+``ParameterVisibility.PUBLIC`` (default) - visible everywhere
+
+``ParameterVisibility.HIDDEN`` - ignored in WEB-view, but saved into database if save db_history is true
+
+``ParameterVisibility.PRIVATE`` - visible only inside task.
+
+
 Parameter types
 ^^^^^^^^^^^^^^^
 

luigi/__init__.py

@@ -43,6 +43,7 @@
 
 from luigi import interface
 from luigi.interface import run, build
+from luigi.execution_summary import LuigiStatusCode
 
 from luigi import event
 from luigi.event import Event
@@ -59,5 +60,5 @@
  'FloatParameter', 'BoolParameter', 'TaskParameter',
  'ListParameter', 'TupleParameter', 'EnumParameter', 'DictParameter',
  'configuration', 'interface', 'local_target', 'run', 'build', 'event', 'Event',
- 'NumericalParameter', 'ChoiceParameter', 'OptionalParameter'
+ 'NumericalParameter', 'ChoiceParameter', 'OptionalParameter', 'LuigiStatusCode'
 ]

luigi/configuration.py

@@ -35,11 +35,17 @@
 
 try:
  from ConfigParser import ConfigParser, NoOptionError, NoSectionError
+ Interpolation = object
 except ImportError:
  from configparser import ConfigParser, NoOptionError, NoSectionError
+ from configparser import Interpolation
 
 
 class LuigiConfigParser(ConfigParser):
+
+ # for python2/3 compatibility
+ _DEFAULT_INTERPOLATION = Interpolation()
+
  NO_DEFAULT = object()
  _instance = None
  _config_paths = [

luigi/execution_summary.py

@@ -24,6 +24,7 @@
 import textwrap
 import collections
 import functools
+import enum
 
 import luigi
 
@@ -32,6 +33,63 @@ class execution_summary(luigi.Config):
  summary_length = luigi.IntParameter(default=5)
 
 
+class LuigiStatusCode(enum.Enum):
+ """
+ All possible status codes for the attribute ``status`` in :class:`~luigi.execution_summary.LuigiRunResult` when
+ the argument ``detailed_summary=True`` in *luigi.run() / luigi.build*.
+ Here are the codes and what they mean:
+
+ ============================= ==========================================================
+ Status Code Name Meaning
+ ============================= ==========================================================
+ SUCCESS There were no failed tasks or missing dependencies
+ SUCCESS_WITH_RETRY There were failed tasks but they all succeeded in a retry
+ FAILED There were failed tasks
+ FAILED_AND_SCHEDULING_FAILED There were failed tasks and tasks whose scheduling failed
+ SCHEDULING_FAILED There were tasks whose scheduling failed
+ NOT_RUN There were tasks that were not granted run permission by the scheduler
+ MISSING_EXT There were missing external dependencies
+ ============================= ==========================================================
+
+ """
+ SUCCESS = (":)", "there were no failed tasks or missing dependencies")
+ SUCCESS_WITH_RETRY = (":)", "there were failed tasks but they all succeeded in a retry")
+ FAILED = (":(", "there were failed tasks")
+ FAILED_AND_SCHEDULING_FAILED = (":(", "there were failed tasks and tasks whose scheduling failed")
+ SCHEDULING_FAILED = (":(", "there were tasks whose scheduling failed")
+ NOT_RUN = (":|", "there were tasks that were not granted run permission by the scheduler")
+ MISSING_EXT = (":|", "there were missing external dependencies")
+
+
+class LuigiRunResult(object):
+ """
+ The result of a call to build/run when passing the detailed_summary=True argument.
+
+ Attributes:
+ - one_line_summary (str): One line summary of the progress.
+ - summary_text (str): Detailed summary of the progress.
+ - status (LuigiStatusCode): Luigi Status Code. See :class:`~luigi.execution_summary.LuigiStatusCode` for what these codes mean.
+ - status_code_num (int): Numeric representation for status (LuigiStatusCode)
+ - worker (luigi.worker.worker): Worker object. See :class:`~luigi.worker.worker`.
+ - scheduling_succeeded (bool): Boolean which is *True* if all the tasks were scheduled without errors.
+
+ """
+ def __init__(self, worker, worker_add_run_status=True):
+ self.worker = worker
+ summary_dict = _summary_dict(worker)
+ self.summary_text = _summary_wrap(_summary_format(summary_dict, worker))
+ self.status = _tasks_status(summary_dict)
+ self.status_code_num = _status_to_code_num(self.status)
+ self.one_line_summary = _create_one_line_summary(self.status)
+ self.scheduling_succeeded = worker_add_run_status
+
+ def __str__(self):
+ return "LuigiRunResult with status {0}".format(self.status)
+
+ def __repr__(self):
+ return "LuigiRunResult(status={0!r},worker={1!r},scheduling_succeeded={2!r})".format(self.status, self.worker, self.scheduling_succeeded)
+
+
 def _partition_tasks(worker):
  """
  Takes a worker and sorts out tasks based on their status.
@@ -377,33 +435,60 @@ def _summary_format(set_tasks, worker):
  if len(ext_workers) == 0:
  str_output += '\n'
  str_output += 'Did not run any tasks'
- smiley = ""
- reason = ""
+ one_line_summary = _create_one_line_summary(_tasks_status(set_tasks))
+ str_output += "\n{0}".format(one_line_summary)
+ if num_all_tasks == 0:
+ str_output = 'Did not schedule any tasks'
+ return str_output
+
+
+def _create_one_line_summary(status_code):
+ """
+ Given a status_code of type LuigiStatusCode which has a tuple value, returns a one line summary
+ """
+ return "This progress looks {0} because {1}".format(*status_code.value)
+
+
+def _tasks_status(set_tasks):
+ """
+ Given a grouped set of tasks, returns a LuigiStatusCode
+ """
  if set_tasks["ever_failed"]:
  if not set_tasks["failed"]:
- smiley = ":)"
- reason = "there were failed tasks but they all succeeded in a retry"
+ return LuigiStatusCode.SUCCESS_WITH_RETRY
  else:
- smiley = ":("
- reason = "there were failed tasks"
  if set_tasks["scheduling_error"]:
- reason += " and tasks whose scheduling failed"
+ return LuigiStatusCode.FAILED_AND_SCHEDULING_FAILED
+ return LuigiStatusCode.FAILED
  elif set_tasks["scheduling_error"]:
- smiley = ":("
- reason = "there were tasks whose scheduling failed"
+ return LuigiStatusCode.SCHEDULING_FAILED
  elif set_tasks["not_run"]:
- smiley = ":|"
- reason = "there were tasks that were not granted run permission by the scheduler"
+ return LuigiStatusCode.NOT_RUN
  elif set_tasks["still_pending_ext"]:
- smiley = ":|"
- reason = "there were missing external dependencies"
+ return LuigiStatusCode.MISSING_EXT
  else:
- smiley = ":)"
- reason = "there were no failed tasks or missing external dependencies"
- str_output += "\nThis progress looks {0} because {1}".format(smiley, reason)
- if num_all_tasks == 0:
- str_output = 'Did not schedule any tasks'
- return str_output
+ return LuigiStatusCode.SUCCESS
+
+
+def _status_to_code_num(status_code):
+ """
+ Given a status_code of type LuigiStatusCode, returns a numeric value representing it
+ POSIX assigns special meanings to 1 and 2 so start from 3
+ """
+ if status_code == LuigiStatusCode.SUCCESS:
+ return 0
+ elif status_code == LuigiStatusCode.SUCCESS_WITH_RETRY:
+ return 3
+ elif status_code == LuigiStatusCode.FAILED:
+ return 4
+ elif status_code == LuigiStatusCode.FAILED_AND_SCHEDULING_FAILED:
+ return 5
+ elif status_code == LuigiStatusCode.SCHEDULING_FAILED:
+ return 6
+ elif status_code == LuigiStatusCode.NOT_RUN:
+ return 7
+ elif status_code == LuigiStatusCode.MISSING_EXT:
+ return 8
 
 
 def _summary_wrap(str_output):

luigi/interface.py

@@ -36,7 +36,7 @@
 from luigi import scheduler
 from luigi import task
 from luigi import worker
-from luigi import execution_summary
+from luigi.execution_summary import LuigiRunResult
 from luigi.cmdline_parser import CmdlineParser
 
 
@@ -205,8 +205,9 @@ def _schedule_and_run(tasks, worker_scheduler_factory=None, override_defaults=No
  success &= worker.add(t, env_params.parallel_scheduling, env_params.parallel_scheduling_processes)
  logger.info('Done scheduling tasks')
  success &= worker.run()
- logger.info(execution_summary.summary(worker))
- return dict(success=success, worker=worker)
+ luigi_run_result = LuigiRunResult(worker, success)
+ logger.info(luigi_run_result.summary_text)
+ return luigi_run_result
 
 
 class PidLockAlreadyTakenExit(SystemExit):
@@ -217,11 +218,16 @@ class PidLockAlreadyTakenExit(SystemExit):
 
 
 def run(*args, **kwargs):
- return _run(*args, **kwargs)['success']
+ luigi_run_result = _run(*args, **kwargs)
+ if kwargs.get('detailed_summary'):
+ # return status code instead of entire class to keep interface similar (used to return bool)
+ return luigi_run_result.status_code_num
+ else:
+ return luigi_run_result.scheduling_succeeded
 
 
 def _run(cmdline_args=None, main_task_cls=None,
- worker_scheduler_factory=None, use_dynamic_argparse=None, local_scheduler=False):
+ worker_scheduler_factory=None, use_dynamic_argparse=None, local_scheduler=False, detailed_summary=False):
  """
  Please dont use. Instead use `luigi` binary.
 
@@ -248,7 +254,7 @@ def _run(cmdline_args=None, main_task_cls=None,
  return _schedule_and_run([cp.get_task_obj()], worker_scheduler_factory)
 
 
-def build(tasks, worker_scheduler_factory=None, **env_params):
+def build(tasks, worker_scheduler_factory=None, detailed_summary=False, **env_params):
  """
  Run internally, bypassing the cmdline parsing.
 
@@ -271,4 +277,10 @@ def build(tasks, worker_scheduler_factory=None, **env_params):
  if "no_lock" not in env_params:
  env_params["no_lock"] = True
 
- return _schedule_and_run(tasks, worker_scheduler_factory, override_defaults=env_params)['success']
+ luigi_run_result = _schedule_and_run(tasks, worker_scheduler_factory,
+ override_defaults=env_params)
+ if detailed_summary:
+ # return status code instead of entire class to keep interface similar (used to return bool)
+ return luigi_run_result.status_code_num
+ else:
+ return luigi_run_result.scheduling_succeeded

luigi/notifications.py

@@ -88,6 +88,10 @@ class email(luigi.Config):
  default=DEFAULT_CLIENT_EMAIL,
  config_path=dict(section='core', name='email-sender'),
  description='Address to send e-mails from')
+ region = luigi.parameter.Parameter(
+ default='',
+ config_path=dict(section='email', name='region'),
+ description='AWS region for SES if you want to override the default AWS region for boto3')
 
 
 class smtp(luigi.Config):
@@ -219,7 +223,8 @@ def send_email_ses(sender, subject, message, recipients, image_png):
  """
  from boto3 import client as boto3_client
 
- client = boto3_client('ses')
+ region = email().region or None
+ client = boto3_client('ses', region_name=region)
 
  msg_root = generate_email(sender, subject, message, recipients, image_png)
  response = client.send_raw_email(Source=sender,

luigi/parameter.py

@@ -23,6 +23,7 @@
 import abc
 import datetime
 import warnings
+from enum import IntEnum
 import json
 from json import JSONEncoder
 from collections import OrderedDict, Mapping
@@ -44,6 +45,23 @@
 _no_value = object()
 
 
+class ParameterVisibility(IntEnum):
+ """
+ Possible values for the parameter visibility option. Public is the default.
+ See :doc:`/parameters` for more info.
+ """
+ PUBLIC = 0
+ HIDDEN = 1
+ PRIVATE = 2
+
+ @classmethod
+ def has_value(cls, value):
+ return any(value == item.value for item in cls)
+
+ def serialize(self):
+ return self.value
+
+
 class ParameterException(Exception):
  """
  Base exception.
@@ -113,7 +131,8 @@ def run(self):
  _counter = 0 # non-atomically increasing counter used for ordering parameters.
 
  def __init__(self, default=_no_value, is_global=False, significant=True, description=None,
- config_path=None, positional=True, always_in_help=False, batch_method=None):
+ config_path=None, positional=True, always_in_help=False, batch_method=None,
+ visibility=ParameterVisibility.PUBLIC):
  """
  :param default: the default value for this parameter. This should match the type of the
  Parameter, i.e. ``datetime.date`` for ``DateParameter`` or ``int`` for
@@ -140,6 +159,10 @@ def __init__(self, default=_no_value, is_global=False, significant=True, descrip
  parameter values into a single value. Used
  when receiving batched parameter lists from
  the scheduler. See :ref:`batch_method`
+
+ :param visibility: A Parameter whose value is a :py:class:`~luigi.parameter.ParameterVisibility`.
+ Default value is ParameterVisibility.PUBLIC
+
  """
  self._default = default
  self._batch_method = batch_method
@@ -150,6 +173,7 @@ def __init__(self, default=_no_value, is_global=False, significant=True, descrip
  positional = False
  self.significant = significant # Whether different values for this parameter will differentiate otherwise equal tasks
  self.positional = positional
+ self.visibility = visibility if ParameterVisibility.has_value(visibility) else ParameterVisibility.PUBLIC
 
  self.description = description
  self.always_in_help = always_in_help
@@ -195,11 +219,11 @@ def _value_iterator(self, task_name, param_name):
  yield (self._get_value_from_config(task_name, param_name), None)
  yield (self._get_value_from_config(task_name, param_name.replace('_', '-')),
  'Configuration [{}] {} (with dashes) should be avoided. Please use underscores.'.format(
- task_name, param_name))
+  task_name, param_name))
  if self._config_path:
  yield (self._get_value_from_config(self._config_path['section'], self._config_path['name']),
  'The use of the configuration [{}] {} is deprecated. Please use [{}] {}'.format(
- self._config_path['section'], self._config_path['name'], task_name, param_name))
+  self._config_path['section'], self._config_path['name'], task_name, param_name))
  yield (self._default, None)
 
  def has_task_value(self, task_name, param_name):
@@ -694,7 +718,8 @@ def field(key):
  def optional_field(key):
  return "(%s)?" % field(key)
  # A little loose: ISO 8601 does not allow weeks in combination with other fields, but this regex does (as does python timedelta)
- regex = "P(%s|%s(T%s)?)" % (field("weeks"), optional_field("days"), "".join([optional_field(key) for key in ["hours", "minutes", "seconds"]]))
+ regex = "P(%s|%s(T%s)?)" % (field("weeks"), optional_field("days"),
+ "".join([optional_field(key) for key in ["hours", "minutes", "seconds"]]))
  return self._apply_regex(regex, input)
 
  def _parseSimple(self, input):