-
Notifications
You must be signed in to change notification settings - Fork 1.1k
/
output.py
829 lines (677 loc) 路 31.4 KB
/
output.py
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
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
# Copyright 2016-2022, Pulumi Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import asyncio
import contextlib
import json
from functools import reduce
from inspect import isawaitable
from typing import (
Tuple,
Type,
TypeVar,
Generic,
Set,
Callable,
Awaitable,
Union,
cast,
Mapping,
Any,
List,
Dict,
Optional,
TYPE_CHECKING,
overload,
)
from . import _types
from . import runtime
from .runtime import rpc
from .runtime.sync_await import _sync_await
if TYPE_CHECKING:
from .resource import Resource
T = TypeVar("T")
T1 = TypeVar("T1")
T2 = TypeVar("T2")
T3 = TypeVar("T3")
T_co = TypeVar("T_co", covariant=True)
U = TypeVar("U")
Input = Union[T, Awaitable[T], "Output[T]"]
Inputs = Mapping[str, Input[Any]]
InputType = Union[T, Mapping[str, Any]]
class Output(Generic[T_co]):
"""
Output helps encode the relationship between Resources in a Pulumi application. Specifically an
Output holds onto a piece of Data and the Resource it was generated from. An Output value can
then be provided when constructing new Resources, allowing that new Resource to know both the
value as well as the Resource the value came from. This allows for a precise 'Resource
dependency graph' to be created, which properly tracks the relationship between resources.
"""
_is_known: Awaitable[bool]
"""
Whether or not this 'Output' should actually perform .apply calls. During a preview,
an Output value may not be known (because it would have to actually be computed by doing an
'update'). In that case, we don't want to perform any .apply calls as the callbacks
may not expect an undefined value. So, instead, we just transition to another Output
value that itself knows it should not perform .apply calls.
"""
_is_secret: Awaitable[bool]
"""
Whether or not this 'Output' should be treated as containing secret data. Secret outputs are tagged when
flowing across the RPC interface to the resource monitor, such that when they are persisted to disk in
our state file, they are encrypted instead of being in plaintext.
"""
_future: Awaitable[T_co]
"""
Future that actually produces the concrete value of this output.
"""
_resources: Awaitable[Set["Resource"]]
"""
The list of resources that this output value depends on.
"""
def __init__(
self,
resources: Union[Awaitable[Set["Resource"]], Set["Resource"]],
future: Awaitable[T_co],
is_known: Awaitable[bool],
is_secret: Optional[Awaitable[bool]] = None,
) -> None:
is_known = asyncio.ensure_future(is_known)
future = asyncio.ensure_future(future)
async def is_value_known() -> bool:
return await is_known and not contains_unknowns(await future)
if isinstance(resources, set):
self._resources = asyncio.Future()
self._resources.set_result(resources)
else:
self._resources = asyncio.ensure_future(resources)
self._future = future
self._is_known = asyncio.ensure_future(is_value_known())
if is_secret is not None:
self._is_secret = asyncio.ensure_future(is_secret)
else:
self._is_secret = asyncio.Future()
self._is_secret.set_result(False)
# Private implementation details - do not document.
def resources(self) -> Awaitable[Set["Resource"]]:
return self._resources
def future(self, with_unknowns: Optional[bool] = None) -> Awaitable[Optional[T_co]]:
# If the caller did not explicitly ask to see unknown values and the value of this output contains unnkowns,
# return None. This preserves compatibility with earlier versios of the Pulumi SDK.
async def get_value() -> Optional[T_co]:
val = await self._future
return None if not with_unknowns and contains_unknowns(val) else val
return asyncio.ensure_future(get_value())
def is_known(self) -> Awaitable[bool]:
return self._is_known
# End private implementation details.
def is_secret(self) -> Awaitable[bool]:
return self._is_secret
def apply(
self, func: Callable[[T_co], Input[U]], run_with_unknowns: Optional[bool] = None
) -> "Output[U]":
"""
Transforms the data of the output with the provided func. The result remains an
Output so that dependent resources can be properly tracked.
'func' is not allowed to make resources.
'func' can return other Outputs. This can be handy if you have a Output<SomeVal>
and you want to get a transitive dependency of it.
This function will be called during execution of a `pulumi up` request. It may not run
during `pulumi preview` (as the values of resources are of course may not be known then).
:param Callable[[T_co],Input[U]] func: A function that will, given this Output's value, transform the value to
an Input of some kind, where an Input is either a prompt value, a Future, or another Output of the given
type.
:return: A transformed Output obtained from running the transformation function on this Output's value.
:rtype: Output[U]
"""
result_resources: asyncio.Future[Set["Resource"]] = asyncio.Future()
result_is_known: asyncio.Future[bool] = asyncio.Future()
result_is_secret: asyncio.Future[bool] = asyncio.Future()
# The "run" coroutine actually runs the apply.
async def run() -> U:
resources: Set["Resource"] = set()
try:
# Await this output's details.
resources = await self._resources
is_known = await self._is_known
is_secret = await self._is_secret
value = await self._future
if runtime.is_dry_run():
# During previews only perform the apply if the engine was able to give us an actual value for this
# Output or if the caller is able to tolerate unknown values.
apply_during_preview = is_known or run_with_unknowns
if not apply_during_preview:
# We didn't actually run the function, our new Output is definitely
# **not** known.
result_resources.set_result(resources)
result_is_known.set_result(False)
result_is_secret.set_result(is_secret)
return cast(U, None)
# If we are running with unknown values and the value is explicitly unknown but does not actually
# contain any unknown values, collapse its value to the unknown value. This ensures that callbacks
# that expect to see unknowns during preview in outputs that are not known will always do so.
if (
not is_known
and run_with_unknowns
and not contains_unknowns(value)
):
value = cast(T_co, UNKNOWN)
transformed: Input[U] = func(value)
# Transformed is an Input, meaning there are three cases:
# 1. transformed is an Output[U]
if isinstance(transformed, Output):
transformed_as_output = cast(Output[U], transformed)
# Forward along the inner output's _resources, _is_known and _is_secret values.
transformed_resources = await transformed_as_output._resources
result_resources.set_result(resources | transformed_resources)
result_is_known.set_result(await transformed_as_output._is_known)
result_is_secret.set_result(
await transformed_as_output._is_secret or is_secret
)
return await transformed.future(with_unknowns=True)
# 2. transformed is an Awaitable[U]
if isawaitable(transformed):
# Since transformed is not an Output, it is known.
result_resources.set_result(resources)
result_is_known.set_result(True)
result_is_secret.set_result(is_secret)
return await cast(Awaitable[U], transformed)
# 3. transformed is U. It is trivially known.
result_resources.set_result(resources)
result_is_known.set_result(True)
result_is_secret.set_result(is_secret)
return cast(U, transformed)
finally:
with contextlib.suppress(asyncio.InvalidStateError):
result_resources.set_result(resources)
with contextlib.suppress(asyncio.InvalidStateError):
result_is_known.set_result(False)
with contextlib.suppress(asyncio.InvalidStateError):
result_is_secret.set_result(False)
run_fut = asyncio.ensure_future(run())
return Output(result_resources, run_fut, result_is_known, result_is_secret)
def __getattr__(self, item: str) -> "Output[Any]": # type: ignore
"""
Syntax sugar for retrieving attributes off of outputs.
:param str item: An attribute name.
:return: An Output of this Output's underlying value's property with the given name.
:rtype: Output[Any]
"""
def lift(v: Any) -> Any:
return UNKNOWN if isinstance(v, Unknown) else getattr(v, item)
return self.apply(lift, True)
def __getitem__(self, key: Any) -> "Output[Any]":
"""
Syntax sugar for looking up attributes dynamically off of outputs.
:param Any key: Key for the attribute dictionary.
:return: An Output of this Output's underlying value, keyed with the given key as if it were a dictionary.
:rtype: Output[Any]
"""
def lift(v: Any) -> Any:
return UNKNOWN if isinstance(v, Unknown) else cast(Any, v)[key]
return self.apply(lift, True)
def __iter__(self) -> Any:
"""
Output instances are not iterable, but since they implement __getitem__ we need to explicitly prevent
iteration by implementing __iter__ to raise a TypeError.
"""
raise TypeError(
"'Output' object is not iterable, consider iterating the underlying value inside an 'apply'"
)
@staticmethod
def from_input(val: Input[T_co]) -> "Output[T_co]":
"""
Takes an Input value and produces an Output value from it, deeply unwrapping nested Input values through nested
lists, dicts, and input classes. Nested objects of other types (including Resources) are not deeply unwrapped.
:param Input[T_co] val: An Input to be converted to an Output.
:return: A deeply-unwrapped Output that is guaranteed to not contain any Input values.
:rtype: Output[T_co]
"""
# Is it an output already? Recurse into the value contained within it.
if isinstance(val, Output):
return val.apply(Output.from_input, True)
# Is it an input type (i.e. args class)? Recurse into the values within.
typ = type(val)
if _types.is_input_type(typ):
# We know that any input type can safely be decomposed into it's `__dict__`, and then reconstructed
# via `type(**d)` from the (unwrapped) properties.
o_typ: Output[typ] = Output.all(**val.__dict__).apply( # type:ignore
lambda d: typ(**d)
)
return cast(Output[T_co], o_typ)
# Is a (non-empty) dict or list? Recurse into the values within them.
if val and isinstance(val, dict):
o_dict: Output[dict] = Output.all(**val)
return cast(Output[T_co], o_dict)
if val and isinstance(val, list):
o_list: Output[list] = Output.all(*val)
return cast(Output[T_co], o_list)
# If it's not an output, list, or dict, it must be known and not secret
is_known_fut: asyncio.Future[bool] = asyncio.Future()
is_secret_fut: asyncio.Future[bool] = asyncio.Future()
is_known_fut.set_result(True)
is_secret_fut.set_result(False)
# Is it awaitable? If so, schedule it for execution and use the resulting future
# as the value future for a new output.
if isawaitable(val):
val_fut = cast(asyncio.Future, val)
promise_output = Output(
set(), asyncio.ensure_future(val_fut), is_known_fut, is_secret_fut
)
return promise_output.apply(Output.from_input, True)
# Is it a prompt value? Set up a new resolved future and use that as the value future.
value_fut: asyncio.Future[Any] = asyncio.Future()
value_fut.set_result(val)
return Output(set(), value_fut, is_known_fut, is_secret_fut)
@staticmethod
def _from_input_shallow(val: Input[T]) -> "Output[T]":
"""
Like `from_input`, but does not recur deeply. Instead, checks if `val` is an `Output` value
and returns it as is. Otherwise, promotes a known value or future to `Output`.
:param Input[T] val: An Input to be converted to an Output.
:return: An Output corresponding to `val`.
:rtype: Output[T]
"""
if isinstance(val, Output):
return val
# If it's not an output, it must be known and not secret
is_known_fut: asyncio.Future[bool] = asyncio.Future()
is_secret_fut: asyncio.Future[bool] = asyncio.Future()
is_known_fut.set_result(True)
is_secret_fut.set_result(False)
if isawaitable(val):
val_fut = cast(asyncio.Future, val)
return Output(
set(), asyncio.ensure_future(val_fut), is_known_fut, is_secret_fut
)
# Is it a prompt value? Set up a new resolved future and use that as the value future.
value_fut: asyncio.Future[Any] = asyncio.Future()
value_fut.set_result(val)
return Output(set(), value_fut, is_known_fut, is_secret_fut)
@staticmethod
def unsecret(val: "Output[T]") -> "Output[T]":
"""
Takes an existing Output, deeply unwraps the nested values and returns a new Output without any secrets included
:param Output[T] val: An Output to be converted to a non-Secret Output.
:return: A deeply-unwrapped Output that is guaranteed to not contain any secret values.
:rtype: Output[T]
"""
is_secret: asyncio.Future[bool] = asyncio.Future()
is_secret.set_result(False)
return Output(val._resources, val._future, val._is_known, is_secret)
@staticmethod
def secret(val: Input[T]) -> "Output[T]":
"""
Takes an Input value and produces an Output value from it, deeply unwrapping nested Input values as necessary
given the type. It also marks the returned Output as a secret, so its contents will be persisted in an encrypted
form in state files.
:param Input[T] val: An Input to be converted to an Secret Output.
:return: A deeply-unwrapped Output that is guaranteed to not contain any Input values and is marked as a Secret.
:rtype: Output[T]
"""
o = Output.from_input(val)
is_secret: asyncio.Future[bool] = asyncio.Future()
is_secret.set_result(True)
return Output(o._resources, o._future, o._is_known, is_secret)
# According to mypy these overloads unsafely overlap, so we ignore the type check.
# https://mypy.readthedocs.io/en/stable/more_types.html#type-checking-the-variants:~:text=considered%20unsafely%20overlapping
@overload
@staticmethod
def all(*args: Input[T]) -> "Output[List[T]]": # type: ignore
...
@overload
@staticmethod
def all(**kwargs: Input[T]) -> "Output[Dict[str, T]]":
...
@staticmethod
def all(*args: Input[T], **kwargs: Input[T]):
"""
Produces an Output of a list (if args i.e a list of inputs are supplied)
or dict (if kwargs i.e. keyworded arguments are supplied).
This function can be used to combine multiple, separate Inputs into a single
Output which can then be used as the target of `apply`. Resource dependencies
are preserved in the returned Output.
Examples::
Output.all(foo, bar) -> Output[[foo, bar]]
Output.all(foo=foo, bar=bar) -> Output[{"foo": foo, "bar": bar}]
:param Input[T] args: A list of Inputs to convert.
:param Input[T] kwargs: A list of named Inputs to convert.
:return: An output of list or dict, converted from unnamed or named Inputs respectively.
"""
# Three asynchronous helper functions to assist in the implementation:
# is_known, which returns True if all of the input's values are known,
# and false if any of them are not known,
async def is_known(outputs: list):
is_known_futures = [o._is_known for o in outputs]
each_is_known = await asyncio.gather(*is_known_futures)
return all(each_is_known)
# is_secret, which returns True if any of the input values are secret, and
# false if none of them are secret.
async def is_secret(outputs: list):
is_secret_futures = [o._is_secret for o in outputs]
each_is_secret = await asyncio.gather(*is_secret_futures)
return any(each_is_secret)
async def get_resources(outputs: list):
resources_futures = [o._resources for o in outputs]
resources_agg = await asyncio.gather(*resources_futures)
# Merge the list of resource dependencies across all inputs.
return reduce(lambda acc, r: acc.union(r), resources_agg, set())
# gather_futures, which aggregates the list or dict of futures in each input to a future of a list or dict.
async def gather_futures(outputs: Union[dict, list]):
if isinstance(outputs, list):
value_futures_list = [
asyncio.ensure_future(o.future(with_unknowns=True)) for o in outputs
]
return await asyncio.gather(*value_futures_list)
value_futures_dict = {
k: asyncio.ensure_future(v.future(with_unknowns=True))
for k, v in outputs.items()
}
return await _gather_from_dict(value_futures_dict)
from_input = cast(
Callable[[Union[T, Awaitable[T], Output[T]]], Output[T]], Output.from_input
)
if args and kwargs:
raise ValueError(
"Output.all() was supplied a mix of named and unnamed inputs"
)
# First, map all inputs to outputs using `from_input`.
all_outputs: Union[list, dict] = (
{k: from_input(v) for k, v in kwargs.items()}
if kwargs
else [from_input(x) for x in args]
)
# Aggregate the list or dict of futures into a future of list or dict.
value_futures = asyncio.ensure_future(gather_futures(all_outputs))
# Aggregate whether or not this output is known.
output_values = (
[all_outputs[k] for k in all_outputs]
if isinstance(all_outputs, dict)
else all_outputs
)
resources_futures = asyncio.ensure_future(get_resources(output_values))
known_futures = asyncio.ensure_future(is_known(output_values))
secret_futures = asyncio.ensure_future(is_secret(output_values))
return Output(resources_futures, value_futures, known_futures, secret_futures)
@staticmethod
def concat(*args: Input[str]) -> "Output[str]":
"""
Concatenates a collection of Input[str] into a single Output[str].
This function takes a sequence of Input[str], stringifies each, and concatenates all values
into one final string. This can be used like so:
url = Output.concat("http://", server.hostname, ":", loadBalancer.port)
:param Input[str] args: A list of string Inputs to concatenate.
:return: A concatenated output string.
:rtype: Output[str]
"""
transformed_items: List[Input[Any]] = [Output.from_input(v) for v in args]
# invariant http://mypy.readthedocs.io/en/latest/common_issues.html#variance
return Output.all(*transformed_items).apply("".join) # type: ignore
@staticmethod
def format(
format_string: Input[str], *args: Input[object], **kwargs: Input[object]
) -> "Output[str]":
"""
Perform a string formatting operation.
This has the same semantics as `str.format` except it handles Input types.
:param Input[str] format_string: A formatting string
:param Input[object] args: Positional arguments for the format string
:param Input[object] kwargs: Keyword arguments for the format string
:return: A formatted output string.
:rtype: Output[str]
"""
if args and kwargs:
return _map3_output(
Output.from_input(format_string),
Output.all(*args),
Output.all(**kwargs),
lambda str, args, kwargs: str.format(*args, **kwargs),
)
if args:
return _map2_output(
Output.from_input(format_string),
Output.all(*args),
lambda str, args: str.format(*args),
)
if kwargs:
return _map2_output(
Output.from_input(format_string),
Output.all(**kwargs),
lambda str, kwargs: str.format(**kwargs),
)
return Output.from_input(format_string).apply(lambda str: str.format())
@staticmethod
def json_dumps(
obj: Input[Any],
*,
skipkeys: bool = False,
ensure_ascii: bool = True,
check_circular: bool = True,
allow_nan: bool = True,
cls: Type[json.JSONEncoder] = None,
indent: Optional[Union[int, str]] = None,
separators: Tuple[str, str] = None,
default: Callable[[Any], Any] = None,
sort_keys: bool = False,
**kw: Any
) -> "Output[str]":
"""
Uses json.dumps to serialize the given Input[object] value into a JSON string.
The arguments have the same meaning as in `json.dumps` except obj is an Input.
"""
if cls is None:
cls = json.JSONEncoder
output = Output.from_input(obj)
result_resources: asyncio.Future[Set["Resource"]] = asyncio.Future()
result_is_known: asyncio.Future[bool] = asyncio.Future()
result_is_secret: asyncio.Future[bool] = asyncio.Future()
async def run() -> str:
resources: Set["Resource"] = set()
try:
seen_unknown = False
seen_secret = False
seen_resources = set()
class OutputEncoder(cls): # type: ignore
def default(self, o):
if isinstance(o, Output):
nonlocal seen_unknown
nonlocal seen_secret
nonlocal seen_resources
# We need to synchronously wait for o to complete
async def wait_output() -> Tuple[object, bool, bool, set]:
return (
await o._future,
await o._is_known,
await o._is_secret,
await o._resources,
)
(result, known, secret, resources) = _sync_await(
asyncio.ensure_future(wait_output())
)
# Update the secret flag and set of seen resources
seen_secret = seen_secret or secret
seen_resources.update(resources)
if known:
return result
# The value wasn't known set the local seenUnknown variable and just return None
# so the serialization doesn't raise an exception at this point
seen_unknown = True
return None
return super().default(o)
# Await the output's details.
resources = await output._resources
is_known = await output._is_known
is_secret = await output._is_secret
value = await output._future
if not is_known:
result_resources.set_result(resources)
result_is_known.set_result(is_known)
result_is_secret.set_result(is_secret)
return cast(str, None)
# Try and dump using our special OutputEncoder to handle nested outputs
result = json.dumps(
value,
skipkeys=skipkeys,
ensure_ascii=ensure_ascii,
check_circular=check_circular,
allow_nan=allow_nan,
cls=OutputEncoder,
indent=indent,
separators=separators,
default=default,
sort_keys=sort_keys,
**kw
)
# Update the final resources and secret flag based on what we saw while dumping
is_secret = is_secret or seen_secret
resources = set(resources)
resources.update(seen_resources)
# If we saw an unknown during dumping then throw away the result and return not known
if seen_unknown:
result_resources.set_result(resources)
result_is_known.set_result(False)
result_is_secret.set_result(is_secret)
return cast(str, None)
result_resources.set_result(resources)
result_is_known.set_result(True)
result_is_secret.set_result(is_secret)
return result
finally:
with contextlib.suppress(asyncio.InvalidStateError):
result_resources.set_result(resources)
with contextlib.suppress(asyncio.InvalidStateError):
result_is_known.set_result(False)
with contextlib.suppress(asyncio.InvalidStateError):
result_is_secret.set_result(False)
run_fut = asyncio.ensure_future(run())
return Output(result_resources, run_fut, result_is_known, result_is_secret)
def __str__(self) -> str:
return """Calling __str__ on an Output[T] is not supported.
To get the value of an Output[T] as an Output[str] consider:
1. o.apply(lambda v: f"prefix{v}suffix")
See https://pulumi.io/help/outputs for more details.
This function may throw in a future version of Pulumi."""
class Unknown:
"""
Unknown represents a value that is unknown.
"""
def __init__(self):
pass
UNKNOWN = Unknown()
"""
UNKNOWN is the singleton unknown value.
"""
def contains_unknowns(val: Any) -> bool:
return rpc.contains_unknowns(val)
def _is_prompt(value: Input[T]) -> bool:
"""Checks if the value is prompty available."""
return not isawaitable(value) and not isinstance(value, Output)
def _map_output(o: Output[T], transform: Callable[[T], U]) -> Output[U]:
"""Transforms an output's result value with a pure function."""
async def fut() -> U:
value = await o.future()
return transform(value) if value is not None else cast(U, UNKNOWN)
return Output(
resources=o.resources(),
future=asyncio.ensure_future(fut()),
is_known=o.is_known(),
is_secret=o.is_secret(),
)
def _map2_output(
o1: Output[T1], o2: Output[T2], transform: Callable[[T1, T2], U]
) -> Output[U]:
"""
Joins two outputs and transforms their result with a pure function.
Similar to `all` but does not deeply await.
"""
async def fut() -> U:
v1 = await o1.future()
v2 = await o2.future()
return (
transform(v1, v2)
if (v1 is not None) and (v2 is not None)
else cast(U, UNKNOWN)
)
async def res() -> Set["Resource"]:
r1 = await o1.resources()
r2 = await o2.resources()
return r1 | r2
return Output(
resources=asyncio.ensure_future(res()),
future=asyncio.ensure_future(fut()),
is_known=o1.is_known() and o2.is_known(),
is_secret=o2.is_secret() or o2.is_secret(),
)
def _map3_output(
o1: Output[T1], o2: Output[T2], o3: Output[T3], transform: Callable[[T1, T2, T3], U]
) -> Output[U]:
"""
Joins three outputs and transforms their result with a pure function.
Similar to `all` but does not deeply await.
"""
async def fut() -> U:
v1 = await o1.future()
v2 = await o2.future()
v3 = await o3.future()
return (
transform(v1, v2, v3)
if (v1 is not None) and (v2 is not None) and (v3 is not None)
else cast(U, UNKNOWN)
)
async def res() -> Set["Resource"]:
r1 = await o1.resources()
r2 = await o2.resources()
r3 = await o3.resources()
return r1 | r2 | r3
return Output(
resources=asyncio.ensure_future(res()),
future=asyncio.ensure_future(fut()),
is_known=o1.is_known() and o2.is_known() and o3.is_known(),
is_secret=o2.is_secret() or o2.is_secret() or o3.is_secret(),
)
def _map_input(i: Input[T], transform: Callable[[T], U]) -> Input[U]:
"""Transforms an input's result value with a pure function."""
if _is_prompt(i):
return transform(cast(T, i))
if isawaitable(i):
inp = cast(Awaitable[T], i)
async def fut() -> U:
return transform(await inp)
return asyncio.ensure_future(fut())
return _map_output(cast(Output[T], i), transform)
def _map2_input(
i1: Input[T1], i2: Input[T2], transform: Callable[[T1, T2], U]
) -> Input[U]:
"""
Joins two inputs and transforms their result with a pure function.
"""
if _is_prompt(i1):
v1 = cast(T1, i1)
return _map_input(i2, lambda v2: transform(v1, v2))
if _is_prompt(i2):
v2 = cast(T2, i2)
return _map_input(i1, lambda v1: transform(v1, v2))
if isawaitable(i1) and isawaitable(i2):
a1 = cast(Awaitable[T1], i1)
a2 = cast(Awaitable[T2], i2)
async def join() -> U:
v1 = await a1
v2 = await a2
return transform(v1, v2)
return asyncio.ensure_future(join())
return _map2_output(
Output._from_input_shallow(i1), Output._from_input_shallow(i2), transform
)
async def _gather_from_dict(tasks: dict) -> dict:
results = await asyncio.gather(*tasks.values())
return dict(zip(tasks.keys(), results))