maxmustermann/pulumi
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Python SDK changes to support input/output classes (#5033)

Browse source 
Python SDK changes to support strongly-typed input/output "dataclasses".
This commit is contained in:
Justin Van Patten 2020-08-19 08:15:56 +00:00 committed by GitHub
parent 6eb475ab95
commit cd9fae599d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
29 changed files with 3173 additions and 89 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
CHANGELOG.md
View file

@ -18,6 +18,8 @@ CHANGELOG
- Add support for extracting jar files in archive resources
[#5150](https://github.com/pulumi/pulumi/pull/5150)
- SDK changes to support Python input/output classes
[#5033](https://github.com/pulumi/pulumi/pull/5033)
## 2.8.2 (2020-08-07)

12
sdk/python/lib/pulumi/__init__.py
View file

@ -69,6 +69,7 @@ from .output import (
Output,
Input,
Inputs,
InputType,
UNKNOWN,
contains_unknowns,
)
@ -84,4 +85,15 @@ from .stack_reference import (
StackReference,
)
# pylint: disable=redefined-builtin
from ._types import (
MISSING,
input_type,
output_type,
property,
getter,
get,
set,
)
from . import runtime, dynamic, policy

911
sdk/python/lib/pulumi/_types.py Normal file
View file

@ -0,0 +1,911 @@
# Copyright 2016-2020, 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.
# This module exports decorators, functions, and other helpers for defining input/output types.
#
# A resource can be declared as:
#
# class FooResource(pulumi.CustomResource):
# nested_value: pulumi.Output[Nested] = pulumi.property("nestedValue")
#
# def __init__(self, resource_name, nested_value: pulumi.InputType[NestedArgs]):
# super().__init__("my:module:FooResource", resource_name, {"nestedValue": nested_value})
#
#
# The resource declares a single output `nested_value` of type `pulumi.Output[Nested]` and uses
# `pulumi.property()` to indicate the Pulumi property name.
#
# The resource's `__init__()` method accepts a `nested_value` argument typed as
# `pulumi.InputType[NestedArgs]`, which is an alias for accepting either an input type (in this
# case `NestedArgs`) or `Mapping[str, Any]`. Input types are converted to a `dict` during
# serialization.
#
# When the resource's outputs are resolved, the `Nested` class is instantiated.
#
# The resource could alternatively be declared using a Python property for nested_value rather than
# using a class annotation:
#
# class FooResource(pulumi.CustomResource):
# def __init__(self, resource_name, nested_value: pulumi.InputType[NestedArgs]):
# super().__init__("my:module:FooResource", resource_name, {"nestedValue": nested_value})
#
# @property
# @pulumi.getter(name="nestedValue")
# def nested_value(self) -> pulumi.Output[Nested]:
# ...
#
#
# Note the `nested_value` property getter function is empty. The `@pulumi.getter` decorator replaces
# empty function bodies with an actual implementation. In this case, it replaces it with a body that
# looks like:
#
# @property
# @pulumi.getter(name="nestedValue")
# def nested_value(self) -> pulumi.Output[Nested]:
# pulumi.get(self, "nested_value")
#
#
# Here's how the `NestedArgs` input class can be declared:
#
# @pulumi.input_type
# class NestedArgs:
# first_arg: pulumi.Input[str] = pulumi.property("firstArg")
# second_arg: Optional[pulumi.Input[float]] = pulumi.property("secondArg", default=None)
#
#
# The class is decorated with the `@pulumi.input_type` decorator, which indicates the class is an
# input type and does some processing of the class (explained below). `NestedArgs` declares two
# inputs (`first_arg` and `second_arg`) and uses type annotations and `pulumi.property()` to
# specify the types and Pulumi input property names. An `__init__()` method is automatically added
# based on the annotations since one was not already present.
#
# A more verbose way to declare the same input type is as follows:
#
# @pulumi.input_type
# class NestedArgs:
# def __init__(self, *, first_arg: pulumi.Input[str], second_arg: Optional[pulumi.Input[float]] = None):
# pulumi.set(self, "first_arg", first_arg)
# if second_arg is not None:
# pulumi.set(self, "second_arg", second_arg)
#
# @property
# @pulumi.getter(name="firstArg")
# def first_arg(self) -> pulumi.Input[str]:
# ...
#
# @first_arg.setter
# def first_arg(self, value: pulumi.Input[str]):
# ...
#
# @property
# @pulumi.getter(name="secondArg")
# def second_arg(self) -> Optional[pulumi.Input[float]]:
# ...
#
# @second_arg.setter
# def second_arg(self, value: Optional[pulumi.Input[float]]):
# ...
#
# This latter (more verbose) declaration is equivalent to the former (simpler) declaration;
# the `@pulumi.input_type` processes the class and transforms the former declaration into the
# latter declaration.
#
# The former (simpler) declaration is syntactic sugar to use when declaring these by hand,
# e.g. when writing a dynamic provider that has nested inputs/outputs. The latter declaration isn't
# as pleasant to write by hand and is closer to what we emit in our provider codegen. The benefit
# of the latter (more verbose) form is that it allows docstrings to be specified on the Python
# property getters, which will show up in IDE tooltips when hovering over the property.
#
# Note the property getter/setter functions are empty in the more verbose declaration.
# Empty getter functions are automatically replaced by the `@pulumi.getter` decorator with an
# actual implementation, and the `@pulumi.input_type` decorator will automatically replace any
# empty setter functions associated with a getter decorated with `@pulumi.getter` with an actual
# implementation. Thus, the above is equivalent to this even more verbose form:
#
# @pulumi.input_type
# class NestedArgs:
# def __init__(self, *, first_arg: pulumi.Input[str], second_arg: Optional[pulumi.Input[float]] = None):
# pulumi.set(self, "first_arg", first_arg)
# if second_arg is not None:
# pulumi.set(self, "second_arg", second_arg)
#
# @property
# @pulumi.getter(name="firstArg")
# def first_arg(self) -> pulumi.Input[str]:
# return pulumi.get(self, "first_arg")
#
# @first_arg.setter
# def first_arg(self, value: pulumi.Input[str]):
# pulumi.set(self, "first_arg", value)
#
# @property
# @pulumi.getter(name="secondArg")
# def second_arg(self) -> Optional[pulumi.Input[float]]:
# return pulumi.get(self, "second_arg")
#
# @second_arg.setter
# def second_arg(self, value: Optional[pulumi.Input[float]]):
# pulumi.set(self, "second_arg", value)
#
#
# Here's how the `Nested` output class can be declared:
#
# @pulumi.output_type
# class Nested:
# first_arg: str = pulumi.property("firstArg")
# second_arg: Optional[float] = pulumi.property("secondArg")
#
#
# This is equivalent to the more verbose form:
#
# @pulumi.output_type
# class Nested:
# def __init__(self, *, first_arg: str, second_arg: Optional[float]):
# pulumi.set(self, "first_arg", first_arg)
# pulumi.set(self, "second_arg", second_arg)
#
# @property
# @pulumi.getter(name="firstArg")
# def first_arg(self) -> str:
# ...
#
# @property
# @pulumi.getter(name="secondArg")
# def second_arg(self) -> Optional[float]:
# ...
#
# An `__init__()` method is added to the class by the `@pulumi.output_type` decorator (if an
# `__init__()` method isn't already present on the class).
#
# Output types only have property getters and the bodies can be empty. Empty getter functions are
# replaced with implementations by the `@pulumi.getter` decorator.
#
# The above form is equivalent to:
#
# @pulumi.output_type
# class Nested:
# def __init__(self, *, first_arg: str, second_arg: Optional[float]):
# pulumi.set(self, "first_arg", first_arg)
# pulumi.set(self, "second_arg", second_arg)
#
# @property
# @pulumi.getter(name="firstArg")
# def first_arg(self) -> str:
# return pulumi.get(self, "first_arg")
#
# @property
# @pulumi.getter(name="secondArg")
# def second_arg(self) -> Optional[float]:
# return pulumi.get(self, "second_arg")
#
#
# Output classes can also be a subclass of `dict`. This is used in our provider codegen to maintain
# backwards compatibility, where previously these objects were instances of `dict`.
#
# @pulumi.output_type
# class Nested(dict):
# first_arg: str = pulumi.property("firstArg")
# second_arg: Optional[float] = pulumi.property("secondArg")
#
#
# The above output type, a subclass of `dict`, is equivalent to:
#
# @pulumi.output_type
# class Nested(dict):
# def __init__(self, *, first_arg: str, second_arg: Optional[float]):
# pulumi.set(self, "first_arg", first_arg)
# pulumi.set(self, "second_arg", second_arg)
#
# @property
# @pulumi.getter(name="firstArg")
# def first_arg(self) -> str:
# ...
#
# @property
# @pulumi.getter(name="secondArg")
# def second_arg(self) -> Optional[float]:
# ...
#
#
# Which is equivalent to:
#
# @pulumi.output_type
# class Nested(dict):
# def __init__(self, *, first_arg: str, second_arg: Optional[float]):
# pulumi.set(self, "first_arg", first_arg)
# pulumi.set(self, "second_arg", second_arg)
#
# @property
# @pulumi.getter(name="firstArg")
# def first_arg(self) -> str:
# return pulumi.get(self, "first_arg")
#
# @property
# @pulumi.getter(name="secondArg")
# def second_arg(self) -> Optional[float]:
# return pulumi.get(self, "second_arg")
#
#
# An output class can optionally include a `_translate_property(self, prop)` method, which
# `pulumi.get` and `pulumi.set` will call to translate the Pulumi property name to a translated
# key name before getting/setting the value in the dictionary. This is to provide backwards
# compatibility with our provider generated code, where mapping tables are used to translate dict
# keys before being returned to the program. This way, existing programs accessing the values as a
# dictionary will continue to see the same translated key names as before, but updated programs can
# now access the values using Python properties, which will always have thecorrect snake_case
# Python names.
#
# @pulumi.output_type
# class Nested(dict):
# def __init__(self, *, first_arg: str, second_arg: Optional[float]):
# pulumi.set(self, "first_arg", first_arg)
# pulumi.set(self, "second_arg", second_arg)
#
# @property
# @pulumi.getter(name="firstArg")
# def first_arg(self) -> str:
# ...
#
# @property
# @pulumi.getter(name="secondArg")
# def second_arg(self) -> Optional[float]:
# ...
#
# def _translate_property(self, prop):
# return _tables.CAMEL_TO_SNAKE_CASE_TABLE.get(prop) or prop
import builtins
import functools
import sys
import typing
from typing import Any, Callable, Dict, Iterator, Optional, Tuple, Type, TypeVar, Union, cast, get_type_hints
from . import _utils
T = TypeVar('T')
_PULUMI_NAME = "_pulumi_name"
_PULUMI_INPUT_TYPE = "_pulumi_input_type"
_PULUMI_OUTPUT_TYPE = "_pulumi_output_type"
_PULUMI_PYTHON_TO_PULUMI_TABLE = "_pulumi_python_to_pulumi_table"
_TRANSLATE_PROPERTY = "_translate_property"
def is_input_type(cls: type) -> bool:
return hasattr(cls, _PULUMI_INPUT_TYPE)
def is_output_type(cls: type) -> bool:
return hasattr(cls, _PULUMI_OUTPUT_TYPE)
class _MISSING_TYPE:
pass
MISSING = _MISSING_TYPE()
"""
MISSING is a singleton sentinel object to detect if a parameter is supplied or not.
"""
class _Property:
"""
Represents a Pulumi property. It is not meant to be created outside this module,
rather, the property() function should be used.
"""
def __init__(self, name: str, default: Any = MISSING) -> None:
if not name:
raise TypeError("Missing name argument")
if not isinstance(name, str):
raise TypeError("Expected name to be a string")
self.name = name
self.default = default
self.type: Any = None
# This function's return type is deliberately annotated as Any so that type checkers do not
# complain about assignments that we want to allow like `my_value: str = property("myValue")`.
# pylint: disable=redefined-builtin
def property(name: str, *, default: Any = MISSING) -> Any:
"""
Return an object to identify Pulumi properties.
name is the Pulumi property name.
"""
return _Property(name, default)
def _properties_from_annotations(cls: type) -> Dict[str, _Property]:
"""
Returns a dictionary of properties from annotations defined on the class.
"""
# Get annotations that are defined on this class (not base classes).
# These are returned in the order declared on Python 3.6+.
cls_annotations = cls.__dict__.get('__annotations__', {})
def get_property(cls: type, a_name: str, a_type: Any) -> _Property:
default = getattr(cls, a_name, MISSING)
p = default if isinstance(default, _Property) else _Property(name=a_name, default=default)
p.type = a_type
return p
return {
name: get_property(cls, name, type)
for name, type in cls_annotations.items()
}
def _process_class(cls: type, signifier_attr: str, is_input: bool = False, setter: bool = False):
# Get properties.
props = _properties_from_annotations(cls)
# Clean-up class attributes.
for name in props:
# If the class attribute (which is the default value for this prop)
# exists and is of type 'Property', delete the class attribute so
# it is not set at all in the post-processed class.
if isinstance(getattr(cls, name, None), _Property):
delattr(cls, name)
# Mark this class with the signifier and save the properties.
setattr(cls, signifier_attr, True)
# Create Python properties.
for name, prop in props.items():
setattr(cls, name, _create_py_property(name, prop.name, prop.type, setter))
# Add an __init__() method if the class doesn't have one.
if "__init__" not in cls.__dict__:
if cls.__module__ in sys.modules:
globals = sys.modules[cls.__module__].__dict__
else:
globals = {}
init_fn = _init_fn(props, globals, issubclass(cls, dict), not is_input and hasattr(cls, _TRANSLATE_PROPERTY))
setattr(cls, "__init__", init_fn)
# Add an __eq__() method if the class doesn't have one.
# There's no need for a __ne__ method, since Python will call __eq__ and negate it.
if "__eq__" not in cls.__dict__:
if issubclass(cls, dict):
def eq_fn(self, other):
return type(other) is type(self) and getattr(dict, "__eq__")(other, self)
else:
def eq_fn(self, other):
return type(other) is type(self) and other.__dict__ == self.__dict__
setattr(cls, "__eq__", eq_fn)
def _create_py_property(a_name: str, pulumi_name: str, typ: Any, setter: bool = False):
"""
Returns a Python property getter that looks up the value using get.
"""
def getter_fn(self):
return get(self, a_name)
getter_fn.__name__ = a_name
getter_fn.__annotations__ = {"return": typ}
setattr(getter_fn, _PULUMI_NAME, pulumi_name)
if setter:
def setter_fn(self, value):
return set(self, a_name, value)
setter_fn.__name__ = a_name
setter_fn.__annotations__ = {"value": typ}
return builtins.property(fget=getter_fn, fset=setter_fn)
return builtins.property(fget=getter_fn)
def _py_properties(cls: type) -> Iterator[Tuple[str, str, builtins.property]]:
for python_name, v in cls.__dict__.items():
if isinstance(v, builtins.property):
prop = cast(builtins.property, v)
pulumi_name = getattr(prop.fget, _PULUMI_NAME, MISSING)
if pulumi_name is not MISSING:
yield (python_name, pulumi_name, prop)
def input_type(cls: Type[T]) -> Type[T]:
"""
Returns the same class as was passed in, but marked as an input type.
"""
if is_input_type(cls) or is_output_type(cls):
raise AssertionError("Cannot apply @input_type and @output_type more than once.")
# Get the input properties and mark the class as an input type.
_process_class(cls, _PULUMI_INPUT_TYPE, is_input=True, setter=True)
# Helper to create a setter function.
def create_setter(name: str) -> Callable:
def setter_fn(self, value):
set(self, name, value)
return setter_fn
# Now, process the class's properties, replacing properties with empty setters with
# an actual setter.
for python_name, _, prop in _py_properties(cls):
if prop.fset is not None and _utils.is_empty_function(prop.fset):
setter_fn = create_setter(python_name)
setter_fn.__name__ = prop.fset.__name__
setter_fn.__annotations__ = prop.fset.__annotations__
# Replace the property with a new property object that has the new setter.
setattr(cls, python_name, prop.setter(setter_fn))
return cls
def input_type_to_dict(obj: Any) -> Dict[str, Any]:
"""
Returns a dict for the input type.
The keys of the dict are Pulumi names that should not be translated.
"""
cls = type(obj)
assert is_input_type(cls)
# Build a dictionary of properties to return
result: Dict[str, Any] = {}
for _, pulumi_name, prop in _py_properties(cls):
value = prop.fget(obj) # type: ignore
# We treat properties with a value of None as if they don't exist.
if value is not None:
result[pulumi_name] = value
return result
def output_type(cls: Type[T]) -> Type[T]:
"""
Returns the same class as was passed in, but marked as an output type.
Python property getters are created for each Pulumi output property
defined in the class.
If the class is not a subclass of dict and doesn't have an __init__
method, an __init__ method is added to the class that accepts a dict
representing the outputs.
"""
if is_input_type(cls) or is_output_type(cls):
raise AssertionError("Cannot apply @input_type and @output_type more than once.")
# Get the output properties and mark the class as an output type.
_process_class(cls, _PULUMI_OUTPUT_TYPE)
# If the class has a _translate_property() method, build a mapping table of Python names to
# Pulumi names. Calls to pulumi.get() will then convert the name passed to pulumi.get() from
# the Python name to the Pulumi name, and then pass the Pulumi name to _translate_property() to
# convert the Pulumi name to whatever name _translate_property() returns (which, for our
# provider codegen, will be the translated name from _tables.CAMEL_TO_SNAKE_CASE_TABLE).
# pylint: disable=too-many-nested-blocks
if hasattr(cls, _TRANSLATE_PROPERTY):
python_to_pulumi_table = None
for python_name, pulumi_name, _ in _py_properties(cls):
if python_name != pulumi_name:
python_to_pulumi_table = python_to_pulumi_table or {}
python_to_pulumi_table[python_name] = pulumi_name
if python_to_pulumi_table is not None:
setattr(cls, _PULUMI_PYTHON_TO_PULUMI_TABLE, python_to_pulumi_table)
return cls
def output_type_from_dict(cls: Type[T], output: Dict[str, Any]) -> T:
assert isinstance(output, dict)
assert is_output_type(cls)
args = {}
for python_name, pulumi_name, _ in _py_properties(cls):
args[python_name] = output.get(pulumi_name)
return cls(**args) # type: ignore
def getter(_fn=None, *, name: Optional[str] = None):
"""
Decorator to indicate a function is a Pulumi property getter.
name is the Pulumi property name. If not set, the name of the function is used.
"""
def decorator(fn: Callable) -> Callable:
if not callable(fn):
raise TypeError("Expected fn to be callable")
# If name isn't specified, use the name of the function.
pulumi_name = name if name is not None else fn.__name__
if _utils.is_empty_function(fn):
@functools.wraps(fn)
def get_fn(self):
# Get the value using the Python name, which is the name of the function.
return get(self, fn.__name__)
fn = get_fn
setattr(fn, _PULUMI_NAME, pulumi_name)
return fn
# See if we're being called as @getter or @getter().
if _fn is None:
# We're called with parens.
return decorator
# We're called as @getter without parens.
return decorator(_fn)
def _translate_name(obj: Any, name: str) -> str:
cls = type(obj)
if hasattr(cls, _PULUMI_OUTPUT_TYPE):
# If the class has a _translate_property() method we need to do two translations:
# 1. Translate Python => Pulumi name.
# 2. Translate Pulumi name => result of _translate_property().
translate = getattr(cls, _TRANSLATE_PROPERTY, None)
if callable(translate):
table = getattr(cls, _PULUMI_PYTHON_TO_PULUMI_TABLE, None)
if isinstance(table, dict):
name = table.get(name) or name
name = translate(obj, name)
return name
def get(self, name: str) -> Any:
"""
Used to get values in types decorated with @input_type or @output_type.
"""
if not name:
raise TypeError("Missing name argument")
if not isinstance(name, str):
raise TypeError("Expected name to be a string")
cls = type(self)
if hasattr(cls, _PULUMI_INPUT_TYPE):
return self.__dict__.get(name)
if hasattr(cls, _PULUMI_OUTPUT_TYPE):
name = _translate_name(self, name)
if issubclass(cls, dict):
# Grab dict's `get` method instead of calling `self.get` directly
# in case the type has a `get` property.
return getattr(dict, "get")(self, name)
return self.__dict__.get(name)
# pylint: disable=import-outside-toplevel
from . import Resource
if isinstance(self, Resource):
return self.__dict__.get(name)
raise AssertionError("get can only be used with classes decorated with @input_type or @output_type")
def set(self, name: str, value: Any) -> None:
"""
Used to set values in types decorated with @input_type or @output_type.
"""
if not name:
raise TypeError("Missing name argument")
if not isinstance(name, str):
raise TypeError("Expected name to be a string")
cls = type(self)
if hasattr(cls, _PULUMI_INPUT_TYPE):
self.__dict__[name] = value
return
if hasattr(cls, _PULUMI_OUTPUT_TYPE):
name = _translate_name(self, name)
if issubclass(cls, dict):
self[name] = value
else:
self.__dict__[name] = value
return
raise AssertionError("set can only be used with classes decorated with @input_type or @output_type")
# Use the built-in `get_origin` and `get_args` functions on Python 3.8+,
# otherwise fallback to downlevel implementations.
if sys.version_info[:2] >= (3, 8):
# pylint: disable=no-member
get_origin = typing.get_origin # type: ignore
# pylint: disable=no-member
get_args = typing.get_args # type: ignore
elif sys.version_info[:2] >= (3, 7):
def get_origin(tp):
if isinstance(tp, typing._GenericAlias): # type: ignore
return tp.__origin__
return None
def get_args(tp):
if isinstance(tp, typing._GenericAlias): # type: ignore
return tp.__args__
return ()
else:
def get_origin(tp):
if hasattr(tp, "__origin__"):
return tp.__origin__
return None
def get_args(tp):
if hasattr(tp, "__args__"):
return tp.__args__
return ()
def _is_union_type(tp):
if sys.version_info[:2] >= (3, 7):
return (tp is Union or
isinstance(tp, typing._GenericAlias) and tp.__origin__ is Union) # type: ignore
# pylint: disable=unidiomatic-typecheck, no-member
return type(tp) is typing._Union # type: ignore
def _is_optional_type(tp):
if tp is type(None):
return True
if _is_union_type(tp):
return any(_is_optional_type(tt) for tt in get_args(tp))
return False
def _types_from_py_properties(cls: type) -> Dict[str, type]:
"""
Returns a dict of Pulumi names to types for a type.
"""
# pylint: disable=import-outside-toplevel
from . import Output
# We use get_type_hints() below on each Python property to resolve the getter function's
# return type annotation, resolving forward references.
#
# We pass the cls's globals to get_type_hints() to ensure any other referenced
# output types (which may exist in other modules of the cls, like `.outputs` or
# `...meta.v1.outputs`) can be resolved. If we didn't pass the cls's globals,
# get_type_hints() would use the __globals__ of the function, which likely does not contain
# the necessary references, as the function was likely created internally inside this module
# (either via the @output_type decorator, which converts class annotations into Python
# properties, or via the @getter decorator, which replaces empty getter functions) and
# therefore has __globals__ of this SDK module.
globalns = None
if cls.__module__ in sys.modules:
globalns = dict(sys.modules[cls.__module__].__dict__)
# Build-up a dictionary of Pulumi property names to types by looping through all the
# Python properties on the class that have a getter marked as a Pulumi property getter,
# and looking at the getter function's return type annotation.
# Types that are Output[T] and Optional[T] are unwrapped to just T.
result: Dict[str, type] = {}
for _, pulumi_name, prop in _py_properties(cls):
cls_hints = get_type_hints(prop.fget, globalns=globalns)
# Get the function's return type hint.
return_hint = cls_hints.get("return")
if return_hint is not None:
typ = _unwrap_type(return_hint)
# If typ is Output, it was specified non-generically (as Output rather than Output[T]),
# because _unwrap_type would have returned the T in Output[T] if it was specified
# generically. To avoid raising a type mismatch error when the deserialized output type
# doesn't match Output, we exclude it from the results.
if typ is Output:
continue
result[pulumi_name] = typ
return result
def _types_from_annotations(cls: type) -> Dict[str, type]:
"""
Returns a dict of Pulumi names to types for a type.
"""
# Get the "Pulumi properties" from the class's type annotations.
props = _properties_from_annotations(cls)
if not props:
return {}
# pylint: disable=import-outside-toplevel
from . import Output
# We want resolved types for just the cls's type annotations (not base classes),
# but get_type_hints() looks at the annotations of the class and its base classes.
# So create a type dynamically that has the annotations from cls but doesn't have
# any base classes, and pass the dynamically created type to get_type_hints().
dynamic_cls_attrs = {"__annotations__": cls.__dict__.get("__annotations__", {})}
dynamic_cls = type(cls.__name__, (object,), dynamic_cls_attrs)
# Pass along globals for the cls, to help resolve forward references.
globalns = None
if getattr(cls, "__module__", None) in sys.modules:
globalns = dict(sys.modules[cls.__module__].__dict__)
# Pass along Output as a local, as it is a forward reference type annotation on the base
# CustomResource class that can be instantiated directly (which our tests do).
localns = {"Output": Output} # type: ignore
# Get the type hints, resolving any forward references.
cls_hints = get_type_hints(dynamic_cls, globalns=globalns, localns=localns)
# Return a dictionary of Pulumi property names to types. Types that are Output[T] and
# Optional[T] are unwrapped to just T.
result: Dict[str, type] = {}
for name, prop in props.items():
typ = _unwrap_type(cls_hints[name])
# If typ is Output, it was specified non-generically (as Output rather than Output[T]),
# because _unwrap_type would have returned the T in Output[T] if it was specified
# generically. To avoid raising a type mismatch error when the deserialized output type
# doesn't match Output, we exclude it from the results.
if typ is Output:
continue
result[prop.name] = typ
return result
def output_type_types(output_type_cls: type) -> Dict[str, type]:
"""
Returns a dict of Pulumi names to types for the output type.
"""
assert is_output_type(output_type_cls)
return _types_from_py_properties(output_type_cls)
def resource_types(resource_cls: type) -> Dict[str, type]:
"""
Returns a dict of Pulumi names to types for the resource.
"""
# First, get the "Pulumi properties" from the class's type annotations.
types_from_annotations = _types_from_annotations(resource_cls)
# Next, get the types from the class's Python properties.
types_from_py_properties = _types_from_py_properties(resource_cls)
# Return the merged dictionaries.
return {**types_from_annotations, **types_from_py_properties}
def unwrap_optional_type(val: type) -> type:
"""
Unwraps the type T in Optional[T].
"""
# If it is Optional[T], extract the arg T. Note that Optional[T] is really Union[T, None],
# and any nested Unions are flattened, so Optional[Union[T, U], None] is Union[T, U, None].
# We'll only "unwrap" for the common case of a single arg T for Union[T, None].
if _is_optional_type(val):
args = get_args(val)
if len(args) == 2:
assert args[1] is type(None)
val = args[0]
return val
def _unwrap_type(val: type) -> type:
"""
Unwraps the type T in Output[T] and Optional[T].
"""
# pylint: disable=import-outside-toplevel
from . import Output
origin = get_origin(val)
# If it is an Output[T], extract the T arg.
if origin is Output:
args = get_args(val)
assert len(args) == 1
val = args[0]
return unwrap_optional_type(val)
# The following functions for creating an __init__() method were adapted
# from Python's dataclasses module.
def _create_fn(name, args, body, *, globals=None, locals=None):
if locals is None:
locals = {}
if "BUILTINS" not in locals:
locals["BUILTINS"] = builtins
args = ",".join(args)
body = "\n".join(f" {b}" for b in body)
# Compute the text of the entire function.
txt = f" def {name}({args}):\n{body}"
local_vars = ", ".join(locals.keys())
txt = f"def __create_fn__({local_vars}):\n{txt}\n return {name}"
ns = {}
exec(txt, globals, ns) # pylint: disable=exec-used
return ns["__create_fn__"](**locals)
def _property_init(python_name: str, prop: _Property, globals, is_dict: bool, has_translate: bool):
# Return the text of the line in the body of __init__() that will
# initialize this property.
default_name = f"_dflt_{python_name}"
if prop.default is MISSING:
# There's no default, just do an assignment.
value = python_name
else:
globals[default_name] = python_name
value = python_name
# Now, actually generate the assignment.
if is_dict:
# It's a dict, store the value in itself.
container = ""
else:
# It isn't a dict, store the value in __dict__.
container = ".__dict__"
# Only assign the value if not None.
if prop.default is None:
check = f"if {value} is not None:\n "
else:
check = ""
# If it has a _translate_property method, use it to translate the name.
if has_translate:
return f"{check}__self__{container}[__self__.{_TRANSLATE_PROPERTY}('{prop.name}')]={value}"
return f"{check}__self__{container}['{python_name}']={value}"
def _init_param(python_name: str, prop: _Property):
# Return the __init__ parameter string for this property. For
# example, the equivalent of 'x:int=3' (except instead of 'int',
# reference a variable set to int, and instead of '3', reference a
# variable set to 3).
if prop.default is MISSING:
# There's no default, just output the variable name and type.
default = ""
else:
# There's a default, this will be the name that's used to look it up.
default = f"=_dflt_{python_name}"
return f"{python_name}:_type_{python_name}{default}"
def _init_fn(props: Dict[str, _Property], globals, is_dict: bool, has_translate: bool):
# Make sure we don't have properties without defaults following properties
# with defaults. This actually would be caught when exec-ing the
# function source code, but catching it here gives a better error
# message, and future-proofs us in case we build up the function
# using ast.
seen_default = False
for python_name, prop in props.items():
if prop.default is not MISSING:
seen_default = True
elif seen_default:
raise TypeError(f"non-default argument {python_name!r} "
"follows default argument")
locals = {f"_type_{python_name}": prop.type for python_name, prop in props.items()}
locals.update({
"MISSING": MISSING,
})
body_lines = []
for python_name, prop in props.items():
line = _property_init(python_name, prop, locals, is_dict, has_translate)
body_lines.append(line)
# If no body lines, use `pass`.
if not body_lines:
body_lines = ["pass"]
first_args = ["__self__"]
# If we have args after __self__, use bare * to force them to be specified by name.
if len(props) > 0:
first_args += ["*"]
return _create_fn("__init__",
first_args + [_init_param(python_name, prop) for python_name, prop in props.items()],
body_lines,
locals=locals,
globals=globals)

57
sdk/python/lib/pulumi/_utils.py Normal file
View file

@ -0,0 +1,57 @@
# Copyright 2016-2020, 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 typing
# Empty function definitions.
def _empty():
...
def _empty_doc():
"""Empty function docstring."""
...
_empty_lambda = lambda: None
_empty_lambda_doc = lambda: None
_empty_lambda_doc.__doc__ = """Empty lambda docstring."""
def _consts(fn: typing.Callable) -> tuple:
"""
Returns a tuple of the function's constants excluding the docstring.
"""
return tuple(x for x in fn.__code__.co_consts if x != fn.__doc__)
# Precompute constants for each of the empty functions.
_consts_empty = _consts(_empty)
_consts_empty_doc = _consts(_empty_doc)
_consts_empty_lambda = _consts(_empty_lambda)
_consts_empty_lambda_doc = _consts(_empty_lambda_doc)
def is_empty_function(fn: typing.Callable) -> bool:
"""
Returns true if the function is empty.
"""
consts = _consts(fn)
return (
(fn.__code__.co_code == _empty.__code__.co_code and consts == _consts_empty) or
(fn.__code__.co_code == _empty_doc.__code__.co_code and consts == _consts_empty_doc) or
(fn.__code__.co_code == _empty_lambda.__code__.co_code and consts == _consts_empty_lambda) or
(fn.__code__.co_code == _empty_lambda_doc.__code__.co_code and consts == _consts_empty_lambda_doc)
)

1
sdk/python/lib/pulumi/output.py
View file

@ -40,6 +40,7 @@ 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]):

43
sdk/python/lib/pulumi/resource.py
View file

@ -18,7 +18,7 @@ from typing import Optional, List, Any, Mapping, Union, Callable, TYPE_CHECKING,
import copy
from .runtime import known_types
from .runtime.resource import _register_resource, register_resource_outputs, _read_resource
from .runtime.resource import register_resource, register_resource_outputs, read_resource
from .runtime.settings import get_root_resource
from .metadata import get_project, get_stack
@ -561,12 +561,6 @@ class Resource:
Resource represents a class whose CRUD operations are implemented by a provider plugin.
"""
urn: 'Output[str]'
"""
The stable, logical URN used to distinctly address a resource, both before and after
deployments.
"""
_providers: Mapping[str, 'ProviderResource']
"""
The set of providers to use for child resources. Keyed by package name (e.g. "aws").
@ -709,18 +703,17 @@ class Resource:
if not custom:
raise Exception(
"Cannot read an existing resource unless it has a custom provider")
res = cast('CustomResource', self)
result = _read_resource(res, t, name, props, opts)
res.urn = result.urn
assert result.id is not None
res.id = result.id
read_resource(cast('CustomResource', self), t, name, props, opts)
else:
result = _register_resource(self, t, name, custom, props, opts)
self.urn = result.urn
if custom:
assert result.id is not None
res = cast('CustomResource', self)
res.id = result.id
register_resource(self, t, name, custom, props, opts)
@property
def urn(self) -> 'Output[str]':
"""
The stable, logical URN used to distinctly address a resource, both before and after
deployments.
"""
return self.__dict__["urn"]
def _convert_providers(self, provider: Optional['ProviderResource'], providers: Optional[Union[Mapping[str, 'ProviderResource'], List['ProviderResource']]]) -> Mapping[str, 'ProviderResource']:
if provider is not None:
@ -787,12 +780,6 @@ class CustomResource(Resource):
dynamically loaded plugin for the defining package.
"""
id: 'Output[str]'
"""
id is the provider-assigned unique ID for this managed resource. It is set during
deployments and may be missing (undefined) during planning phases.
"""
__pulumi_type: str
"""
Private field containing the type ID for this object. Useful for implementing `isInstance` on
@ -814,6 +801,14 @@ class CustomResource(Resource):
Resource.__init__(self, t, name, True, props, opts)
self.__pulumi_type = t
@property
def id(self) -> 'Output[str]':
"""
id is the provider-assigned unique ID for this managed resource. It is set during
deployments and may be missing (undefined) during planning phases.
"""
return self.__dict__["id"]
class ComponentResource(Resource):
"""

12
sdk/python/lib/pulumi/runtime/invoke.py
View file

@ -13,10 +13,11 @@
# limitations under the License.
import asyncio
import sys
from typing import Any, Awaitable, TYPE_CHECKING
from typing import Any, Awaitable, Optional, TYPE_CHECKING
import grpc
from .. import log
from .. import _types
from ..invoke import InvokeOptions
from ..runtime.proto import provider_pb2
from . import rpc
@ -59,7 +60,7 @@ class InvokeResult:
__iter__ = __await__
def invoke(tok: str, props: 'Inputs', opts: InvokeOptions = None) -> InvokeResult:
def invoke(tok: str, props: 'Inputs', opts: Optional[InvokeOptions] = None, typ: Optional[type] = None) -> InvokeResult:
"""
invoke dynamically invokes the function, tok, which is offered by a provider plugin. The inputs
can be a bag of computed values (Ts or Awaitable[T]s), and the result is a Awaitable[Any] that
@ -69,6 +70,9 @@ def invoke(tok: str, props: 'Inputs', opts: InvokeOptions = None) -> InvokeResul
if opts is None:
opts = InvokeOptions()
if typ and not _types.is_output_type(typ):
raise TypeError("Expected typ to be decorated with @output_type")
async def do_invoke():
# If a parent was provided, but no provider was provided, use the parent's provider if one was specified.
if opts.parent is not None and opts.provider is None:
@ -115,7 +119,9 @@ def invoke(tok: str, props: 'Inputs', opts: InvokeOptions = None) -> InvokeResul
# Otherwise, return the output properties.
ret_obj = getattr(resp, 'return')
if ret_obj:
return rpc.deserialize_properties(ret_obj)
deserialized = rpc.deserialize_properties(ret_obj)
# If typ is not None, call translate_output_properties to instantiate any output types.
return rpc.translate_output_properties(deserialized, lambda prop: prop, typ) if typ else deserialized
return {}
async def do_rpc():

63
sdk/python/lib/pulumi/runtime/resource.py
View file

@ -138,20 +138,7 @@ async def prepare_resource(res: 'Resource',
)
class _ResourceResult(NamedTuple):
urn: 'Output[str]'
"""
The URN of the resource.
"""
id: Optional['Output[str]'] = None
"""
The id of the resource, if it's a CustomResource, otherwise None.
"""
# pylint: disable=too-many-locals,too-many-statements
def _read_resource(res: 'CustomResource', ty: str, name: str, props: 'Inputs', opts: 'ResourceOptions') -> _ResourceResult:
def read_resource(res: 'CustomResource', ty: str, name: str, props: 'Inputs', opts: 'ResourceOptions') -> None:
from .. import Output # pylint: disable=import-outside-toplevel
if opts.id is None:
raise Exception(
@ -172,7 +159,7 @@ def _read_resource(res: 'CustomResource', ty: str, name: str, props: 'Inputs', o
urn_secret.set_result(False)
resolve_urn = urn_future.set_result
resolve_urn_exn = urn_future.set_exception
result_urn = Output({res}, urn_future, urn_known, urn_secret)
res.__dict__["urn"] = Output({res}, urn_future, urn_known, urn_secret)
# Furthermore, since resources being Read must always be custom resources (enforced in the
# Resource constructor), we'll need to set up the ID field which will be populated at the end of
@ -184,7 +171,7 @@ def _read_resource(res: 'CustomResource', ty: str, name: str, props: 'Inputs', o
resolve_value: asyncio.Future[Any] = asyncio.Future()
resolve_perform_apply: asyncio.Future[bool] = asyncio.Future()
resolve_secret: asyncio.Future[bool] = asyncio.Future()
result_id = Output(
res.__dict__["id"] = Output(
{res}, resolve_value, resolve_perform_apply, resolve_secret)
def do_resolve(value: Any, perform_apply: bool, exn: Optional[Exception]):
@ -273,23 +260,14 @@ def _read_resource(res: 'CustomResource', ty: str, name: str, props: 'Inputs', o
asyncio.ensure_future(RPC_MANAGER.do_rpc("read resource", do_read)())
return _ResourceResult(result_urn, result_id)
def read_resource(res: 'CustomResource', ty: str, name: str, props: 'Inputs', opts: 'ResourceOptions') -> None:
result = _read_resource(res, ty, name, props, opts)
res.urn = result.urn
assert result.id is not None
res.id = result.id
# pylint: disable=too-many-locals,too-many-statements
def _register_resource(res: 'Resource',
ty: str,
name: str,
custom: bool,
props: 'Inputs',
opts: Optional['ResourceOptions']) -> _ResourceResult:
def register_resource(res: 'Resource', ty: str, name: str, custom: bool, props: 'Inputs', opts: Optional['ResourceOptions']) -> None:
"""
Registers a new resource object with a given type t and name. It returns the
auto-generated URN and the ID that will resolve after the deployment has completed. All
properties will be initialized to property objects that the registration operation will resolve
at the right time (or remain unresolved for deployments).
"""
log.debug(f"registering resource: ty={ty}, name={name}, custom={custom}")
monitor = settings.get_monitor()
from .. import Output # pylint: disable=import-outside-toplevel
@ -307,17 +285,16 @@ def _register_resource(res: 'Resource',
urn_secret.set_result(False)
resolve_urn = urn_future.set_result
resolve_urn_exn = urn_future.set_exception
result_urn = Output({res}, urn_future, urn_known, urn_secret)
res.__dict__["urn"] = Output({res}, urn_future, urn_known, urn_secret)
# If a custom resource, make room for the ID property.
result_id = None
resolve_id: Optional[Callable[[
Any, bool, Optional[Exception]], None]] = None
if custom:
resolve_value: asyncio.Future[Any] = asyncio.Future()
resolve_perform_apply: asyncio.Future[bool] = asyncio.Future()
resolve_secret: asyncio.Future[bool] = asyncio.Future()
result_id = Output(
res.__dict__["id"] = Output(
{res}, resolve_value, resolve_perform_apply, resolve_secret)
def do_resolve(value: Any, perform_apply: bool, exn: Optional[Exception]):
@ -451,22 +428,6 @@ def _register_resource(res: 'Resource',
asyncio.ensure_future(RPC_MANAGER.do_rpc(
"register resource", do_register)())
return _ResourceResult(result_urn, result_id)
def register_resource(res: 'Resource', ty: str, name: str, custom: bool, props: 'Inputs', opts: Optional['ResourceOptions']) -> None:
"""
Registers a new resource object with a given type t and name. It returns the
auto-generated URN and the ID that will resolve after the deployment has completed. All
properties will be initialized to property objects that the registration operation will resolve
at the right time (or remain unresolved for deployments).
"""
result = _register_resource(res, ty, name, custom, props, opts)
res.urn = result.urn
if custom:
assert result.id is not None
res = cast('CustomResource', res)
res.id = result.id
def register_resource_outputs(res: 'Resource', outputs: 'Union[Inputs, Output[Inputs]]'):
async def do_register_resource_outputs():

94
sdk/python/lib/pulumi/runtime/rpc.py
View file

@ -15,15 +15,18 @@
Support for serializing and deserializing properties going into or flowing
out of RPC calls.
"""
import sys
import asyncio
import collections
import functools
import inspect
from typing import List, Any, Callable, Dict, Optional, TYPE_CHECKING, cast
from typing import List, Any, Callable, Dict, Mapping, Optional, Tuple, Union, TYPE_CHECKING, cast, get_type_hints
from google.protobuf import struct_pb2
import six
from . import known_types, settings
from .. import log
from .. import _types
if TYPE_CHECKING:
from ..output import Inputs, Input, Output
@ -182,11 +185,20 @@ async def serialize_property(value: 'Input[Any]',
}
return value
if isinstance(value, dict):
transform_keys = True
# If value is an input type, convert it to a dict, and set transform_keys to False to prevent
# transforming the keys of the resulting dict as the keys should already be the final names.
value_cls = type(value)
if _types.is_input_type(value_cls):
value = _types.input_type_to_dict(value)
transform_keys = False
if isinstance(value, Mapping): # pylint: disable=bad-option-value,isinstance-second-argument-not-valid-type
obj = {}
for k, v in value.items():
transformed_key = k
if input_transformer is not None:
if transform_keys and input_transformer is not None:
transformed_key = input_transformer(k)
log.debug(f"transforming input property: {k} -> {transformed_key}")
obj[transformed_key] = await serialize_property(v, deps, input_transformer)
@ -357,15 +369,20 @@ def transfer_properties(res: 'Resource', props: 'Inputs') -> Dict[str, Resolver]
# using res.translate_output_property and then use *that* name to index into the resolvers table.
log.debug(f"adding resolver {name}")
resolvers[name] = functools.partial(do_resolve, resolve_value, resolve_is_known, resolve_is_secret)
res.__setattr__(name, Output({res}, resolve_value, resolve_is_known, resolve_is_secret))
res.__dict__[name] = Output({res}, resolve_value, resolve_is_known, resolve_is_secret)
return resolvers
def translate_output_properties(res: 'Resource', output: Any) -> Any:
def translate_output_properties(output: Any,
output_transformer: Callable[[str], str],
typ: Optional[type] = None) -> Any:
"""
Recursively rewrite keys of objects returned by the engine to conform with a naming
convention specified by the resource's implementation of `translate_output_property`.
convention specified by `output_transformer`.
Additionally, if output is a `dict` and `typ` is an output type, instantiate the output type,
passing the dict as an argument to the output type's __init__() method.
If output is a `dict`, every key is translated using `translate_output_property` while every value is transformed
by recursing.
@ -374,11 +391,65 @@ def translate_output_properties(res: 'Resource', output: Any) -> Any:
If output is a primitive (i.e. not a dict or list), the value is returned without modification.
"""
# If it's a secret, unwrap the value so the output is in alignment with the expected type.
if is_rpc_secret(output):
output = unwrap_rpc_secret(output)
# Unwrap optional types.
typ = _types.unwrap_optional_type(typ) if typ else typ
if isinstance(output, dict):
return {res.translate_output_property(k): translate_output_properties(res, v) for k, v in output.items()}
# Function called to lookup a type for a given key.
# The default always returns None.
get_type: Callable[[str], Optional[type]] = lambda k: None
if typ and _types.is_output_type(typ):
# If typ is an output type, get its types, so we can pass
# the type along for each property.
types = _types.output_type_types(typ)
get_type = lambda k: types.get(k) # pylint: disable=unnecessary-lambda
elif typ:
# If typ is a dict, get the type for its values, to pass
# along for each key.
origin = _types.get_origin(typ)
if typ is dict or origin in {dict, Dict, Mapping, collections.abc.Mapping}:
args = _types.get_args(typ)
if len(args) == 2 and args[0] is str:
get_type = lambda k: args[1]
else:
raise AssertionError(f"Unexpected type; expected 'dict' got '{typ}'")
# If typ is an output type, instantiate it. We do not translate the top-level keys,
# as the output type will take care of doing that if it has a _translate_property()
# method.
if typ and _types.is_output_type(typ):
translated_values = {
k: translate_output_properties(v, output_transformer, get_type(k))
for k, v in output.items()
}
return _types.output_type_from_dict(typ, translated_values)
# Otherwise, return the fully translated dict.
return {
output_transformer(k):
translate_output_properties(v, output_transformer, get_type(k))
for k, v in output.items()
}
if isinstance(output, list):
return [translate_output_properties(res, v) for v in output]
element_type: Optional[type] = None
if typ:
# If typ is a list, get the type for its values, to pass
# along for each item.
origin = _types.get_origin(typ)
if typ is list or origin in {list, List}:
args = _types.get_args(typ)
if len(args) == 1:
element_type = args[0]
else:
raise AssertionError(f"Unexpected type. Expected 'list' got '{typ}'")
return [translate_output_properties(v, output_transformer, element_type) for v in output]
return output
@ -407,10 +478,13 @@ async def resolve_outputs(res: 'Resource',
# Produce a combined set of property states, starting with inputs and then applying
# outputs. If the same property exists in the inputs and outputs states, the output wins.
all_properties = {}
# Get the resource's output types, so we can convert dicts from the engine into actual
# instantiated output types as needed.
types = _types.resource_types(type(res))
for key, value in deserialize_properties(outputs).items():
# Outputs coming from the provider are NOT translated. Do so here.
translated_key = res.translate_output_property(key)
translated_value = translate_output_properties(res, value)
translated_value = translate_output_properties(value, res.translate_output_property, types.get(key))
log.debug(f"incoming output property translated: {key} -> {translated_key}")
log.debug(f"incoming output value translated: {value} -> {translated_value}")
all_properties[translated_key] = translated_value
@ -421,7 +495,7 @@ async def resolve_outputs(res: 'Resource',
if translated_key not in all_properties:
# input prop the engine didn't give us a final value for.Just use the value passed into the resource by
# the user.
all_properties[translated_key] = translate_output_properties(res, deserialize_property(value))
all_properties[translated_key] = translate_output_properties(deserialize_property(value), res.translate_output_property, types.get(key))
for key, value in all_properties.items():
# Skip "id" and "urn", since we handle those specially.

13
sdk/python/lib/test/langhost/invoke_types/__init__.py Normal file
View file

@ -0,0 +1,13 @@
# Copyright 2016-2020, 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.

57
sdk/python/lib/test/langhost/invoke_types/__main__.py Normal file
View file

@ -0,0 +1,57 @@
# Copyright 2016-2020, 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 pulumi
from pulumi.runtime import invoke
import outputs
def my_function(first_value: str, second_value: float) -> outputs.MyFunctionResult:
return invoke("test:index:MyFunction",
props={"firstValue": first_value, "secondValue": second_value},
typ=outputs.MyFunctionResult).value
def my_other_function(first_value: str, second_value: float) -> outputs.MyOtherFunctionResult:
return invoke("test:index:MyOtherFunction",
props={"firstValue": first_value, "secondValue": second_value},
typ=outputs.MyOtherFunctionResult).value
def assert_eq(l, r):
assert l == r
class MyResource(pulumi.CustomResource):
first_value: pulumi.Output[str]
second_value: pulumi.Output[float]
def __init__(self, name: str, first_value: str, second_value: float):
super().__init__("test:index:MyResource", name, {
"first_value": first_value,
"second_value": second_value,
})
result = my_function("hello", 42)
res = MyResource("resourceA", result.nested.first_value, result.nested.second_value)
res.first_value.apply(lambda v: assert_eq(v, "hellohello"))
res.second_value.apply(lambda v: assert_eq(v, 43))
result = my_other_function("world", 100)
res2 = MyResource("resourceB", result.nested.first_value, result.nested.second_value)
res2.first_value.apply(lambda v: assert_eq(v, "worldworld"))
res2.second_value.apply(lambda v: assert_eq(v, 101))

57
sdk/python/lib/test/langhost/invoke_types/outputs.py Normal file
View file

@ -0,0 +1,57 @@
# Copyright 2016-2020, 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 pulumi
import outputs
@pulumi.output_type
class MyFunctionNestedResult:
first_value: str = pulumi.property("firstValue")
second_value: float = pulumi.property("secondValue")
@pulumi.output_type
class MyFunctionResult:
# Deliberately using a qualified (with `outputs.`) forward reference
# to mimic our provider codegen, to ensure the type can be resolved.
nested: 'outputs.MyFunctionNestedResult'
@pulumi.output_type
class MyOtherFunctionNestedResult:
def __init__(self, first_value: str, second_value: float):
pulumi.set(self, "first_value", first_value)
pulumi.set(self, "second_value", second_value)
@property
@pulumi.getter(name="firstValue")
def first_value(self) -> str:
...
@property
@pulumi.getter(name="secondValue")
def second_value(self) -> float:
...
@pulumi.output_type
class MyOtherFunctionResult:
def __init__(self, nested: 'outputs.MyOtherFunctionNestedResult'):
pulumi.set(self, "nested", nested)
@property
@pulumi.getter
# Deliberately using a qualified (with `outputs.`) forward reference
# to mimic our provider codegen, to ensure the type can be resolved.
def nested(self) -> 'outputs.MyOtherFunctionNestedResult':
...

66
sdk/python/lib/test/langhost/invoke_types/test_invoke.py Normal file
View file

@ -0,0 +1,66 @@
# Copyright 2016-2020, 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.
from os import path
from ..util import LanghostTest
class TestInvoke(LanghostTest):
def test_invoke_success(self):
self.run_test(
program=path.join(self.base_path(), "invoke_types"),
expected_resource_count=2)
def invoke(self, _ctx, token, args, provider, _version):
def result(expected_first_value: str, expected_second_value: float):
self.assertDictEqual({
"firstValue": expected_first_value,
"secondValue": expected_second_value,
}, args)
return ([], {
"nested": {
"firstValue": args["firstValue"] * 2,
"secondValue": args["secondValue"] + 1,
},
})
if token == "test:index:MyFunction":
return result("hello", 42)
elif token == "test:index:MyOtherFunction":
return result("world", 100)
else:
self.fail(f"unexpected token {token}")
def register_resource(self, _ctx, _dry_run, ty, name, resource, _deps,
_parent, _custom, _protect, _provider, _property_deps, _delete_before_replace,
_ignore_changes, _version):
if name == "resourceA":
self.assertEqual({
"first_value": "hellohello",
"second_value": 43,
}, resource)
elif name == "resourceB":
self.assertEqual({
"first_value": "worldworld",
"second_value": 101,
}, resource)
else:
self.fail(f"unknown resource: {name}")
return {
"urn": self.make_urn(ty, name),
"id": name,
"object": resource,
}

13
sdk/python/lib/test/langhost/types/__init__.py Normal file
View file

@ -0,0 +1,13 @@
# Copyright 2016-2020, 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.

402
sdk/python/lib/test/langhost/types/__main__.py Normal file
View file

@ -0,0 +1,402 @@
# Copyright 2016-2020, 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.
from typing import Optional
import pulumi
@pulumi.input_type
class AdditionalArgs:
first_value: pulumi.Input[str] = pulumi.property("firstValue")
second_value: Optional[pulumi.Input[float]] = pulumi.property("secondValue")
def __init__(self, first_value: pulumi.Input[str], second_value: Optional[pulumi.Input[float]] = None):
pulumi.set(self, "first_value", first_value)
pulumi.set(self, "second_value", second_value)
@pulumi.output_type
class Additional(dict):
first_value: str = pulumi.property("firstValue")
second_value: Optional[float] = pulumi.property("secondValue")
class AdditionalResource(pulumi.CustomResource):
additional: pulumi.Output[Additional]
def __init__(self, name: str, additional: pulumi.InputType[AdditionalArgs]):
super().__init__("test:index:AdditionalResource", name, {"additional": additional})
# Create a resource with input object.
res = AdditionalResource("testres", additional=AdditionalArgs(first_value="hello", second_value=42))
# Create a resource using the output object of another resource.
res2 = AdditionalResource("testres2", additional=AdditionalArgs(
first_value=res.additional.first_value,
second_value=res.additional.second_value,
))
# Create a resource using the output object of another resource, accessing the output as a dict.
res3 = AdditionalResource("testres3", additional=AdditionalArgs(
first_value=res.additional["first_value"],
second_value=res.additional["second_value"],
))
# Create a resource using a dict as the input.
# Note: These are camel case (not snake_case) since the resource does not do any translation of
# property names.
res4 = AdditionalResource("testres4", additional={
"firstValue": "hello",
"secondValue": 42,
})
# Now, test some resources that use property translations.
SNAKE_TO_CAMEL_CASE_TABLE = {
"first_value": "firstValue",
"second_value": "secondValue",
}
CAMEL_TO_SNAKE_CASE_TABLE = {
"firstValue": "first_value",
"secondValue": "second_value",
}
@pulumi.input_type
class ExtraArgs:
first_value: pulumi.Input[str] = pulumi.property("firstValue")
second_value: Optional[pulumi.Input[float]] = pulumi.property("secondValue")
def __init__(self, first_value: pulumi.Input[str], second_value: Optional[pulumi.Input[float]] = None):
pulumi.set(self, "first_value", first_value)
pulumi.set(self, "second_value", second_value)
@pulumi.output_type
class Extra(dict):
first_value: str = pulumi.property("firstValue")
second_value: Optional[float] = pulumi.property("secondValue")
def _translate_property(self, prop):
return CAMEL_TO_SNAKE_CASE_TABLE.get(prop) or prop
class ExtraResource(pulumi.CustomResource):
extra: pulumi.Output[Extra]
def __init__(self, name: str, extra: pulumi.InputType[ExtraArgs]):
super().__init__("test:index:ExtraResource", name, {"extra": extra})
def translate_output_property(self, prop):
return CAMEL_TO_SNAKE_CASE_TABLE.get(prop) or prop
def translate_input_property(self, prop):
return SNAKE_TO_CAMEL_CASE_TABLE.get(prop) or prop
# Create a resource with input object.
res5 = ExtraResource("testres5", extra=ExtraArgs(first_value="foo", second_value=100))
# Create a resource using the output object of another resource.
res6 = ExtraResource("testres6", extra=ExtraArgs(
first_value=res5.extra.first_value,
second_value=res5.extra.second_value,
))
# Create a resource using the output object of another resource, accessing the output as a dict.
# Note: the output dict's keys are translated keys.
res7 = ExtraResource("testres7", extra=ExtraArgs(
first_value=res5.extra["first_value"],
second_value=res5.extra["second_value"],
))
# Create a resource using a dict as the input.
# Note: these are specified as snake_case, and the resource will translate to camelCase.
res8 = ExtraResource("testres8", extra={
"first_value": res5.extra["first_value"],
"second_value": res5.extra["second_value"],
})
# Now test some resources that use explicitly declared properties.
@pulumi.input_type
class SupplementaryArgs:
def __init__(self,
first_value: pulumi.Input[str],
second_value: Optional[pulumi.Input[float]] = None,
third: Optional[pulumi.Input[str]] = None,
fourth: Optional[pulumi.Input[str]] = None):
pulumi.set(self, "first_value", first_value)
pulumi.set(self, "second_value", second_value)
pulumi.set(self, "third", third)
pulumi.set(self, "fourth", fourth)
# Property with empty getter/setter bodies.
@property
@pulumi.getter(name="firstValue")
def first_value(self) -> pulumi.Input[str]:
...
@first_value.setter
def first_value(self, value: pulumi.Input[str]):
pulumi.set(self, "first_value", value)
# Property with explicitly specified getter/setter bodies.
@property
@pulumi.getter(name="secondValue")
def second_value(self) -> Optional[pulumi.Input[float]]:
return pulumi.get(self, "second_value")
@second_value.setter
def second_value(self, value: Optional[pulumi.Input[float]]):
pulumi.set(self, "second_value", value)
# Single word property name that doesn't require a name to be
# passed to the getter decorator.
@property
@pulumi.getter
def third(self) -> Optional[pulumi.Input[str]]:
...
@third.setter
def third(self, value: Optional[pulumi.Input[str]]):
...
# Another single word property name that doesn't require a name to be
# passed to the getter decorator, this time using the decorator with
# parens.
@property
@pulumi.getter()
def fourth(self) -> Optional[pulumi.Input[str]]:
...
@fourth.setter
def fourth(self, value: Optional[pulumi.Input[str]]):
...
@pulumi.output_type
class Supplementary(dict):
def __init__(self, first_value: str, second_value: Optional[float], third: str, fourth: str):
pulumi.set(self, "first_value", first_value)
pulumi.set(self, "second_value", second_value)
pulumi.set(self, "third", third)
pulumi.set(self, "fourth", fourth)
# Property with empty getter/setter bodies.
@property
@pulumi.getter(name="firstValue")
def first_value(self) -> str:
...
# Property with explicitly specified getter/setter bodies.
@property
@pulumi.getter(name="secondValue")
def second_value(self) -> Optional[float]:
return pulumi.get(self, "second_value")
# Single word property name that doesn't require a name to be
# passed to the getter decorator.
@property
@pulumi.getter
def third(self) -> str:
...
# Another single word property name that doesn't require a name to be
# passed to the getter decorator, this time using the decorator with
# parens.
@property
@pulumi.getter
def fourth(self) -> str:
...
class SupplementaryResource(pulumi.CustomResource):
supplementary: pulumi.Output[Supplementary]
def __init__(self, name: str, supplementary: pulumi.InputType[SupplementaryArgs]):
super().__init__("test:index:SupplementaryResource", name, {"supplementary": supplementary})
# Create a resource with input object.
res9 = SupplementaryResource("testres9", supplementary=SupplementaryArgs(
first_value="bar",
second_value=200,
third="third value",
fourth="fourth value",
))
# Create a resource using the output object of another resource.
res10 = SupplementaryResource("testres10", supplementary=SupplementaryArgs(
first_value=res9.supplementary.first_value,
second_value=res9.supplementary.second_value,
third=res9.supplementary.third,
fourth=res9.supplementary.fourth,
))
# Create a resource using the output object of another resource, accessing the output as a dict.
res11 = SupplementaryResource("testres11", supplementary=SupplementaryArgs(
first_value=res9.supplementary["first_value"],
second_value=res9.supplementary["second_value"],
third=res9.supplementary["third"],
fourth=res9.supplementary["fourth"],
))
# Create a resource using a dict as the input.
# Note: These are camel case (not snake_case) since the resource does not do any translation of
# property names.
res12 = SupplementaryResource("testres12", supplementary={
"firstValue": "bar",
"secondValue": 200,
"third": "third value",
"fourth": "fourth value",
})
# Now, test some resources that use property translations and explicitly declared properties.
@pulumi.input_type
class AncillaryArgs:
def __init__(self,
first_value: pulumi.Input[str],
second_value: Optional[pulumi.Input[float]] = None,
third: Optional[pulumi.Input[str]] = None,
fourth: Optional[pulumi.Input[str]] = None):
pulumi.set(self, "first_value", first_value)
pulumi.set(self, "second_value", second_value)
pulumi.set(self, "third", third)
pulumi.set(self, "fourth", fourth)
# Property with empty getter/setter bodies.
@property
@pulumi.getter(name="firstValue")
def first_value(self) -> pulumi.Input[str]:
...
@first_value.setter
def first_value(self, value: pulumi.Input[str]):
pulumi.set(self, "first_value", value)
# Property with explicitly specified getter/setter bodies.
@property
@pulumi.getter(name="secondValue")
def second_value(self) -> Optional[pulumi.Input[float]]:
return pulumi.get(self, "second_value")
@second_value.setter
def second_value(self, value: Optional[pulumi.Input[float]]):
pulumi.set(self, "second_value", value)
# Single word property name that doesn't require a name to be
# passed to the getter decorator.
@property
@pulumi.getter
def third(self) -> Optional[pulumi.Input[str]]:
...
@third.setter
def third(self, value: Optional[pulumi.Input[str]]):
...
# Another single word property name that doesn't require a name to be
# passed to the getter decorator, this time using the decorator with
# parens.
@property
@pulumi.getter()
def fourth(self) -> Optional[pulumi.Input[str]]:
...
@fourth.setter
def fourth(self, value: Optional[pulumi.Input[str]]):
...
@pulumi.output_type
class Ancillary(dict):
def __init__(self, first_value: str, second_value: Optional[float], third: str, fourth: str):
pulumi.set(self, "first_value", first_value)
pulumi.set(self, "second_value", second_value)
pulumi.set(self, "third", third)
pulumi.set(self, "fourth", fourth)
# Property with empty getter/setter bodies.
@property
@pulumi.getter(name="firstValue")
def first_value(self) -> str:
...
# Property with explicitly specified getter/setter bodies.
@property
@pulumi.getter(name="secondValue")
def second_value(self) -> Optional[float]:
return pulumi.get(self, "second_value")
# Single word property name that doesn't require a name to be
# passed to the getter decorator.
@property
@pulumi.getter
def third(self) -> str:
...
# Another single word property name that doesn't require a name to be
# passed to the getter decorator, this time using the decorator with
# parens.
@property
@pulumi.getter()
def fourth(self) -> str:
...
def _translate_property(self, prop):
return CAMEL_TO_SNAKE_CASE_TABLE.get(prop) or prop
class AncillaryResource(pulumi.CustomResource):
ancillary: pulumi.Output[Ancillary]
def __init__(self, name: str, ancillary: pulumi.InputType[AncillaryArgs]):
super().__init__("test:index:AncillaryResource", name, {"ancillary": ancillary})
def translate_output_property(self, prop):
return CAMEL_TO_SNAKE_CASE_TABLE.get(prop) or prop
def translate_input_property(self, prop):
return SNAKE_TO_CAMEL_CASE_TABLE.get(prop) or prop
# Create a resource with input object.
res13 = AncillaryResource("testres13", ancillary=AncillaryArgs(
first_value="baz",
second_value=500,
third="third value!",
fourth="fourth!",
))
# Create a resource using the output object of another resource.
res14 = AncillaryResource("testres14", ancillary=AncillaryArgs(
first_value=res13.ancillary.first_value,
second_value=res13.ancillary.second_value,
third=res13.ancillary.third,
fourth=res13.ancillary.fourth,
))
# Create a resource using the output object of another resource, accessing the output as a dict.
# Note: the output dict's keys are translated keys.
res15 = AncillaryResource("testres15", ancillary=AncillaryArgs(
first_value=res13.ancillary["first_value"],
second_value=res13.ancillary["second_value"],
third=res13.ancillary["third"],
fourth=res13.ancillary["fourth"],
))
# Create a resource using a dict as the input.
# Note: these are specified as snake_case, and the resource will translate to camelCase.
res16 = AncillaryResource("testres16", ancillary={
"first_value": "baz",
"second_value": 500,
"third": "third value!",
"fourth": "fourth!",
})

62
sdk/python/lib/test/langhost/types/test_types.py Normal file
View file

@ -0,0 +1,62 @@
# Copyright 2016-2020, 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.
from os import path
from ..util import LanghostTest
class TestTypes(LanghostTest):
def test_types(self):
self.run_test(
program=path.join(self.base_path(), "types"),
expected_resource_count=16)
def register_resource(self, ctx, dry_run, ty, name, _resource,
_dependencies, _parent, _custom, _protect,
_provider, _property_deps, _delete_before_replace, _ignore_changes, version):
if name in ["testres", "testres2", "testres3", "testres4"]:
self.assertIn("additional", _resource)
self.assertEqual({
"firstValue": "hello",
"secondValue": 42,
}, _resource["additional"])
elif name in ["testres5", "testres6", "testres7", "testres8"]:
self.assertIn("extra", _resource)
self.assertEqual({
"firstValue": "foo",
"secondValue": 100,
}, _resource["extra"])
elif name in ["testres9", "testres10", "testres11", "testres12"]:
self.assertIn("supplementary", _resource)
self.assertEqual({
"firstValue": "bar",
"secondValue": 200,
"third": "third value",
"fourth": "fourth value",
}, _resource["supplementary"])
elif name in ["testres13", "testres14", "testres15", "testres16"]:
self.assertIn("ancillary", _resource)
self.assertEqual({
"firstValue": "baz",
"secondValue": 500,
"third": "third value!",
"fourth": "fourth!",
}, _resource["ancillary"])
else:
self.fail(f"unknown resource: {name}")
return {
"urn": self.make_urn(ty, name),
"id": name,
"object": _resource,
}

49
sdk/python/lib/test/test_next_serialize.py
View file

@ -18,7 +18,7 @@ from typing import Any, Optional
from google.protobuf import struct_pb2
from pulumi.resource import CustomResource
from pulumi.runtime import rpc, known_types, settings
from pulumi.output import Output, UNKNOWN
from pulumi import Input, Output, UNKNOWN, input_type
from pulumi.asset import (
FileAsset,
RemoteAsset,
@ -27,6 +27,7 @@ from pulumi.asset import (
FileArchive,
RemoteArchive
)
import pulumi
class FakeCustomResource:
@ -911,3 +912,49 @@ class DeserializationTests(unittest.TestCase):
self.assertEqual(val["listWithMap"][rpc._special_sig_key], rpc._special_secret_sig)
self.assertEqual(val["listWithMap"]["value"][0]["regular"], "a normal value")
self.assertEqual(val["listWithMap"]["value"][0]["secret"], "a secret value")
@input_type
class FooArgs:
first_arg: Input[str] = pulumi.property("firstArg")
second_arg: Optional[Input[float]] = pulumi.property("secondArg")
def __init__(self, first_arg: Input[str], second_arg: Optional[Input[float]]=None):
pulumi.set(self, "first_arg", first_arg)
pulumi.set(self, "second_arg", second_arg)
@input_type
class BarArgs:
tag_args: Input[dict] = pulumi.property("tagArgs")
def __init__(self, tag_args: Input[dict]):
pulumi.set(self, "tag_args", tag_args)
class InputTypeSerializationTests(unittest.TestCase):
@async_test
async def test_simple_input_type(self):
it = FooArgs(first_arg="hello", second_arg=42)
prop = await rpc.serialize_property(it, [])
self.assertDictEqual(prop, {"firstArg": "hello", "secondArg": 42})
@async_test
async def test_input_type_with_dict_property(self):
def transformer(prop: str) -> str:
return {
"tag_args": "a",
"tagArgs": "b",
"foo_bar": "c",
}.get(prop) or prop
it = BarArgs({"foo_bar": "hello", "foo_baz": "world"})
prop = await rpc.serialize_property(it, [], transformer)
# Input type keys are not be transformed, but keys of nested
# dicts are still transformed.
self.assertDictEqual(prop, {
"tagArgs": {
"c": "hello",
"foo_baz": "world",
},
})

552
sdk/python/lib/test/test_translate_output_properties.py Normal file
View file

@ -0,0 +1,552 @@
# Copyright 2016-2020, 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 unittest
from typing import Dict, List, Optional
from pulumi.runtime import rpc
import pulumi
camel_case_to_snake_case = {
"firstArg": "first_arg",
"secondArg": "second_arg",
}
def translate_output_property(prop: str) -> str:
return camel_case_to_snake_case.get(prop) or prop
@pulumi.output_type
class Foo(dict):
first_arg: str = pulumi.property("firstArg")
second_arg: float = pulumi.property("secondArg")
def _translate_property(self, prop: str) -> str:
return camel_case_to_snake_case.get(prop) or prop
@pulumi.output_type
class Bar(dict):
third_arg: Foo = pulumi.property("thirdArg")
third_optional_arg: Optional[Foo] = pulumi.property("thirdOptionalArg")
fourth_arg: Dict[str, Foo] = pulumi.property("fourthArg")
fourth_optional_arg: Dict[str, Optional[Foo]] = pulumi.property("fourthOptionalArg")
fifth_arg: List[Foo] = pulumi.property("fifthArg")
fifth_optional_arg: List[Optional[Foo]] = pulumi.property("fifthOptionalArg")
sixth_arg: Dict[str, List[Foo]] = pulumi.property("sixthArg")
sixth_optional_arg: Dict[str, Optional[List[Foo]]] = pulumi.property("sixthOptionalArg")
sixth_optional_optional_arg: Dict[str, Optional[List[Optional[Foo]]]] = pulumi.property("sixthOptionalOptionalArg")
seventh_arg: List[Dict[str, Foo]] = pulumi.property("seventhArg")
seventh_optional_arg: List[Optional[Dict[str, Foo]]] = pulumi.property("seventhOptionalArg")
seventh_optional_optional_arg: List[Optional[Dict[str, Optional[Foo]]]] = pulumi.property("seventhOptionalOptionalArg")
eighth_arg: List[Dict[str, List[Foo]]] = pulumi.property("eighthArg")
eighth_optional_arg: List[Optional[Dict[str, List[Foo]]]] = pulumi.property("eighthOptionalArg")
eighth_optional_optional_arg: List[Optional[Dict[str, Optional[List[Foo]]]]] = pulumi.property("eighthOptionalOptionalArg")
eighth_optional_optional_optional_arg: List[Optional[Dict[str, Optional[List[Optional[Foo]]]]]] = pulumi.property("eighthOptionalOptionalOptionalArg")
def _translate_property(self, prop: str) -> str:
return camel_case_to_snake_case.get(prop) or prop
@pulumi.output_type
class BarDeclared(dict):
def __init__(self,
third_arg: Foo,
third_optional_arg: Optional[Foo],
fourth_arg: Dict[str, Foo],
fourth_optional_arg: Dict[str, Optional[Foo]],
fifth_arg: List[Foo],
fifth_optional_arg: List[Optional[Foo]],
sixth_arg: Dict[str, List[Foo]],
sixth_optional_arg: Dict[str, Optional[List[Foo]]],
sixth_optional_optional_arg: Dict[str, Optional[List[Optional[Foo]]]],
seventh_arg: List[Dict[str, Foo]],
seventh_optional_arg: List[Optional[Dict[str, Foo]]],
seventh_optional_optional_arg: List[Optional[Dict[str, Optional[Foo]]]],
eighth_arg: List[Dict[str, List[Foo]]],
eighth_optional_arg: List[Optional[Dict[str, List[Foo]]]],
eighth_optional_optional_arg: List[Optional[Dict[str, Optional[List[Foo]]]]],
eighth_optional_optional_optional_arg: List[Optional[Dict[str, Optional[List[Optional[Foo]]]]]]):
pulumi.set(self, "third_arg", third_arg)
pulumi.set(self, "third_optional_arg", third_optional_arg)
pulumi.set(self, "fourth_arg", fourth_arg)
pulumi.set(self, "fourth_optional_arg", fourth_optional_arg)
pulumi.set(self, "fifth_arg", fifth_arg)
pulumi.set(self, "fifth_optional_arg", fifth_optional_arg)
pulumi.set(self, "sixth_arg", sixth_arg)
pulumi.set(self, "sixth_optional_arg", sixth_optional_arg)
pulumi.set(self, "sixth_optional_optional_arg", sixth_optional_optional_arg)
pulumi.set(self, "seventh_arg", seventh_arg)
pulumi.set(self, "seventh_optional_arg", seventh_optional_arg)
pulumi.set(self, "seventh_optional_optional_arg", seventh_optional_optional_arg)
pulumi.set(self, "eighth_arg", eighth_arg)
pulumi.set(self, "eighth_optional_arg", eighth_optional_arg)
pulumi.set(self, "eighth_optional_optional_arg", eighth_optional_optional_arg)
pulumi.set(self, "eighth_optional_optional_optional_arg", eighth_optional_optional_optional_arg)
@property
@pulumi.getter(name="thirdArg")
def third_arg(self) -> Foo:
...
@property
@pulumi.getter(name="thirdOptionalArg")
def third_optional_arg(self) -> Optional[Foo]:
...
@property
@pulumi.getter(name="fourthArg")
def fourth_arg(self) -> Dict[str, Foo]:
...
@property
@pulumi.getter(name="fourthOptionalArg")
def fourth_optional_arg(self) -> Dict[str, Optional[Foo]]:
...
@property
@pulumi.getter(name="fifthArg")
def fifth_arg(self) -> List[Foo]:
...
@property
@pulumi.getter(name="fifthOptionalArg")
def fifth_optional_arg(self) -> List[Optional[Foo]]:
...
@property
@pulumi.getter(name="sixthArg")
def sixth_arg(self) -> Dict[str, List[Foo]]:
...
@property
@pulumi.getter(name="sixthOptionalArg")
def sixth_optional_arg(self) -> Dict[str, Optional[List[Foo]]]:
...
@property
@pulumi.getter(name="sixthOptionalOptionalArg")
def sixth_optional_optional_arg(self) -> Dict[str, Optional[List[Optional[Foo]]]]:
...
@property
@pulumi.getter(name="seventhArg")
def seventh_arg(self) -> List[Dict[str, Foo]]:
...
@property
@pulumi.getter(name="seventhOptionalArg")
def seventh_optional_arg(self) -> List[Optional[Dict[str, Foo]]]:
...
@property
@pulumi.getter(name="seventhOptionalOptionalArg")
def seventh_optional_optional_arg(self) -> List[Optional[Dict[str, Optional[Foo]]]]:
...
@property
@pulumi.getter(name="eighthArg")
def eighth_arg(self) -> List[Dict[str, List[Foo]]]:
...
@property
@pulumi.getter(name="eighthOptionalArg")
def eighth_optional_arg(self) -> List[Optional[Dict[str, List[Foo]]]]:
...
@property
@pulumi.getter(name="eighthOptionalOptionalArg")
def eighth_optional_optional_arg(self) -> List[Optional[Dict[str, Optional[List[Foo]]]]]:
...
@property
@pulumi.getter(name="eighthOptionalOptionalOptionalArg")
def eighth_optional_optional_optional_arg(self) -> List[Optional[Dict[str, Optional[List[Optional[Foo]]]]]]:
...
def _translate_property(self, prop: str) -> str:
return camel_case_to_snake_case.get(prop) or prop
@pulumi.output_type
class InvalidTypeStr(dict):
value: str = pulumi.property("value")
@pulumi.output_type
class InvalidTypeDeclaredStr(dict):
def __init__(self, value: str):
pulumi.set(self, "value", value)
@property
@pulumi.getter
def value(self) -> str:
...
@pulumi.output_type
class InvalidTypeOptionalStr(dict):
value: Optional[str] = pulumi.property("value")
@pulumi.output_type
class InvalidTypeDeclaredOptionalStr(dict):
def __init__(self, value: Optional[str]):
pulumi.set(self, "value", value)
@property
@pulumi.getter
def value(self) -> Optional[str]:
...
@pulumi.output_type
class InvalidTypeDictStr(dict):
value: Dict[str, str] = pulumi.property("value")
@pulumi.output_type
class InvalidTypeDeclaredDictStr(dict):
def __init__(self, value: Dict[str, str]):
pulumi.set(self, "value", value)
@property
@pulumi.getter
def value(self) -> Dict[str, str]:
...
@pulumi.output_type
class InvalidTypeOptionalDictStr(dict):
value: Optional[Dict[str, str]] = pulumi.property("value")
@pulumi.output_type
class InvalidTypeDeclaredOptionalDictStr(dict):
def __init__(self, value: Optional[Dict[str, str]]):
pulumi.set(self, "value", value)
@property
@pulumi.getter
def value(self) -> Optional[Dict[str, str]]:
...
@pulumi.output_type
class InvalidTypeDictOptionalStr(dict):
value: Dict[str, Optional[str]] = pulumi.property("value")
@pulumi.output_type
class InvalidTypeDeclaredDictOptionalStr(dict):
def __init__(self, value: Dict[str, Optional[str]]):
pulumi.set(self, "value", value)
@property
@pulumi.getter
def value(self) -> Dict[str, Optional[str]]:
...
@pulumi.output_type
class InvalidTypeOptionalDictOptionalStr(dict):
value: Optional[Dict[str, Optional[str]]] = pulumi.property("value")
@pulumi.output_type
class InvalidTypeDeclaredOptionalDictOptionalStr(dict):
def __init__(self, value: Optional[Dict[str, Optional[str]]]):
pulumi.set(self, "value", value)
@property
@pulumi.getter
def value(self) -> Optional[Dict[str, Optional[str]]]:
...
@pulumi.output_type
class InvalidTypeListStr(dict):
value: List[str] = pulumi.property("value")
@pulumi.output_type
class InvalidTypeDeclaredListStr(dict):
def __init__(self, value: List[str]):
pulumi.set(self, "value", value)
@property
@pulumi.getter
def value(self) -> List[str]:
...
@pulumi.output_type
class InvalidTypeOptionalListStr(dict):
value: Optional[List[str]] = pulumi.property("value")
@pulumi.output_type
class InvalidTypeDeclaredOptionalListStr(dict):
def __init__(self, value: Optional[List[str]]):
pulumi.set(self, "value", value)
@property
@pulumi.getter
def value(self) -> Optional[List[str]]:
...
@pulumi.output_type
class InvalidTypeListOptionalStr(dict):
value: List[Optional[str]] = pulumi.property("value")
@pulumi.output_type
class InvalidTypeDeclaredListOptionalStr(dict):
def __init__(self, value: List[Optional[str]]):
pulumi.set(self, "value", value)
@property
@pulumi.getter
def value(self) -> List[Optional[str]]:
...
@pulumi.output_type
class InvalidTypeOptionalListOptionalStr(dict):
value: Optional[List[Optional[str]]] = pulumi.property("value")
@pulumi.output_type
class InvalidTypeDeclaredOptionalListOptionalStr(dict):
def __init__(self, value: Optional[List[Optional[str]]]):
pulumi.set(self, "value", value)
@property
@pulumi.getter
def value(self) -> Optional[List[Optional[str]]]:
...
class TranslateOutputPropertiesTests(unittest.TestCase):
def test_translate(self):
output = {
"firstArg": "hello",
"secondArg": 42,
}
result = rpc.translate_output_properties(output, translate_output_property, Foo)
self.assertIsInstance(result, Foo)
self.assertEqual(result.first_arg, "hello")
self.assertEqual(result["first_arg"], "hello")
self.assertEqual(result.second_arg, 42)
self.assertEqual(result["second_arg"], 42)
def test_nested_types(self):
def assertFoo(val, first_arg, second_arg):
self.assertIsInstance(val, Foo)
self.assertEqual(val.first_arg, first_arg)
self.assertEqual(val["first_arg"], first_arg)
self.assertEqual(val.second_arg, second_arg)
self.assertEqual(val["second_arg"], second_arg)
output = {
"thirdArg": {
"firstArg": "hello",
"secondArg": 42,
},
"thirdOptionalArg": {
"firstArg": "hello-opt",
"secondArg": 142,
},
"fourthArg": {
"foo": {
"firstArg": "hi",
"secondArg": 41,
},
},
"fourthOptionalArg": {
"foo": {
"firstArg": "hi-opt",
"secondArg": 141,
},
},
"fifthArg": [{
"firstArg": "bye",
"secondArg": 40,
}],
"fifthOptionalArg": [{
"firstArg": "bye-opt",
"secondArg": 140,
}],
"sixthArg": {
"bar": [{
"firstArg": "goodbye",
"secondArg": 39,
}],
},
"sixthOptionalArg": {
"bar": [{
"firstArg": "goodbye-opt",
"secondArg": 139,
}],
},
"sixthOptionalOptionalArg": {
"bar": [{
"firstArg": "goodbye-opt-opt",
"secondArg": 1139,
}],
},
"seventhArg": [{
"baz": {
"firstArg": "adios",
"secondArg": 38,
},
}],
"seventhOptionalArg": [{
"baz": {
"firstArg": "adios-opt",
"secondArg": 138,
},
}],
"seventhOptionalOptionalArg": [{
"baz": {
"firstArg": "adios-opt-opt",
"secondArg": 1138,
},
}],
"eighthArg": [{
"blah": [{
"firstArg": "farewell",
"secondArg": 37,
}],
}],
"eighthOptionalArg": [{
"blah": [{
"firstArg": "farewell-opt",
"secondArg": 137,
}],
}],
"eighthOptionalOptionalArg": [{
"blah": [{
"firstArg": "farewell-opt-opt",
"secondArg": 1137,
}],
}],
"eighthOptionalOptionalOptionalArg": [{
"blah": [{
"firstArg": "farewell-opt-opt-opt",
"secondArg": 11137,
}],
}],
}
def convert_properties_to_secrets(output: dict) -> dict:
return {k: {rpc._special_sig_key: rpc._special_secret_sig, "value": v } for k, v in output.items()}
def run_test(output: dict):
result = rpc.translate_output_properties(output, translate_output_property, typ)
self.assertIsInstance(result, typ)
self.assertIs(result.third_arg, result["thirdArg"])
assertFoo(result.third_arg, "hello", 42)
self.assertIs(result.third_optional_arg, result["thirdOptionalArg"])
assertFoo(result.third_optional_arg, "hello-opt", 142)
self.assertIs(result.fourth_arg, result["fourthArg"])
assertFoo(result.fourth_arg["foo"], "hi", 41)
self.assertIs(result.fourth_optional_arg, result["fourthOptionalArg"])
assertFoo(result.fourth_optional_arg["foo"], "hi-opt", 141)
self.assertIs(result.fifth_arg, result["fifthArg"])
assertFoo(result.fifth_arg[0], "bye", 40)
self.assertIs(result.fifth_optional_arg, result["fifthOptionalArg"])
assertFoo(result.fifth_optional_arg[0], "bye-opt", 140)
self.assertIs(result.sixth_arg, result["sixthArg"])
assertFoo(result.sixth_arg["bar"][0], "goodbye", 39)
self.assertIs(result.sixth_optional_arg, result["sixthOptionalArg"])
assertFoo(result.sixth_optional_arg["bar"][0], "goodbye-opt", 139)
self.assertIs(result.sixth_optional_optional_arg, result["sixthOptionalOptionalArg"])
assertFoo(result.sixth_optional_optional_arg["bar"][0], "goodbye-opt-opt", 1139)
self.assertIs(result.seventh_arg, result["seventhArg"])
assertFoo(result.seventh_arg[0]["baz"], "adios", 38)
self.assertIs(result.seventh_optional_arg, result["seventhOptionalArg"])
assertFoo(result.seventh_optional_arg[0]["baz"], "adios-opt", 138)
self.assertIs(result.seventh_optional_optional_arg, result["seventhOptionalOptionalArg"])
assertFoo(result.seventh_optional_optional_arg[0]["baz"], "adios-opt-opt", 1138)
self.assertIs(result.eighth_arg, result["eighthArg"])
assertFoo(result.eighth_arg[0]["blah"][0], "farewell", 37)
self.assertIs(result.eighth_optional_arg, result["eighthOptionalArg"])
assertFoo(result.eighth_optional_arg[0]["blah"][0], "farewell-opt", 137)
self.assertIs(result.eighth_optional_optional_arg, result["eighthOptionalOptionalArg"])
assertFoo(result.eighth_optional_optional_arg[0]["blah"][0], "farewell-opt-opt", 1137)
self.assertIs(result.eighth_optional_optional_optional_arg, result["eighthOptionalOptionalOptionalArg"])
assertFoo(result.eighth_optional_optional_optional_arg[0]["blah"][0], "farewell-opt-opt-opt", 11137)
for typ in [Bar, BarDeclared]:
run_test(output)
run_test(convert_properties_to_secrets(output))
def test_nested_types_raises(self):
dict_value = {
"firstArg": "hello",
"secondArg": 42,
}
list_value = ["hello"]
tests = [
(InvalidTypeStr, dict_value),
(InvalidTypeDeclaredStr, dict_value),
(InvalidTypeOptionalStr, dict_value),
(InvalidTypeDeclaredOptionalStr, dict_value),
(InvalidTypeStr, list_value),
(InvalidTypeDeclaredStr, list_value),
(InvalidTypeOptionalStr, list_value),
(InvalidTypeDeclaredOptionalStr, list_value),
(InvalidTypeDictStr, {"foo": dict_value}),
(InvalidTypeDeclaredDictStr, {"foo": dict_value}),
(InvalidTypeOptionalDictStr, {"foo": dict_value}),
(InvalidTypeDeclaredOptionalDictStr, {"foo": dict_value}),
(InvalidTypeDictOptionalStr, {"foo": dict_value}),
(InvalidTypeDeclaredDictOptionalStr, {"foo": dict_value}),
(InvalidTypeOptionalDictOptionalStr, {"foo": dict_value}),
(InvalidTypeDeclaredOptionalDictOptionalStr, {"foo": dict_value}),
(InvalidTypeDictStr, {"foo": list_value}),
(InvalidTypeDeclaredDictStr, {"foo": list_value}),
(InvalidTypeOptionalDictStr, {"foo": list_value}),
(InvalidTypeDeclaredOptionalDictStr, {"foo": list_value}),
(InvalidTypeDictOptionalStr, {"foo": list_value}),
(InvalidTypeDeclaredDictOptionalStr, {"foo": list_value}),
(InvalidTypeOptionalDictOptionalStr, {"foo": list_value}),
(InvalidTypeDeclaredOptionalDictOptionalStr, {"foo": list_value}),
(InvalidTypeListStr, [dict_value]),
(InvalidTypeDeclaredListStr, [dict_value]),
(InvalidTypeOptionalListStr, [dict_value]),
(InvalidTypeDeclaredOptionalListStr, [dict_value]),
(InvalidTypeListOptionalStr, [dict_value]),
(InvalidTypeDeclaredListOptionalStr, [dict_value]),
(InvalidTypeOptionalListOptionalStr, [dict_value]),
(InvalidTypeDeclaredOptionalListOptionalStr, [dict_value]),
(InvalidTypeListStr, [list_value]),
(InvalidTypeDeclaredListStr, [list_value]),
(InvalidTypeOptionalListStr, [list_value]),
(InvalidTypeDeclaredOptionalListStr, [list_value]),
(InvalidTypeListOptionalStr, [list_value]),
(InvalidTypeDeclaredListOptionalStr, [list_value]),
(InvalidTypeOptionalListOptionalStr, [list_value]),
(InvalidTypeDeclaredOptionalListOptionalStr, [list_value]),
]
for typ, value in tests:
outputs = [
{"value": value},
{"value": {rpc._special_sig_key: rpc._special_secret_sig, "value": value}},
]
for output in outputs:
with self.assertRaises(AssertionError):
rpc.translate_output_properties(output, translate_output_property, typ)

162
sdk/python/lib/test/test_types_input_type.py Normal file
View file

@ -0,0 +1,162 @@
# Copyright 2016-2020, 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 unittest
from typing import Optional
import pulumi
import pulumi._types as _types
@pulumi.input_type
class MySimpleInputType:
first_value: pulumi.Input[str] = pulumi.property("firstValue")
second_value: Optional[pulumi.Input[float]] = pulumi.property("secondValue", default=None)
@pulumi.input_type
class MyInputType:
first_value: pulumi.Input[str] = pulumi.property("firstValue")
second_value: Optional[pulumi.Input[float]] = pulumi.property("secondValue")
def __init__(self,
first_value: pulumi.Input[str],
second_value: Optional[pulumi.Input[float]] = None):
pulumi.set(self, "first_value", first_value)
pulumi.set(self, "second_value", second_value)
@pulumi.input_type
class MyDeclaredPropertiesInputType:
def __init__(self,
first_value: pulumi.Input[str],
second_value: Optional[pulumi.Input[float]] = None):
pulumi.set(self, "first_value", first_value)
pulumi.set(self, "second_value", second_value)
# Property with empty getter/setter bodies.
@property
@pulumi.getter(name="firstValue")
def first_value(self) -> pulumi.Input[str]:
"""First value docstring."""
...
@first_value.setter
def first_value(self, value: pulumi.Input[str]):
...
# Property with implementations.
@property
@pulumi.getter(name="secondValue")
def second_value(self) -> Optional[pulumi.Input[float]]:
"""Second value docstring."""
return pulumi.get(self, "second_value")
@second_value.setter
def second_value(self, value: Optional[pulumi.Input[float]]):
pulumi.set(self, "second_value", value)
class InputTypeTests(unittest.TestCase):
def test_decorator_raises(self):
with self.assertRaises(AssertionError) as cm:
@pulumi.input_type
@pulumi.input_type
class Foo:
pass
with self.assertRaises(AssertionError) as cm:
@pulumi.input_type
@pulumi.output_type
class Bar:
pass
def test_is_input_type(self):
types = [
MyInputType,
MyDeclaredPropertiesInputType,
]
for typ in types:
self.assertTrue(_types.is_input_type(typ))
self.assertEqual(True, typ._pulumi_input_type)
def test_input_type(self):
types = [
(MySimpleInputType, False),
(MyInputType, False),
(MyDeclaredPropertiesInputType, True),
]
for typ, has_doc in types:
t = typ(first_value="hello", second_value=42)
self.assertEqual("hello", t.first_value)
self.assertEqual(42, t.second_value)
t.first_value = "world"
self.assertEqual("world", t.first_value)
t.second_value = 500
self.assertEqual(500, t.second_value)
first = typ.first_value
self.assertIsInstance(first, property)
self.assertTrue(callable(first.fget))
self.assertEqual("first_value", first.fget.__name__)
self.assertEqual({"return": pulumi.Input[str]}, first.fget.__annotations__)
if has_doc:
self.assertEqual("First value docstring.", first.fget.__doc__)
self.assertEqual("firstValue", first.fget._pulumi_name)
self.assertTrue(callable(first.fset))
self.assertEqual("first_value", first.fset.__name__)
self.assertEqual({"value": pulumi.Input[str]}, first.fset.__annotations__)
second = typ.second_value
self.assertIsInstance(second, property)
self.assertTrue(callable(second.fget))
self.assertEqual("second_value", second.fget.__name__)
self.assertEqual({"return": Optional[pulumi.Input[float]]}, second.fget.__annotations__)
if has_doc:
self.assertEqual("Second value docstring.", second.fget.__doc__)
self.assertEqual("secondValue", second.fget._pulumi_name)
self.assertTrue(callable(second.fset))
self.assertEqual("second_value", second.fset.__name__)
self.assertEqual({"value": Optional[pulumi.Input[float]]}, second.fset.__annotations__)
self.assertEqual({
"firstValue": "world",
"secondValue": 500,
}, _types.input_type_to_dict(t))
self.assertTrue(hasattr(t, "__eq__"))
self.assertTrue(t.__eq__(t))
self.assertTrue(t == t)
self.assertFalse(t != t)
self.assertFalse(t == "not equal")
t2 = typ(first_value="world", second_value=500)
self.assertTrue(t.__eq__(t2))
self.assertTrue(t == t2)
self.assertFalse(t != t2)
self.assertEqual({
"firstValue": "world",
"secondValue": 500,
}, _types.input_type_to_dict(t2))
t3 = typ(first_value="foo", second_value=1)
self.assertFalse(t.__eq__(t3))
self.assertFalse(t == t3)
self.assertTrue(t != t3)
self.assertEqual({
"firstValue": "foo",
"secondValue": 1,
}, _types.input_type_to_dict(t3))

242
sdk/python/lib/test/test_types_output_type.py Normal file
View file

@ -0,0 +1,242 @@
# Copyright 2016-2020, 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 unittest
from typing import Optional
import pulumi
import pulumi._types as _types
CAMEL_TO_SNAKE_CASE_TABLE = {
"firstValue": "first_value",
"secondValue": "second_value",
}
@pulumi.output_type
class MyOutputType:
first_value: str = pulumi.property("firstValue")
second_value: Optional[float] = pulumi.property("secondValue", default=None)
@pulumi.output_type
class MyOutputTypeDict(dict):
first_value: str = pulumi.property("firstValue")
second_value: Optional[float] = pulumi.property("secondValue", default=None)
@pulumi.output_type
class MyOutputTypeTranslated:
first_value: str = pulumi.property("firstValue")
second_value: Optional[float] = pulumi.property("secondValue", default=None)
def _translate_property(self, prop):
return CAMEL_TO_SNAKE_CASE_TABLE.get(prop) or prop
@pulumi.output_type
class MyOutputTypeDictTranslated(dict):
first_value: str = pulumi.property("firstValue")
second_value: Optional[float] = pulumi.property("secondValue", default=None)
def _translate_property(self, prop):
return CAMEL_TO_SNAKE_CASE_TABLE.get(prop) or prop
@pulumi.output_type
class MyDeclaredPropertiesOutputType:
def __init__(self, first_value: str, second_value: Optional[float] = None):
pulumi.set(self, "first_value", first_value)
if second_value is not None:
pulumi.set(self, "second_value", second_value)
# Property with empty body.
@property
@pulumi.getter(name="firstValue")
def first_value(self) -> str:
"""First value docstring."""
...
# Property with implementation.
@property
@pulumi.getter(name="secondValue")
def second_value(self) -> Optional[float]:
"""Second value docstring."""
return pulumi.get(self, "second_value")
@pulumi.output_type
class MyDeclaredPropertiesOutputTypeDict(dict):
def __init__(self, first_value: str, second_value: Optional[float] = None):
pulumi.set(self, "first_value", first_value)
if second_value is not None:
pulumi.set(self, "second_value", second_value)
# Property with empty body.
@property
@pulumi.getter(name="firstValue")
def first_value(self) -> str:
"""First value docstring."""
...
# Property with implementation.
@property
@pulumi.getter(name="secondValue")
def second_value(self) -> Optional[float]:
"""Second value docstring."""
return pulumi.get(self, "second_value")
@pulumi.output_type
class MyDeclaredPropertiesOutputTypeTranslated:
def __init__(self, first_value: str, second_value: Optional[float] = None):
pulumi.set(self, "first_value", first_value)
if second_value is not None:
pulumi.set(self, "second_value", second_value)
# Property with empty body.
@property
@pulumi.getter(name="firstValue")
def first_value(self) -> str:
"""First value docstring."""
...
# Property with implementation.
@property
@pulumi.getter(name="secondValue")
def second_value(self) -> Optional[float]:
"""Second value docstring."""
return pulumi.get(self, "second_value")
def _translate_property(self, prop):
return CAMEL_TO_SNAKE_CASE_TABLE.get(prop) or prop
@pulumi.output_type
class MyDeclaredPropertiesOutputTypeDictTranslated(dict):
def __init__(self, first_value: str, second_value: Optional[float] = None):
pulumi.set(self, "first_value", first_value)
if second_value is not None:
pulumi.set(self, "second_value", second_value)
# Property with empty body.
@property
@pulumi.getter(name="firstValue")
def first_value(self) -> str:
"""First value docstring."""
...
# Property with implementation.
@property
@pulumi.getter(name="secondValue")
def second_value(self) -> Optional[float]:
"""Second value docstring."""
return pulumi.get(self, "second_value")
def _translate_property(self, prop):
return CAMEL_TO_SNAKE_CASE_TABLE.get(prop) or prop
class InputTypeTests(unittest.TestCase):
def test_decorator_raises(self):
with self.assertRaises(AssertionError) as cm:
@pulumi.output_type
@pulumi.input_type
class Foo:
pass
with self.assertRaises(AssertionError) as cm:
@pulumi.output_type
@pulumi.input_type
class Bar:
pass
def test_is_output_type(self):
types = [
MyOutputType,
MyOutputTypeDict,
MyOutputTypeTranslated,
MyOutputTypeDictTranslated,
MyDeclaredPropertiesOutputType,
MyDeclaredPropertiesOutputTypeDict,
MyDeclaredPropertiesOutputTypeTranslated,
MyDeclaredPropertiesOutputTypeDictTranslated,
]
for typ in types:
self.assertTrue(_types.is_output_type(typ))
self.assertEqual(True, typ._pulumi_output_type)
self.assertTrue(hasattr(typ, "__init__"))
def test_output_type_types(self):
self.assertEqual({
"firstValue": str,
"secondValue": float,
}, _types.output_type_types(MyOutputType))
def test_output_type(self):
types = [
(MyOutputType, False),
(MyOutputTypeDict, False),
(MyOutputTypeTranslated, False),
(MyOutputTypeDictTranslated, False),
(MyDeclaredPropertiesOutputType, True),
(MyDeclaredPropertiesOutputTypeDict, True),
(MyDeclaredPropertiesOutputTypeTranslated, True),
(MyDeclaredPropertiesOutputTypeDictTranslated, True),
]
for typ, has_doc in types:
self.assertTrue(hasattr(typ, "__init__"))
t = _types.output_type_from_dict(typ, {"firstValue": "hello", "secondValue": 42})
self.assertEqual("hello", t.first_value)
self.assertEqual(42, t.second_value)
if isinstance(t, dict):
self.assertEqual("hello", t["first_value"])
self.assertEqual(42, t["second_value"])
first = typ.first_value
self.assertIsInstance(first, property)
self.assertTrue(callable(first.fget))
self.assertEqual("first_value", first.fget.__name__)
self.assertEqual({"return": str}, first.fget.__annotations__)
if has_doc:
self.assertEqual("First value docstring.", first.fget.__doc__)
self.assertEqual("firstValue", first.fget._pulumi_name)
second = typ.second_value
self.assertIsInstance(second, property)
self.assertTrue(callable(second.fget))
self.assertEqual("second_value", second.fget.__name__)
self.assertEqual({"return": Optional[float]}, second.fget.__annotations__)
if has_doc:
self.assertEqual("Second value docstring.", second.fget.__doc__)
self.assertEqual("secondValue", second.fget._pulumi_name)
self.assertTrue(hasattr(t, "__eq__"))
self.assertTrue(t.__eq__(t))
self.assertTrue(t == t)
self.assertFalse(t != t)
self.assertFalse(t == "not equal")
t2 = _types.output_type_from_dict(typ, {"firstValue": "hello", "secondValue": 42})
self.assertTrue(t.__eq__(t2))
self.assertTrue(t == t2)
self.assertFalse(t != t2)
if isinstance(t2, dict):
self.assertEqual("hello", t2["first_value"])
self.assertEqual(42, t2["second_value"])
t3 = _types.output_type_from_dict(typ, {"firstValue": "foo", "secondValue": 1})
self.assertFalse(t.__eq__(t3))
self.assertFalse(t == t3)
self.assertTrue(t != t3)
if isinstance(t3, dict):
self.assertEqual("foo", t3["first_value"])
self.assertEqual(1, t3["second_value"])

104
sdk/python/lib/test/test_types_resource_types.py Normal file
View file

@ -0,0 +1,104 @@
# Copyright 2016-2020, 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 unittest
from pulumi._types import resource_types
import pulumi
class Resource1(pulumi.Resource):
pass
class Resource2(pulumi.Resource):
foo: pulumi.Output[str]
class Resource3(pulumi.Resource):
nested: pulumi.Output['Nested']
class Resource4(pulumi.Resource):
nested_value: pulumi.Output['Nested'] = pulumi.property("nestedValue")
class Resource5(pulumi.Resource):
@property
@pulumi.getter
def foo(self) -> pulumi.Output[str]:
...
class Resource6(pulumi.Resource):
@property
@pulumi.getter
def nested(self) -> pulumi.Output['Nested']:
...
class Resource7(pulumi.Resource):
@property
@pulumi.getter(name="nestedValue")
def nested_value(self) -> pulumi.Output['Nested']:
...
class Resource8(pulumi.Resource):
foo: pulumi.Output
class Resource9(pulumi.Resource):
@property
@pulumi.getter
def foo(self) -> pulumi.Output:
...
class Resource10(pulumi.Resource):
foo: str
class Resource11(pulumi.Resource):
@property
@pulumi.getter
def foo(self) -> str:
...
class Resource12(pulumi.Resource):
@property
@pulumi.getter
def foo(self):
...
@pulumi.output_type
class Nested:
first: str
second: str
class ResourceTypesTests(unittest.TestCase):
def test_resource_types(self):
self.assertEqual({}, resource_types(Resource1))
self.assertEqual({"foo": str}, resource_types(Resource2))
self.assertEqual({"nested": Nested}, resource_types(Resource3))
self.assertEqual({"nestedValue": Nested}, resource_types(Resource4))
self.assertEqual({"foo": str}, resource_types(Resource5))
self.assertEqual({"nested": Nested}, resource_types(Resource6))
self.assertEqual({"nestedValue": Nested}, resource_types(Resource7))
# Non-generic Output excluded from types.
self.assertEqual({}, resource_types(Resource8))
self.assertEqual({}, resource_types(Resource9))
# Type annotations not using Output.
self.assertEqual({"foo": str}, resource_types(Resource10))
self.assertEqual({"foo": str}, resource_types(Resource11))
# No return type annotation from the property getter.
self.assertEqual({}, resource_types(Resource12))

91
sdk/python/lib/test/test_utils.py Normal file
View file

@ -0,0 +1,91 @@
# Copyright 2016-2020, 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 unittest
from pulumi._utils import is_empty_function
# Function with return value based on input, called in the non_empty function
# bodies below.
def compute(val: int) -> str:
return f"{val} + {1} = {val + 1}"
class Foo:
def empty_a(self) -> str:
...
def empty_b(self) -> str:
"""A docstring."""
...
def empty_c(self, value: str):
...
def non_empty_a(self) -> str:
return "hello"
def non_empty_b(self) -> str:
"""A docstring."""
return "hello"
def non_empty_c(self) -> str:
return compute(41)
def non_empty_d(self) -> str:
"""F's docstring."""
return compute(41)
empty_lambda_a = lambda: None
empty_lambda_b = lambda: None
empty_lambda_b.__doc__ = """A docstring."""
non_empty_lambda_a = lambda: "hello"
non_empty_lambda_b = lambda: "hello"
non_empty_lambda_b.__doc__ = """A docstring."""
non_empty_lambda_c = lambda: compute(41)
non_empty_lambda_d = lambda: compute(41)
non_empty_lambda_d.__doc__ = """A docstring."""
class IsEmptyFunctionTests(unittest.TestCase):
def test_is_empty(self):
f = Foo()
self.assertTrue(is_empty_function(Foo.empty_a))
self.assertTrue(is_empty_function(Foo.empty_b))
self.assertTrue(is_empty_function(Foo.empty_c))
self.assertTrue(is_empty_function(f.empty_a))
self.assertTrue(is_empty_function(f.empty_b))
self.assertTrue(is_empty_function(f.empty_c))
self.assertFalse(is_empty_function(Foo.non_empty_a))
self.assertFalse(is_empty_function(Foo.non_empty_b))
self.assertFalse(is_empty_function(Foo.non_empty_c))
self.assertFalse(is_empty_function(Foo.non_empty_d))
self.assertFalse(is_empty_function(f.non_empty_a))
self.assertFalse(is_empty_function(f.non_empty_b))
self.assertFalse(is_empty_function(f.non_empty_c))
self.assertFalse(is_empty_function(f.non_empty_d))
self.assertTrue(is_empty_function(empty_lambda_a))
self.assertTrue(is_empty_function(empty_lambda_b))
self.assertFalse(is_empty_function(non_empty_lambda_a))
self.assertFalse(is_empty_function(non_empty_lambda_b))
self.assertFalse(is_empty_function(non_empty_lambda_c))
self.assertFalse(is_empty_function(non_empty_lambda_d))

3
tests/integration/types/python/declared/Pulumi.yaml Normal file
View file

@ -0,0 +1,3 @@
name: types_python
description: A program that uses input/output types with explicitly declared properties.
runtime: python

96
tests/integration/types/python/declared/__main__.py Normal file
View file

@ -0,0 +1,96 @@
# Copyright 2016-2020, Pulumi Corporation. All rights reserved.
from typing import Optional
import pulumi
from pulumi.dynamic import Resource, ResourceProvider, CreateResult
@pulumi.input_type
class AdditionalArgs:
def __init__(self, first_value: pulumi.Input[str], second_value: Optional[pulumi.Input[float]] = None):
pulumi.set(self, "first_value", first_value)
pulumi.set(self, "second_value", second_value)
# Property with empty getter/setter bodies.
@property
@pulumi.getter(name="firstValue")
def first_value(self) -> pulumi.Input[str]:
...
@first_value.setter
def first_value(self, value: pulumi.Input[str]):
...
# Property with explicitly specified getter/setter bodies.
@property
@pulumi.getter(name="secondValue")
def second_value(self) -> Optional[pulumi.Input[float]]:
return pulumi.get(self, "second_value")
@second_value.setter
def second_value(self, value: Optional[pulumi.Input[float]]):
pulumi.set(self, "second_value", value)
@pulumi.output_type
class Additional(dict):
def __init__(self, first_value: str, second_value: Optional[float]):
pulumi.set(self, "first_value", first_value)
pulumi.set(self, "second_value", second_value)
# Property with empty getter body.
@property
@pulumi.getter(name="firstValue")
def first_value(self) -> str:
...
# Property with explicitly specified getter/setter bodies.
@property
@pulumi.getter(name="secondValue")
def second_value(self) -> Optional[float]:
return pulumi.get(self, "second_value")
current_id = 0
class MyResourceProvider(ResourceProvider):
def create(self, inputs):
global current_id
current_id += 1
return CreateResult(str(current_id), {"additional": inputs["additional"]})
class MyResource(Resource):
additional: pulumi.Output[Additional]
def __init__(self, name: str, additional: pulumi.InputType[AdditionalArgs]):
super().__init__(MyResourceProvider(), name, {"additional": additional})
# Create a resource with input object.
res = MyResource("testres", additional=AdditionalArgs(first_value="hello", second_value=42))
# Create a resource using the output object of another resource.
res2 = MyResource("testres2", additional=AdditionalArgs(
first_value=res.additional.first_value,
second_value=res.additional.second_value))
# Create a resource using the output object of another resource, accessing the output as a dict.
res3 = MyResource("testres3", additional=AdditionalArgs(
first_value=res.additional["first_value"],
second_value=res.additional["second_value"]))
# Create a resource using a dict as the input.
# Note: These are camel case (not snake_case) since the resource does not do any translation of
# property names.
res4 = MyResource("testres4", additional={
"firstValue": "hello",
"secondValue": 42,
})
pulumi.export("res_first_value", res.additional.first_value)
pulumi.export("res_second_value", res.additional.second_value)
pulumi.export("res2_first_value", res2.additional.first_value)
pulumi.export("res2_second_value", res2.additional.second_value)
pulumi.export("res3_first_value", res3.additional.first_value)
pulumi.export("res3_second_value", res3.additional.second_value)
pulumi.export("res4_first_value", res4.additional.first_value)
pulumi.export("res4_second_value", res4.additional.second_value)

0
tests/integration/types/python/declared/requirements.txt Normal file
View file

3
tests/integration/types/python/simple/Pulumi.yaml Normal file
View file

@ -0,0 +1,3 @@
name: types_python
description: A program that uses input/output types.
runtime: python

62
tests/integration/types/python/simple/__main__.py Normal file
View file

@ -0,0 +1,62 @@
# Copyright 2016-2020, Pulumi Corporation. All rights reserved.
from typing import Optional
from pulumi import Input, InputType, Output, export, input_type, output_type, property
from pulumi.dynamic import Resource, ResourceProvider, CreateResult
@input_type
class AdditionalArgs:
first_value: Input[str] = property("firstValue")
second_value: Optional[Input[float]] = property("secondValue", default=None)
@output_type
class Additional(dict):
first_value: str = property("firstValue")
second_value: Optional[float] = property("secondValue", default=None)
current_id = 0
class MyResourceProvider(ResourceProvider):
def create(self, inputs):
global current_id
current_id += 1
return CreateResult(str(current_id), {"additional": inputs["additional"]})
class MyResource(Resource):
additional: Output[Additional]
def __init__(self, name: str, additional: InputType[AdditionalArgs]):
super().__init__(MyResourceProvider(), name, {"additional": additional})
# Create a resource with input object.
res = MyResource("testres", additional=AdditionalArgs(first_value="hello", second_value=42))
# Create a resource using the output object of another resource.
res2 = MyResource("testres2", additional=AdditionalArgs(
first_value=res.additional.first_value,
second_value=res.additional.second_value))
# Create a resource using the output object of another resource, accessing the output as a dict.
res3 = MyResource("testres3", additional=AdditionalArgs(
first_value=res.additional["first_value"],
second_value=res.additional["second_value"]))
# Create a resource using a dict as the input.
# Note: These are camel case (not snake_case) since the resource does not do any translation of
# property names.
res4 = MyResource("testres4", additional={
"firstValue": "hello",
"secondValue": 42,
})
export("res_first_value", res.additional.first_value)
export("res_second_value", res.additional.second_value)
export("res2_first_value", res2.additional.first_value)
export("res2_second_value", res2.additional.second_value)
export("res3_first_value", res3.additional.first_value)
export("res3_second_value", res3.additional.second_value)
export("res4_first_value", res4.additional.first_value)
export("res4_second_value", res4.additional.second_value)

0
tests/integration/types/python/simple/requirements.txt Normal file
View file

33
tests/integration/types/types_test.go Normal file
View file

@ -0,0 +1,33 @@
// Copyright 2016-2020, Pulumi Corporation. All rights reserved.
package ints
import (
"fmt"
"path/filepath"
"testing"
"github.com/pulumi/pulumi/pkg/v2/testing/integration"
"github.com/stretchr/testify/assert"
)
func TestPythonTypes(t *testing.T) {
for _, dir := range []string{"simple", "declared"} {
d := filepath.Join("python", dir)
t.Run(d, func(t *testing.T) {
integration.ProgramTest(t, &integration.ProgramTestOptions{
Dir: d,
Dependencies: []string{
filepath.Join("..", "..", "..", "sdk", "python", "env", "src"),
},
ExtraRuntimeValidation: func(t *testing.T, stack integration.RuntimeValidationStackInfo) {
for _, res := range []string{"", "2", "3", "4"} {
assert.Equal(t, "hello", stack.Outputs[fmt.Sprintf("res%s_first_value", res)])
assert.Equal(t, 42.0, stack.Outputs[fmt.Sprintf("res%s_second_value", res)])
}
},
UseAutomaticVirtualEnv: true,
})
})
}
}