in src/buildstream/_loader/loader.py [0:0]
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
)
if not loader:
# `loader` should never be None if `load_subprojects` is True
assert not load_subprojects
return None
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)
# 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
#
element._query_source_cache()
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