runway.cfngin.lookups.handlers.dynamodb module

DynamoDB lookup.

class runway.cfngin.lookups.handlers.dynamodb.ArgsDataModel[source]

Bases: runway.utils.BaseModel

Arguments data model.

region: Optional[str]

AWS region.

__contains__(name: object) → bool

Implement evaluation of ‘in’ conditional.

Parameters

name – The name to check for existence in the model.

__getitem__(name: str) → Any

Implement evaluation of self[name].

Parameters

name – Attribute name to return the value for.

Returns

The value associated with the provided name/attribute name.

Raises

AttributeError – If attribute does not exist on this object.

__init__(**data: Any) → None

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

__iter__() → TupleGenerator

so dict(model) works

__new__(**kwargs)
__pretty__(fmt: Callable[[Any], Any], **kwargs: Any) → Generator[Any, None, None]

Used by devtools (https://python-devtools.helpmanual.io/) to provide a human readable representations of objects

__repr_name__() → unicode

Name of the instance’s class, used in __repr__.

__rich_repr__() → RichReprResult

Get fields for Rich library

__setitem__(name: str, value: Any) → None

Implement item assignment (e.g. self[name] = value).

Parameters

  • name – Attribute name to set.

  • value – Value to assign to the attribute.

classmethod __try_update_forward_refs__(**localns: Any) → None

Same as update_forward_refs but will not raise exception when forward references are not defined.

classmethod construct(_fields_set: Optional[SetStr] = None, **values: Any) → Model

Creates a new model setting __dict__ and __fields_set__ from trusted or pre-validated data. Default values are respected, but no other validation is performed. Behaves as if Config.extra = ‘allow’ was set since it adds all passed values

copy(*, include: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, exclude: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, update: Optional[DictStrAny] = None, deep: bool = False) → Model

Duplicate a model, optionally choose which fields to include, exclude and change.

Parameters

  • include – fields to include in new model

  • exclude – fields to exclude from new model, as with values this takes precedence over include

  • update – values to change/add in the new model. Note: the data is not validated before creating the new model: you should trust this data

  • deep – set to True to make a deep copy of the model

Returns

new model instance

dict(*, include: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, exclude: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, by_alias: bool = False, skip_defaults: Optional[bool] = None, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False) → DictStrAny

Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.

get(name: str, default: Optional[Any] = None) → Any

Safely get the value of an attribute.

Parameters

  • name – Attribute name to return the value for.

  • default – Value to return if attribute is not found.

json(*, include: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, exclude: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, by_alias: bool = False, skip_defaults: Optional[bool] = None, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False, encoder: Optional[Callable[[Any], Any]] = None, models_as_dict: bool = True, **dumps_kwargs: Any) → unicode

Generate a JSON representation of the model, include and exclude arguments as per dict().

encoder is an optional function to supply as default to json.dumps(), other arguments as per json.dumps().

classmethod update_forward_refs(**localns: Any) → None

Try to update ForwardRefs on fields based on this Model, globalns and localns.

class runway.cfngin.lookups.handlers.dynamodb.QueryDataModel[source]

Bases: runway.utils.BaseModel

Arguments data model.

attribute: str

The attribute to be returned by this lookup. Supports additional syntax to retrieve a nested value.

partition_key: str

The DynamoDB Table’s partition key.

partition_key_value: str

The value of the partition key to query the database.

table_name: str

Name of the DynamoDB Table to query.

property item_key: Dict[str, Dict[typing_extensions.Literal[B, N, S], Any]]

Value to pass to boto3 .get_item() call as the Key argument.

Raises

ValueError – The value of partition_key_value doesn’t match the required regex and so it can’t be parsed.

__contains__(name: object) → bool

Implement evaluation of ‘in’ conditional.

Parameters

name – The name to check for existence in the model.

__getitem__(name: str) → Any

Implement evaluation of self[name].

Parameters

name – Attribute name to return the value for.

Returns

The value associated with the provided name/attribute name.

Raises

AttributeError – If attribute does not exist on this object.

__init__(**data: Any) → None

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

__iter__() → TupleGenerator

so dict(model) works

__new__(**kwargs)
__pretty__(fmt: Callable[[Any], Any], **kwargs: Any) → Generator[Any, None, None]

Used by devtools (https://python-devtools.helpmanual.io/) to provide a human readable representations of objects

__repr_name__() → unicode

Name of the instance’s class, used in __repr__.

__rich_repr__() → RichReprResult

Get fields for Rich library

__setitem__(name: str, value: Any) → None

Implement item assignment (e.g. self[name] = value).

Parameters

  • name – Attribute name to set.

  • value – Value to assign to the attribute.

classmethod __try_update_forward_refs__(**localns: Any) → None

Same as update_forward_refs but will not raise exception when forward references are not defined.

classmethod construct(_fields_set: Optional[SetStr] = None, **values: Any) → Model

Creates a new model setting __dict__ and __fields_set__ from trusted or pre-validated data. Default values are respected, but no other validation is performed. Behaves as if Config.extra = ‘allow’ was set since it adds all passed values

copy(*, include: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, exclude: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, update: Optional[DictStrAny] = None, deep: bool = False) → Model

Duplicate a model, optionally choose which fields to include, exclude and change.

Parameters

  • include – fields to include in new model

  • exclude – fields to exclude from new model, as with values this takes precedence over include

  • update – values to change/add in the new model. Note: the data is not validated before creating the new model: you should trust this data

  • deep – set to True to make a deep copy of the model

Returns

new model instance

dict(*, include: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, exclude: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, by_alias: bool = False, skip_defaults: Optional[bool] = None, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False) → DictStrAny

Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.

get(name: str, default: Optional[Any] = None) → Any

Safely get the value of an attribute.

Parameters

  • name – Attribute name to return the value for.

  • default – Value to return if attribute is not found.

json(*, include: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, exclude: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, by_alias: bool = False, skip_defaults: Optional[bool] = None, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False, encoder: Optional[Callable[[Any], Any]] = None, models_as_dict: bool = True, **dumps_kwargs: Any) → unicode

Generate a JSON representation of the model, include and exclude arguments as per dict().

encoder is an optional function to supply as default to json.dumps(), other arguments as per json.dumps().

classmethod update_forward_refs(**localns: Any) → None

Try to update ForwardRefs on fields based on this Model, globalns and localns.

class runway.cfngin.lookups.handlers.dynamodb.DynamodbLookup[source]

Bases: runway.lookups.handlers.base.LookupHandler

DynamoDB lookup.

TYPE_NAME: Final[typing_extensions.Literal[dynamodb]] = 'dynamodb'

Name that the Lookup is registered as.

classmethod parse(value: str) → Tuple[str, Dict[str, str]][source]

Parse the value passed to the lookup.

This overrides the default parsing to account for special requirements.

Parameters

value – The raw value passed to a lookup.

Returns

The lookup query and a dict of arguments

Raises

ValueError – The value provided does not appear to contain the name of a DynamoDB Table. The name of a Table is required.

classmethod parse_query(value: str) → runway.cfngin.lookups.handlers.dynamodb.QueryDataModel[source]

Parse query string to extract. Does not support arguments in value.

Raises

ValueError – The argument provided does not match the expected format defined with a regex pattern.

__init__()
__new__(**kwargs)
classmethod dependencies(_LookupHandler__lookup_query: VariableValue) → Set[str]

Calculate any dependencies required to perform this lookup.

Note that lookup_query may not be (completely) resolved at this time.

classmethod format_results(value: Any, get: Optional[str] = None, load: Optional[str] = None, transform: Optional[typing_extensions.Literal[bool, str]] = None, **kwargs: Any) → Any

Format results to be returned by a lookup.

Parameters

  • value – Data collected by the Lookup.

  • get – Nested value to get from a dictionary like object.

  • load – Parser to use to parse a formatted string before the get and transform method.

  • transform – Convert the final value to a different data type before returning it.

Raises

TypeError – If get is provided but the value value is not a dictionary like object.

Runs the following actions in order:

  1. load() if load is provided.

  2. runway.util.MutableMap.find() or dict.get() depending on the data type if get is provided.

  3. Convert null value string to NoneType object. This includes string values of “None” and “null”. This conversion is case insensitive.

  4. transform() if transform is provided.

classmethod handle(value: str, context: runway.context.CfnginContext, *_DynamodbLookup__args: Any, **_DynamodbLookup__kwargs: Any) → Any[source]

Get a value from a DynamoDB table.

Parameters

  • value – Parameter(s) given to this lookup. [<region>:]<tablename>@<partitionkey>:<keyvalue>.<keyvalue>...

  • context – Context instance.

Raises

ValueError – The value provided to the lookup resulted in an error.

Note

The region is optional, and defaults to the environment’s AWS_DEFAULT_REGION if not specified.

classmethod load(value: Any, parser: Optional[str] = None, **kwargs: Any) → Any

Load a formatted string or object into a python data type.

First action taken in format_results(). If a lookup needs to handling loading data to process it before it enters format_results(), is should use args.pop('load') to prevent the data from being loaded twice.

Parameters

  • value – What is being loaded.

  • parser – Name of the parser to use.

Returns

The loaded value.

classmethod transform(value: Any, *, to_type: Optional[typing_extensions.Literal[bool, str]] = 'str', **kwargs: Any) → Any

Transform the result of a lookup into another datatype.

Last action taken in format_results(). If a lookup needs to handling transforming the data in a way that the base class can’t support it should overwrite this method of the base class to register different transform methods.

Parameters

  • value – What is to be transformed.

  • to_type – The type the value will be transformed into.

Returns

The transformed value.

class runway.cfngin.lookups.handlers.dynamodb.ParsedLookupKey[source]

Bases: typing_extensions.TypedDict

Return value of _lookup_key_parse.