#
# Copyright (C) 2018 Codethink Limited
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2 of the License, or (at your option) any later version.
#
# This library 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
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library. If not, see .
#
# Authors:
# Tristan Van Berkom
import os
from contextlib import suppress
from .._exceptions import LoadError
from ..exceptions import LoadErrorReason
from .. import _yaml
from ..element import Element
from ..node import Node
from .._profile import Topics, PROFILER
from .._includes import Includes
from ._loader import valid_chars_name
from .types import Symbol
from . import loadelement
from .loadelement import LoadElement, Dependency, DependencyType, extract_depends_from_node
from ..types import CoreWarnings, _KeyStrength
# Loader():
#
# The Loader class does the heavy lifting of parsing target
# bst files and ultimately transforming them into a list of LoadElements
# ready for instantiation by the core.
#
# Args:
# project (Project): The toplevel Project object
# parent (Loader): A parent Loader object, in the case this is a junctioned Loader
# provenance_node (Node): The provenance of the reference to this project's junction
#
class Loader:
def __init__(self, project, *, parent=None, provenance_node=None):
# Ensure we have an absolute path for the base directory
basedir = project.element_path
if not os.path.isabs(basedir):
basedir = os.path.abspath(basedir)
#
# Public members
#
self.load_context = project.load_context # The LoadContext
self.project = project # The associated Project
self.provenance_node = provenance_node # The provenance of whence this loader was instantiated
self.loaded = None # The number of loaded Elements
#
# Private members
#
self._options = project.options # Project options (OptionPool)
self._basedir = basedir # Base project directory
self._first_pass_options = project.first_pass_config.options # Project options (OptionPool)
self._parent = parent # The parent loader
self._alternative_parents = [] # Overridden parent loaders
self._meta_elements = {} # Dict of resolved meta elements by name
self._elements = {} # Dict of elements
self._links = {} # Dict of link target target paths indexed by link element paths
self._loaders = {} # Dict of junction loaders
self._loader_search_provenances = {} # Dictionary of provenance nodes of ongoing child loader searches
self._includes = Includes(self, copy_tree=True)
assert project.name is not None
self.load_context.register_loader(self)
# The __str__ of a Loader is used to clearly identify the Loader,
# the junction is was loaded as, and the provenance causing the
# junction to be loaded.
#
def __str__(self):
project_name = self.project.name
if self.project.junction:
junction_name = self.project.junction._get_full_name()
if self.provenance_node:
provenance = "({}): {}".format(junction_name, self.provenance_node.get_provenance())
else:
provenance = "({})".format(junction_name)
else:
provenance = "(toplevel)"
return "{} {}".format(project_name, provenance)
# load():
#
# Loads the project based on the parameters given to the constructor
#
# Args:
# targets (list of str): Target, element-path relative bst filenames in the project
#
# Raises: LoadError
#
# Returns:
# (list): The corresponding LoadElement instances matching the `targets`
#
def load(self, targets):
for filename in targets:
if os.path.isabs(filename):
# XXX Should this just be an assertion ?
# Expect that the caller gives us the right thing at least ?
raise LoadError(
"Target '{}' was not specified as a relative "
"path to the base project directory: {}".format(filename, self._basedir),
LoadErrorReason.INVALID_DATA,
)
self._warn_invalid_elements(targets)
# First pass, recursively load files and populate our table of LoadElements
#
target_elements = []
for target in targets:
with PROFILER.profile(Topics.LOAD_PROJECT, target):
_junction, name, loader = self._parse_name(target, None)
element = loader._load_file(name, None)
target_elements.append(element)
#
# Now that we've resolved the dependencies, scan them for circular dependencies
#
# Set up a dummy element that depends on all top-level targets
# to resolve potential circular dependencies between them
dummy_target = LoadElement(Node.from_dict({}), "", self)
# Pylint is not very happy with Cython and can't understand 'dependencies' is a list
dummy_target.dependencies.extend( # pylint: disable=no-member
Dependency(element, DependencyType.RUNTIME) for element in target_elements
)
with PROFILER.profile(Topics.CIRCULAR_CHECK, "_".join(targets)):
self._check_circular_deps(dummy_target)
#
# Sort direct dependencies of elements by their dependency ordering
#
# Keep a list of all visited elements, to not sort twice the same
visited_elements = set()
for element in target_elements:
loader = element._loader
with PROFILER.profile(Topics.SORT_DEPENDENCIES, element.name):
loadelement.sort_dependencies(element, visited_elements)
self._clean_caches()
# Cache how many Elements have just been loaded
if self.load_context.task:
self.loaded = self.load_context.task.current_progress
return target_elements
# get_loader():
#
# Obtains the appropriate loader for the specified junction
#
# If `load_subprojects` is enabled, then this function will
# either return the desired loader or raise a LoadError. If
# `load_subprojects` is disabled, then it can also return None
# in the case that a loader could not be found. In either case,
# a non-existant file in a loaded project will result in a LoadError.
#
# Args:
# name (str): Name of junction, may have multiple `:` in the name
# provenance_node (Node): The provenance from where this loader was requested
# load_subprojects (bool): Whether to load subprojects on demand
#
# Returns:
# (Loader): loader for sub-project
#
def get_loader(self, name, provenance_node, *, load_subprojects=True):
junction_path = name.split(":")
loader = self
circular_provenance_node = self._loader_search_provenances.get(name, None)
if circular_provenance_node and load_subprojects:
assert provenance_node
detail = None
if circular_provenance_node is not provenance_node:
detail = "Already searching for '{}' at: {}".format(name, circular_provenance_node.get_provenance())
raise LoadError(
"{}: Circular reference while searching for '{}'".format(provenance_node.get_provenance(), name),
LoadErrorReason.CIRCULAR_REFERENCE,
detail=detail,
)
if load_subprojects:
self._loader_search_provenances[name] = provenance_node
for junction_name in junction_path:
loader = loader._get_loader(junction_name, provenance_node, load_subprojects=load_subprojects)
if load_subprojects:
del self._loader_search_provenances[name]
return loader
# ancestors()
#
# This will traverse all active loaders in the ancestry for which this
# project is reachable using a relative path.
#
# Yields:
# (Loader): Each loader in the ancestry
#
def ancestors(self):
traversed = {}
def foreach_parent(parent):
while parent:
if parent in traversed:
return
traversed[parent] = True
yield parent
parent = parent._parent
# Yield from the direct/active ancestry
yield from foreach_parent(self._parent)
# Yield from alternative parents which have been replaced by
# overrides in the ancestry.
for parent in self._alternative_parents:
yield from foreach_parent(parent)
###########################################
# Private Methods #
###########################################
# _load_file_no_deps():
#
# Load a bst file as a LoadElement
#
# This loads a bst file into a LoadElement but does no work to resolve
# the element's dependencies. The dependencies must be resolved properly
# before the LoadElement makes its way out of the loader.
#
# Args:
# filename (str): The element-path relative bst file
# provenance_node (Node): The location from where the file was referred to, or None
#
# Returns:
# (LoadElement): A partially-loaded LoadElement
#
def _load_file_no_deps(self, filename, provenance_node=None):
# Load the data and process any conditional statements therein
fullpath = os.path.join(self._basedir, filename)
try:
node = _yaml.load(
fullpath, shortname=filename, copy_tree=self.load_context.rewritable, project=self.project
)
except LoadError as e:
if e.reason == LoadErrorReason.MISSING_FILE:
if self.project.junction:
message = "Could not find element '{}' in project referred to by junction element '{}'".format(
filename, self.project.junction.name
)
else:
message = "Could not find element '{}' in elements directory '{}'".format(filename, self._basedir)
if provenance_node:
message = "{}: {}".format(provenance_node.get_provenance(), message)
# If we can't find the file, try to suggest plausible
# alternatives by stripping the element-path from the given
# filename, and verifying that it exists.
detail = None
elements_dir = os.path.relpath(self._basedir, self.project.directory)
element_relpath = os.path.relpath(filename, elements_dir)
if filename.startswith(elements_dir) and os.path.exists(os.path.join(self._basedir, element_relpath)):
detail = "Did you mean '{}'?".format(element_relpath)
raise LoadError(message, LoadErrorReason.MISSING_FILE, detail=detail) from e
if e.reason == LoadErrorReason.LOADING_DIRECTORY:
# If a .bst file exists in the element path,
# let's suggest this as a plausible alternative.
message = str(e)
if provenance_node:
message = "{}: {}".format(provenance_node.get_provenance(), message)
detail = None
if os.path.exists(os.path.join(self._basedir, filename + ".bst")):
element_name = filename + ".bst"
detail = "Did you mean '{}'?\n".format(element_name)
raise LoadError(message, LoadErrorReason.LOADING_DIRECTORY, detail=detail) from e
# Otherwise, we don't know the reason, so just raise
raise
kind = node.get_str(Symbol.KIND)
if kind in ("junction", "link"):
self._first_pass_options.process_node(node)
else:
self.project.ensure_fully_loaded()
self._includes.process(node)
element = LoadElement(node, filename, self)
self._elements[filename] = element
#
# Update link caches in the ancestry
#
if element.link_target is not None:
link_path = filename
target_path = element.link_target.as_str() # pylint: disable=no-member
# First resolve the link in this loader's cache
#
self._resolve_link(link_path, target_path)
# Now resolve the link in parent project loaders
#
loader = self
while loader._parent:
junction = loader.project.junction
link_path = junction.name + ":" + link_path
target_path = junction.name + ":" + target_path
# Resolve the link
loader = loader._parent
loader._resolve_link(link_path, target_path)
return element
# _resolve_link():
#
# Resolves a link in the loader's link cache.
#
# This will first insert the new link -> target relationship
# into the cache, and will also update any existing targets
# which might have pointed to this link, to point to the new
# target instead.
#
# Args:
# link_path (str): The local project relative real path to a link
# target_path (str): The new target for this link
#
def _resolve_link(self, link_path, target_path):
self._links[link_path] = target_path
for cached_link_path, cached_target_path in self._links.items():
if self._expand_link(cached_target_path) == link_path:
self._links[cached_link_path] = target_path
# _expand_link():
#
# Expands any links in the provided path and returns a real path with
# known link elements substituted for their targets.
#
# Args:
# path (str): A project relative path
#
# Returns:
# (str): The same path with any links expanded
#
def _expand_link(self, path):
# FIXME: This simply returns the first link, maybe
# this needs to be more iterative, or sorted by
# number of path components, or smth
for link, target in self._links.items():
if path.startswith(link):
return target + path[len(link) :]
return path
# _load_one_file():
#
# A helper function to load a single file within the _load_file() process,
# this allows us to handle redirections more consistently.
#
# Args:
# filename (str): The element-path relative bst file
# provenance_node (Node): The location from where the file was referred to, or None
# load_subprojects (bool): Whether to load subprojects
#
# Returns:
# (LoadElement): A LoadElement, which might be shallow loaded or fully loaded.
#
def _load_one_file(self, filename, provenance_node, *, load_subprojects=True):
element = None
# First check the cache, the cache might contain shallow loaded
# elements.
#
try:
element = self._elements[filename]
# If the cached element has already entered the loop which loads
# it's dependencies, it is fully loaded and any further checks in
# this function are expected to have already been performed.
#
if element.fully_loaded:
return element
except KeyError:
# Shallow load if it's not yet loaded.
element = self._load_file_no_deps(filename, provenance_node)
# Check if there was an override for this element
#
override = self._search_for_override_element(filename)
if override:
#
# If there was an override for the element, then it was
# implicitly fully loaded by _search_for_override_element(),
#
return override
# If this element is a link then we need to resolve it, and return
# the linked element instead of this one.
#
if element.link_target is not None:
link_target = element.link_target.as_str() # pylint: disable=no-member
_, filename, loader = self._parse_name(link_target, element.link_target, load_subprojects=load_subprojects)
#
# Redirect the loading of the file and it's dependencies to the appropriate loader,
# which might or might not be the same loader.
#
return loader._load_file(filename, element.link_target, load_subprojects=load_subprojects)
return element
# _load_file():
#
# Semi-Iteratively load bst files
#
# The "Semi-" qualification is because where junctions get involved there
# is a measure of recursion, though this is limited only to the points at
# which junctions are crossed.
#
# Args:
# filename (str): The element-path relative bst file
# provenance_node (Node): The location from where the file was referred to, or None
# load_subprojects (bool): Whether to load subprojects
#
# Returns:
# (LoadElement): A loaded LoadElement
#
def _load_file(self, filename, provenance_node, *, load_subprojects=True):
top_element = self._load_one_file(filename, provenance_node, load_subprojects=load_subprojects)
# Already loaded dependencies for a fully loaded element, early return.
#
if top_element.fully_loaded:
return top_element
#
# Mark the top element here as "fully loaded", so that we will avoid trying to
# load it's dependencies more than once.
#
top_element.mark_fully_loaded()
dependencies = extract_depends_from_node(top_element.node)
# The loader queue is a stack of tuples
# [0] is the LoadElement instance
# [1] is a stack of Dependency objects to load
# [2] is a list of dependency names used to warn when all deps are loaded
loader_queue = [(top_element, list(reversed(dependencies)), [])]
# Load all dependency files for the new LoadElement
while loader_queue:
if loader_queue[-1][1]:
current_element = loader_queue[-1]
# Process the first dependency of the last loaded element
dep = current_element[1].pop()
# And record its name for checking later
current_element[2].append(dep.name)
if dep.junction:
loader = self.get_loader(dep.junction, dep.node)
dep_element = loader._load_file(dep.name, dep.node)
else:
dep_element = self._load_one_file(dep.name, dep.node, load_subprojects=load_subprojects)
# If the loaded element is not fully loaded, queue up the dependencies to be loaded in this loop.
#
if not dep_element.fully_loaded:
# Mark the dep_element as fully_loaded, as we're already queueing it's deps
dep_element.mark_fully_loaded()
dep_deps = extract_depends_from_node(dep_element.node)
loader_queue.append((dep_element, list(reversed(dep_deps)), []))
# Pylint is not very happy about Cython and can't understand 'node' is a 'MappingNode'
if dep_element.node.get_str(Symbol.KIND) == "junction": # pylint: disable=no-member
raise LoadError(
"{}: Cannot depend on junction".format(dep.node.get_provenance()),
LoadErrorReason.INVALID_DATA,
)
# We've now resolved the element for this dependency, lets set the resolved
# LoadElement on the dependency and append the dependency to the owning
# LoadElement dependency list.
dep.set_element(dep_element)
current_element[0].dependencies.append(dep) # pylint: disable=no-member
else:
# We do not have any more dependencies to load for this
# element on the queue, report any invalid dep names
self._warn_invalid_elements(loader_queue[-1][2])
# And pop the element off the queue
loader_queue.pop()
# Nothing more in the queue, return the top level element we loaded.
return top_element
# _check_circular_deps():
#
# Detect circular dependencies on LoadElements with
# dependencies already resolved.
#
# Args:
# element (str): The element to check
#
# Raises:
# (LoadError): In case there was a circular dependency error
#
@staticmethod
def _check_circular_deps(top_element):
sequence = [top_element]
sequence_indices = [0]
check_elements = set(sequence)
validated = set()
while sequence:
this_element = sequence[-1]
index = sequence_indices[-1]
if index < len(this_element.dependencies):
element = this_element.dependencies[index].element
sequence_indices[-1] = index + 1
if element in check_elements:
# Create `chain`, the loop of element dependencies from this
# element back to itself, by trimming everything before this
# element from the sequence under consideration.
chain = [element.full_name for element in sequence[sequence.index(element) :]]
chain.append(element.full_name)
raise LoadError(
("Circular dependency detected at element: {}\n" + "Dependency chain: {}").format(
element.full_name, " -> ".join(chain)
),
LoadErrorReason.CIRCULAR_DEPENDENCY,
)
if element not in validated:
# We've not already validated this element, so let's
# descend into it to check it out
sequence.append(element)
sequence_indices.append(0)
check_elements.add(element)
# Otherwise we'll head back around the loop to validate the
# next dependency in this entry
else:
# Done with entry, pop it off, indicate we're no longer
# in its chain, and mark it valid
sequence.pop()
sequence_indices.pop()
check_elements.remove(this_element)
validated.add(this_element)
# _search_for_local_override():
#
# Search this project's active override list for an override, while
# considering any link elements.
#
# Args:
# override_path (str): The real relative path to search for
#
# Returns:
# (ScalarNode): The overridding node from this project's junction, or None
#
def _search_for_local_override(self, override_path):
junction = self.project.junction
if junction is None:
return None
# Try the override without any link substitutions first
with suppress(KeyError):
return junction.overrides[override_path]
#
# If we did not get an exact match here, we might still have
# an override where a link was used to specify the override.
#
for override_key, override_node in junction.overrides.items():
resolved_path = self._expand_link(override_key)
if resolved_path == override_path:
return override_node
return None
# _search_for_overrides():
#
# Search for parent loaders which have an override for the specified element,
# returning a list of loaders with the highest level overriding loader at the
# end of the list, and the closest ancestor being at the beginning of the list.
#
# Args:
# filename (str): The local element name
#
# Returns:
# (list): A list of loaders which override this element
#
def _search_for_overrides(self, filename):
loader = self
override_path = filename
# Collect any overrides to this junction in the ancestry
#
overriding_loaders = []
while loader._parent:
junction = loader.project.junction
override_node = loader._search_for_local_override(override_path)
if override_node:
overriding_loaders.append((loader._parent, override_node))
override_path = junction.name + ":" + override_path
loader = loader._parent
return overriding_loaders
# _search_for_override_loader():
#
# Search parent projects an override of the junction specified by @filename,
# returning the loader object which should be used in place of the local
# junction specified by @filename.
#
# This function is called once for each direct child while looking up
# child loaders, after which point the child loader is cached in the `_loaders`
# table. This function also has the side effect of recording alternative parents
# of a child loader in the case that the child loader is overridden.
#
# Args:
# filename (str): Junction name
#
# Returns:
# (Loader): The loader to use, in case @filename was overridden, otherwise None.
#
def _search_for_override_loader(self, filename):
overriding_loaders = self._search_for_overrides(filename)
# If there are any overriding loaders, use the highest one in
# the ancestry to lookup the loader for this project.
#
if overriding_loaders:
overriding_loader, override_node = overriding_loaders[-1]
loader = overriding_loader.get_loader(override_node.as_str(), override_node)
#
# Record alternative loaders which were overridden.
#
# When a junction is overridden by another higher priority junction,
# the resulting loader is still reachable with the original element paths,
# which will now traverse override redirections.
#
# In order to iterate over every project/loader in the ancestry which can
# reach the actually selected loader, we need to keep track of the parent
# loaders of all overridden junctions.
#
if loader is not self:
loader._alternative_parents.append(self)
del overriding_loaders[-1]
loader._alternative_parents.extend(l for l, _ in overriding_loaders)
return loader
# No overrides were found in the ancestry
#
return None
# _search_for_override_element():
#
# Search parent projects an override of the element specified by @filename,
# returning the loader object which should be used in place of the local
# element specified by @filename.
#
# Args:
# filename (str): Junction name
#
# Returns:
# (Loader): The loader to use, in case @filename was overridden, otherwise None.
#
def _search_for_override_element(self, filename):
element = None
# If there are any overriding loaders, use the highest one in
# the ancestry to lookup the element which should be used in place
# of @filename.
#
overriding_loaders = self._search_for_overrides(filename)
if overriding_loaders:
overriding_loader, override_node = overriding_loaders[-1]
_, filename, loader = overriding_loader._parse_name(override_node.as_str(), override_node)
element = loader._load_file(filename, override_node)
return element
# _get_loader():
#
# Return loader for specified junction
#
# Args:
# filename (str): Junction name
# load_subprojects (bool): Whether to load subprojects
# provenance_node (Node): The location from where the file was referred to, or None
#
# Raises: LoadError
#
# Returns: A Loader or None if specified junction does not exist
#
def _get_loader(self, filename, provenance_node, *, load_subprojects=True):
loader = None
# return previously determined result
if filename in self._loaders:
return self._loaders[filename]
# Local function to conditionally resolve the provenance prefix string
def provenance_str():
if provenance_node is not None:
return "{}: ".format(provenance_node.get_provenance())
return ""
#
# Search the ancestry for an overridden loader to use in place
# of using the locally defined junction.
#
override_loader = self._search_for_override_loader(filename)
if override_loader:
self._loaders[filename] = override_loader
return override_loader
#
# Load the junction file
#
self._load_file(filename, provenance_node, load_subprojects=load_subprojects)
# At this point we've loaded the LoadElement
load_element = self._elements[filename]
# If the loaded element is a link, then just follow it
# immediately and move on to the target.
#
if load_element.link_target:
_, filename, loader = self._parse_name(
load_element.link_target.as_str(), load_element.link_target, load_subprojects=load_subprojects
)
return loader.get_loader(filename, load_element.link_target, load_subprojects=load_subprojects)
# If we're only performing a lookup, we're done here.
#
if not load_subprojects:
return None
if load_element.kind != "junction":
raise LoadError(
"{}{}: Expected junction but element kind is {}".format(provenance_str(), filename, load_element.kind),
LoadErrorReason.INVALID_DATA,
)
# We check that junctions have no dependencies a little
# early. This is cheating, since we don't technically know
# that junctions aren't allowed to have dependencies.
#
# However, this makes progress reporting more intuitive
# because we don't need to load dependencies of an element
# that shouldn't have any, and therefore don't need to
# duplicate the load count for elements that shouldn't be.
#
# We also fail slightly earlier (since we don't need to go
# through the entire loading process), which is nice UX. It
# would be nice if this could be done for *all* element types,
# but since we haven't loaded those yet that's impossible.
if load_element.dependencies:
# Use the first dependency in the list as provenance
p = load_element.dependencies[0].node.get_provenance()
raise LoadError(
"{}: Dependencies are forbidden for 'junction' elements".format(p), LoadErrorReason.INVALID_JUNCTION
)
element = Element._new_from_load_element(load_element)
element._initialize_state()
# Handle the case where a subproject has no ref
#
if not element._has_all_sources_resolved():
detail = "Try tracking the junction element with `bst source track {}`".format(filename)
raise LoadError(
"{}Subproject has no ref for junction: {}".format(provenance_str(), filename),
LoadErrorReason.SUBPROJECT_INCONSISTENT,
detail=detail,
)
# Handle the case where a subproject needs to be fetched
#
if element._should_fetch():
self.load_context.fetch_subprojects([element])
sources = list(element.sources())
if len(sources) == 1 and sources[0]._get_local_path():
# Optimization for junctions with a single local source
basedir = sources[0]._get_local_path()
else:
# Stage sources
element._set_required()
# Note: We use _KeyStrength.WEAK here because junctions
# cannot have dependencies, therefore the keys are
# equivalent.
#
# Since the element has not necessarily been given a
# strong cache key at this point (in a non-strict build
# that is set *after* we complete building/pulling, which
# we haven't yet for this element),
# element._get_cache_key() can fail if used with the
# default _KeyStrength.STRONG.
basedir = os.path.join(
self.project.directory, ".bst", "staged-junctions", filename, element._get_cache_key(_KeyStrength.WEAK)
)
if not os.path.exists(basedir):
os.makedirs(basedir, exist_ok=True)
element._stage_sources_at(basedir)
# Load the project
project_dir = os.path.join(basedir, element.path)
try:
from .._project import Project # pylint: disable=cyclic-import
project = Project(
project_dir,
self.load_context.context,
junction=element,
parent_loader=self,
search_for_project=False,
provenance_node=provenance_node,
)
except LoadError as e:
if e.reason == LoadErrorReason.MISSING_PROJECT_CONF:
message = provenance_str() + "Could not find the project.conf file in the project " "referred to by junction element '{}'.".format(
element.name
)
if element.path:
message += " Was expecting it at path '{}' in the junction's source.".format(element.path)
raise LoadError(message=message, reason=LoadErrorReason.INVALID_JUNCTION) from e
# Otherwise, we don't know the reason, so just raise
raise
loader = project.loader
self._loaders[filename] = loader
# Now we've loaded a junction and it's project, we need to try to shallow
# load the overrides of this project and any projects in the ancestry which
# have overrides referring to this freshly loaded project.
#
# This is to ensure that link elements have been resolved as much as possible
# before we try to look for an override.
#
iter_loader = loader
while iter_loader._parent:
iter_loader._shallow_load_overrides()
iter_loader = iter_loader._parent
return loader
# _shallow_load_overrides():
#
# Loads any of the override elements on this loader's junction
#
def _shallow_load_overrides(self):
if not self.project.junction:
return
junction = self.project.junction
# Iterate over the keys, we want to ensure that links are resolved for
# override paths specified in junctions, while the targets of these paths
# are not consequential.
#
for override_path, override_target in junction.overrides.items():
# Ensure that we resolve indirect links, in case that shallow loading
# an element results in loading a link, we need to discover if it's
# target is also a link.
#
path = override_path
provenance_node = override_target
while path is not None:
path, provenance_node = self._shallow_load_path(path, provenance_node)
# _shallow_load_path()
#
# Perform a shallow load of an element by it's relative path, this is
# used to load elements which might be specified by their path and might
# not be used in the resulting load, like paths to elements overridden by
# junctions.
#
# It is only important to shallow load these referenced elements in case
# they are links which need to be known later on.
#
# Args:
# path (str): The path to load
# provenance_node (Node): The node to use for provenance
#
# Returns:
# (str): The target of the loaded link element, if it was a link element
# and it could be loaded presently, otherwise None.
# (ScalarNode): The link target real node, if a link target was returned
#
def _shallow_load_path(self, path, provenance_node):
if ":" in path:
junction, element_name = path.rsplit(":", 1)
target_loader = self.get_loader(junction, provenance_node, load_subprojects=False)
# Subproject not loaded, discard this shallow load attempt
#
if target_loader is None:
return None, None
else:
junction = None
element_name = path
target_loader = self
# If the element is already loaded in the target loader, then there
# is no need for a shallow load.
try:
element = target_loader._elements[element_name]
except KeyError:
# Shallow load the the element.
element = target_loader._load_file_no_deps(element_name, provenance_node)
if element.link_target:
link_target = element.link_target.as_str()
if junction:
return "{}:{}".format(junction, link_target), element.link_target
return link_target, element.link_target
return None, None
# _parse_name():
#
# Get junction and base name of element along with loader for the sub-project
#
# Args:
# name (str): Name of target
# provenance_node (Node): The provenance node
# load_subprojects (bool): Whether to load subprojects
#
# Returns:
# (tuple): - (str): name of the junction element
# - (str): name of the element
# - (Loader): loader for sub-project
#
def _parse_name(self, name, provenance_node, *, load_subprojects=True):
# We allow to split only once since deep junctions names are forbidden.
# Users who want to refer to elements in sub-sub-projects are required
# to create junctions on the top level project.
junction_path = name.rsplit(":", 1)
if len(junction_path) == 1:
return None, junction_path[-1], self
else:
loader = self.get_loader(junction_path[-2], provenance_node, load_subprojects=load_subprojects)
return junction_path[-2], junction_path[-1], loader
# Print a warning message, checks warning_token against project configuration
#
# Args:
# brief (str): The brief message
# warning_token (str): An optional configurable warning assosciated with this warning,
# this will cause PluginError to be raised if this warning is configured as fatal.
#
# Raises:
# (:class:`.LoadError`): When warning_token is considered fatal by the project configuration
#
def _warn(self, brief, *, warning_token=None):
if warning_token:
if self.project._warning_is_fatal(warning_token):
raise LoadError(brief, warning_token)
self.load_context.context.messenger.warn(brief)
# Print warning messages if any of the specified elements have invalid names.
#
# Valid filenames should end with ".bst" extension.
#
# Args:
# elements (list): List of element names
#
# Raises:
# (:class:`.LoadError`): When warning_token is considered fatal by the project configuration
#
def _warn_invalid_elements(self, elements):
# invalid_elements
#
# A dict that maps warning types to the matching elements.
invalid_elements = {
CoreWarnings.BAD_ELEMENT_SUFFIX: [],
CoreWarnings.BAD_CHARACTERS_IN_NAME: [],
}
for filename in elements:
if not filename.endswith(".bst"):
invalid_elements[CoreWarnings.BAD_ELEMENT_SUFFIX].append(filename)
if not valid_chars_name(filename):
invalid_elements[CoreWarnings.BAD_CHARACTERS_IN_NAME].append(filename)
if invalid_elements[CoreWarnings.BAD_ELEMENT_SUFFIX]:
self._warn(
"Target elements '{}' do not have expected file extension `.bst` "
"Improperly named elements will not be discoverable by commands".format(
invalid_elements[CoreWarnings.BAD_ELEMENT_SUFFIX]
),
warning_token=CoreWarnings.BAD_ELEMENT_SUFFIX,
)
if invalid_elements[CoreWarnings.BAD_CHARACTERS_IN_NAME]:
self._warn(
"Target elements '{}' have invalid characerts in their name.".format(
invalid_elements[CoreWarnings.BAD_CHARACTERS_IN_NAME]
),
warning_token=CoreWarnings.BAD_CHARACTERS_IN_NAME,
)
# _clean_caches()
#
# Clean internal loader caches, recursively
#
# When loading the elements, the loaders use caches in order to not load the
# same element twice. These are kept after loading and prevent garbage
# collection. Cleaning them explicitely is required.
#
def _clean_caches(self):
for loader in self._loaders.values():
# value may be None with nested junctions without overrides
if loader is not None:
loader._clean_caches()
self._meta_elements = {}
self._elements = {}