Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
the shell is not terrifically expensive, but must be done at a specific
point in time to ensure the "only known config keys are loaded from the
env" behavior works correctly.
See :ref:`env-vars` for details on this design decision and other info
re: how environment variables are scanned and loaded.
.. versionadded:: 1.0
"""
# Force merge of existing data to ensure we have an up to date picture
debug("Running pre-merge for shell env loading...")
self.merge()
debug("Done with pre-merge.")
loader = Environment(config=self._config, prefix=self._env_prefix)
self._set(_env=loader.load())
debug("Loaded shell environment, triggering final merge")
self.merge()
def no_tasks_given(self):
debug(
"No tasks specified for execution and no default task; printing global help as fallback" # noqa
)
self.print_help()
raise Exit
def parse_core(self, argv):
debug("argv given to Program.run: {!r}".format(argv))
self.normalize_argv(argv)
# Obtain core args (sets self.core)
self.parse_core_args()
debug("Finished parsing core args")
# Set interpreter bytecode-writing flag
sys.dont_write_bytecode = not self.args["write-pyc"].value
# Enable debugging from here on out, if debug flag was given.
# (Prior to this point, debugging requires setting INVOKE_DEBUG).
if self.args.debug.value:
enable_logging()
# Short-circuit if --version
if self.args.version.value:
debug("Saw --version, printing version & exiting")
self.print_version()
raise Exit
# Print (dynamic, no tasks required) completion script if requested
except AttributeError:
msg = "Config files of type {!r} (from file {!r}) are not supported! Please use one of: {!r}" # noqa
raise UnknownFileType(
msg.format(type_, filepath, self._file_suffixes)
)
# Store data, the path it was found at, and fact that it was
# found
self._set(data, loader(filepath))
self._set(path, filepath)
self._set(found, True)
break
# Typically means 'no such file', so just note & skip past.
except IOError as e:
if e.errno == 2:
err = "Didn't see any {}, skipping."
debug(err.format(filepath))
else:
raise
# Still None -> no suffixed paths were found, record this fact
if getattr(self, path) is None:
self._set(found, False)
# Merge loaded data in if any was found
elif merge:
self.merge()
contexts = parser.parse_argv(tokens)
except ParseError as e:
debug("Got parser error ({0!r}), grabbing its last-seen context {1!r}".format(e, e.context)) # noqa
contexts = [e.context]
# Fall back to core context if no context seen.
debug("Parsed invocation, contexts: {0!r}".format(contexts))
if not contexts or not contexts[-1]:
context = initial_context
else:
context = contexts[-1]
debug("Selected context: {0!r}".format(context))
# Unknown flags (could be e.g. only partially typed out; could be
# wholly invalid; doesn't matter) complete with flags.
debug("Looking for {0!r} in {1!r}".format(tail, context.flags))
if tail not in context.flags:
debug("Not found, completing with flag names")
# Long flags - partial or just the dashes - complete w/ long flags
if tail.startswith('--'):
for name in filter(
lambda x: x.startswith('--'),
context.flag_names()
):
print(name)
# Just a dash, completes with all flags
elif tail == '-':
for name in context.flag_names():
print(name)
# Otherwise, it's something entirely invalid (a shortflag not
# recognized, or a java style flag like -foo) so return nothing
# (the shell will still try completing with files, but that doesn't
# hurt really.)
else:
def parse_core_args(self):
"""
Filter out core args, leaving any tasks or their args for later.
Sets ``self.core`` to the `.ParseResult` from this step.
.. versionadded:: 1.0
"""
debug("Parsing initial context (core args)")
parser = Parser(initial=self.initial_context, ignore_unknown=True)
self.core = parser.parse_argv(self.argv[1:])
msg = "Core-args parse result: {!r} & unparsed: {!r}"
debug(msg.format(self.core, self.core.unparsed))
# recognized, or a java style flag like -foo) so return nothing
# (the shell will still try completing with files, but that doesn't
# hurt really.)
else:
pass
# Known flags complete w/ nothing or tasks, depending
else:
# Flags expecting values: do nothing, to let default (usually
# file) shell completion occur (which we actively want in this
# case.)
if context.flags[tail].takes_value:
debug("Found, and it takes a value, so no completion")
pass
# Not taking values (eg bools): print task names
else:
debug("Found, takes no value, printing task names")
print_task_names(collection)
# If not a flag, is either task name or a flag value, so just complete
# task names.
else:
debug("Last token isn't flag-like, just printing task names")
print_task_names(collection)
raise Exit
def parse_cleanup(self):
"""
Post-parsing, pre-execution steps such as --help, --list, etc.
.. versionadded:: 1.0
"""
halp = self.args.help.value
# Core (no value given) --help output (only when bundled namespace)
if halp is True:
debug("Saw bare --help, printing help & exiting")
self.print_help()
raise Exit
# Print per-task help, if necessary
if halp:
if halp in self.parser.contexts:
msg = "Saw --help , printing per-task help & exiting"
debug(msg)
self.print_task_help(halp)
raise Exit
else:
# TODO: feels real dumb to factor this out of Parser, but...we
# should?
raise ParseError("No idea what '{}' is!".format(halp))
# Print discovered tasks if necessary
# the tree we expect. (This is a concession to testing.)
try:
dedupe = self.config.tasks.dedupe
except AttributeError:
dedupe = True
# Dedupe across entire run now that we know about all calls in order
calls = self.dedupe(expanded) if dedupe else expanded
# Execute
results = {}
# TODO: maybe clone initial config here? Probably not necessary,
# especially given Executor is not designed to execute() >1 time at the
# moment...
for call in calls:
autoprint = call in direct and call.autoprint
args = call.args
debug("Executing {!r}".format(call))
# Hand in reference to our config, which will preserve user
# modifications across the lifetime of the session.
config = self.config
# But make sure we reset its task-sensitive levels each time
# (collection & shell env)
# TODO: load_collection needs to be skipped if task is anonymous
# (Fabric 2 or other subclassing libs only)
collection_config = self.collection.configuration(call.called_as)
config.load_collection(collection_config)
config.load_shell_env()
debug("Finished loading collection & shell env configs")
# Get final context from the Call (which will know how to generate
# an appropriate one; e.g. subclasses might use extra data from
# being parameterized), handing in this config for use there.
context = call.make_context(config)
args = (context,) + args
# Obtain core args (sets self.core)
self.parse_core_args()
debug("Finished parsing core args")
# Set interpreter bytecode-writing flag
sys.dont_write_bytecode = not self.args["write-pyc"].value
# Enable debugging from here on out, if debug flag was given.
# (Prior to this point, debugging requires setting INVOKE_DEBUG).
if self.args.debug.value:
enable_logging()
# Short-circuit if --version
if self.args.version.value:
debug("Saw --version, printing version & exiting")
self.print_version()
raise Exit
# Print (dynamic, no tasks required) completion script if requested
if self.args["print-completion-script"].value:
print_completion_script(
shell=self.args["print-completion-script"].value,
names=self.binary_names,
)
raise Exit