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
|
#
# 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 <http://www.gnu.org/licenses/>.
#
# Authors:
# Tristan Van Berkom <tristan.vanberkom@codethink.co.uk>
import os
from . import _yaml
from .node import _new_synthetic_file
from ._exceptions import LoadError
from .exceptions import LoadErrorReason
# ProjectRefStorage()
#
# Indicates the type of ref storage
class ProjectRefStorage:
# Source references are stored inline
#
INLINE = "inline"
# Source references are stored in a central project.refs file
#
PROJECT_REFS = "project.refs"
# ProjectRefs()
#
# The project.refs file management
#
# Args:
# directory (str): The project directory
# base_name (str): The project.refs basename
#
class ProjectRefs:
def __init__(self, directory, base_name):
directory = os.path.abspath(directory)
self._fullpath = os.path.join(directory, base_name)
self._base_name = base_name
self._toplevel_node = None
self._toplevel_save = None
# load()
#
# Load the project.refs file
#
# Args:
# options (OptionPool): To resolve conditional statements
#
def load(self, options):
try:
self._toplevel_node = _yaml.load(self._fullpath, shortname=self._base_name, copy_tree=True)
provenance = self._toplevel_node.get_provenance()
self._toplevel_save = provenance._toplevel
# Process any project options immediately
options.process_node(self._toplevel_node)
# Run any final assertions on the project.refs, just incase there
# are list composition directives or anything left unprocessed.
self._toplevel_node._assert_fully_composited()
except LoadError as e:
if e.reason != LoadErrorReason.MISSING_FILE:
raise
# Ignore failure if the file doesnt exist, it'll be created and
# for now just assumed to be empty
self._toplevel_node = _new_synthetic_file(self._fullpath)
self._toplevel_save = self._toplevel_node
self._toplevel_node.validate_keys(["projects"])
# Ensure we create our toplevel entry point on the fly here
for node in [self._toplevel_node, self._toplevel_save]:
if "projects" not in node:
node["projects"] = {}
# lookup_ref()
#
# Fetch the ref node for a given Source. If the ref node does not
# exist and `write` is specified, it will be automatically created.
#
# Args:
# project (str): The project to lookup
# element (str): The element name to lookup
# source_index (int): The index of the Source in the specified element
# write (bool): Whether we want to read the node or write to it
#
# Returns:
# (node): The YAML dictionary where the ref is stored
#
def lookup_ref(self, project, element, source_index, *, write=False):
node = self._lookup(self._toplevel_node, project, element, source_index)
if write:
# If we couldnt find the orignal, create a new one.
#
if node is None:
node = self._lookup(self._toplevel_save, project, element, source_index, ensure=True)
return node
# _lookup()
#
# Looks up a ref node in the project.refs file, creates one if ensure is True.
#
def _lookup(self, toplevel, project, element, source_index, *, ensure=False):
projects = toplevel.get_mapping("projects")
# Fetch the project
try:
project_node = projects.get_mapping(project)
except LoadError:
if not ensure:
return None
projects[project] = {}
project_node = projects.get_mapping(project)
# Fetch the element
try:
element_list = project_node.get_sequence(element)
except LoadError:
if not ensure:
return None
project_node[element] = []
element_list = project_node.get_sequence(element)
# Fetch the source index
try:
node = element_list.mapping_at(source_index)
except IndexError:
if not ensure:
return None
element_list.append({})
node = element_list.mapping_at(source_index)
return node
|
