1import sys
2from typing import Any, Dict, Optional, Type, TypeVar
3
4from .cache import Cacheable
5from .utils import dump_json, parse_yaml
6
7R = TypeVar("R", bound="Resource")
8
9
10class Resource(Cacheable):
11 """
12 A resource that's part of the overall Ambassador configuration world. This is
13 the base class for IR resources, Ambassador-config resources, etc.
14
15 Elements in a Resource:
16 - rkey is a short identifier that is used as the primary key for _all_ the
17 Ambassador classes to identify this single specific resource. It should be
18 something like "ambassador-default.1" or the like: very specific, doesn't
19 have to be fun for humans.
20
21 - location is a more human-readable string describing where the human should
22 go to find the source of this resource. "Service ambassador, namespace default,
23 object 1". This isn't really used by the Config class, but the Diagnostics class
24 makes heavy use of it.
25
26 - kind (keyword-only) is what kind of Ambassador resource this is.
27
28 - serialization (keyword-only) is the _original input serialization_, if we have
29 it, of the object. If we don't have it, this should be None -- don't just serialize
30 the object to no purpose.
31
32 - any additional keyword arguments are saved in the Resource.
33
34 :param rkey: unique identifier for this source, should be short
35 :param location: where should a human go to find the source of this resource?
36 :param kind: what kind of thing is this?
37 :param serialization: original input serialization of obj, if we have it
38 :param kwargs: key-value pairs that form the data object for this resource
39 """
40
41 rkey: str
42 location: str
43 kind: str
44 serialization: Optional[str]
45
46 # _errors: List[RichStatus]
47 _errored: bool
48 _referenced_by: Dict[str, "Resource"]
49
50 def __init__(
51 self, rkey: str, location: str, *, kind: str, serialization: Optional[str] = None, **kwargs
52 ) -> None:
53
54 if not rkey:
55 raise Exception("Resource requires rkey")
56
57 if not kind:
58 raise Exception("Resource requires kind")
59
60 # print("Resource __init__ (%s %s)" % (kind, name))
61
62 super().__init__(
63 rkey=rkey,
64 location=location,
65 kind=kind,
66 serialization=serialization,
67 # _errors=[],
68 _referenced_by={},
69 **kwargs
70 )
71
72 def sourced_by(self, other: "Resource"):
73 self.rkey = other.rkey
74 self.location = other.location
75
76 def referenced_by(self, other: "Resource") -> None:
77 # print("%s %s REF BY %s %s" % (self.kind, self.name, other.kind, other.rkey))
78 self._referenced_by[other.location] = other
79
80 def is_referenced_by(self, other_location) -> Optional["Resource"]:
81 return self._referenced_by.get(other_location, None)
82
83 def __getattr__(self, key: str) -> Any:
84 try:
85 return self[key]
86 except KeyError:
87 raise AttributeError(key)
88
89 def __setattr__(self, key: str, value: Any) -> None:
90 self[key] = value
91
92 def __str__(self) -> str:
93 return "<%s %s>" % (self.kind, self.rkey)
94
95 def as_dict(self) -> Dict[str, Any]:
96 ad = dict(self)
97
98 ad.pop("rkey", None)
99 ad.pop("serialization", None)
100 ad.pop("location", None)
101 ad.pop("_referenced_by", None)
102 ad.pop("_errored", None)
103
104 return ad
105
106 def as_json(self):
107 return dump_json(self.as_dict(), pretty=True)
108
109 @classmethod
110 def from_resource(
111 cls: Type[R],
112 other: R,
113 rkey: Optional[str] = None,
114 location: Optional[str] = None,
115 kind: Optional[str] = None,
116 serialization: Optional[str] = None,
117 **kwargs
118 ) -> R:
119 """
120 Create a Resource by copying another Resource, possibly overriding elements
121 along the way.
122
123 NOTE WELL: if you pass in kwargs, we assume that any values are safe to use as-is
124 and DO NOT COPY THEM. Otherwise, we SHALLOW COPY other.attrs for the new Resource.
125
126 :param other: the base Resource we're copying
127 :param rkey: optional new rkey
128 :param location: optional new location
129 :param kind: optional new kind
130 :param serialization: optional new original input serialization
131 :param kwargs: optional new key-value pairs -- see discussion about copying above!
132 """
133
134 # rkey and location are required positional arguments. Fine.
135 new_rkey = rkey or other.rkey
136 new_location = location or other.location
137
138 # Make a shallow-copied dict that we can muck with...
139 new_attrs = dict(kwargs) if kwargs else dict(other)
140
141 # Don't include kind unless it comes in on this call.
142 if kind:
143 new_attrs["kind"] = kind
144 else:
145 new_attrs.pop("kind", None)
146
147 # Don't include serialization at all if we don't have one.
148 if serialization:
149 new_attrs["serialization"] = serialization
150 elif other.serialization:
151 new_attrs["serialization"] = other.serialization
152
153 # Make sure that things that shouldn't propagate are gone...
154 new_attrs.pop("rkey", None)
155 new_attrs.pop("location", None)
156 new_attrs.pop("_errors", None)
157 new_attrs.pop("_errored", None)
158 new_attrs.pop("_referenced_by", None)
159
160 # ...and finally, use new_attrs for all the keyword args when we set up
161 # the new instance.
162 return cls(new_rkey, new_location, **new_attrs)
163
164 @classmethod
165 def from_dict(
166 cls: Type[R], rkey: str, location: str, serialization: Optional[str], attrs: Dict
167 ) -> R:
168 """
169 Create a Resource or subclass thereof from a dictionary. The new Resource's rkey
170 and location must be handed in explicitly.
171
172 The difference between this and simply intializing a Resource object is that
173 from_dict will introspect the attrs passed in and create whatever kind of Resource
174 matches attrs['kind'] -- so for example, if kind is "Mapping", this method will
175 return a Mapping rather than a Resource.
176
177 :param rkey: unique identifier for this source, should be short
178 :param location: where should a human go to find the source of this resource?
179 :param serialization: original input serialization of obj
180 :param attrs: dictionary from which to initialize the new object
181 """
182
183 # So this is a touch odd but here we go. We want to use the Kind here to find
184 # the correct type.
185 ambassador = sys.modules["ambassador"]
186
187 resource_class: Optional[Type[R]] = getattr(ambassador, attrs["kind"], None)
188
189 if not resource_class:
190 resource_class = getattr(ambassador, "AC" + attrs["kind"], cls)
191 assert resource_class
192
193 # print("%s.from_dict: %s => %s" % (cls, attrs['kind'], resource_class))
194
195 return resource_class(rkey, location=location, serialization=serialization, **attrs)
196
197 @classmethod
198 def from_yaml(cls: Type[R], rkey: str, location: str, serialization: str) -> R:
199 """
200 Create a Resource from a YAML serialization. The new Resource's rkey
201 and location must be handed in explicitly, and of course in this case the
202 serialization is mandatory.
203
204 Raises an exception if the serialization is not parseable.
205
206 :param rkey: unique identifier for this source, should be short
207 :param location: where should a human go to find the source of this resource?
208 :param serialization: original input serialization of obj
209 """
210
211 attrs = parse_yaml(serialization)
212
213 return cls.from_dict(rkey, location, serialization, attrs)
View as plain text