c7n.utils module

class c7n.utils.DateTimeEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)

Bases: json.encoder.JSONEncoder

default(obj)

Implement this method in a subclass such that it returns a serializable object for o, or calls the base implementation (to raise a TypeError).

For example, to support arbitrary iterators, you could implement default like this:

def default(self, o):
    try:
        iterable = iter(o)
    except TypeError:
        pass
    else:
        return list(iterable)
    # Let the base class default method raise the TypeError
    return JSONEncoder.default(self, o)

class c7n.utils.FormatDate(d=None)

Bases: object

a datetime wrapper with extended pyformat syntax

date_increment = re.compile('\\+[0-9]+[Mdh]')

classmethod utcnow()

class c7n.utils.IPv4Network(address, strict=True)

Bases: c7n.ipaddress.IPv4Network

class c7n.utils.QueryParser

Bases: object

QuerySchema = {}

multi_value = True

classmethod parse(data)

type_name = ''

value_key = 'Values'

class c7n.utils.SafeDumper(stream, default_style=None, default_flow_style=False, canonical=None, indent=None, width=None, allow_unicode=None, line_break=None, encoding=None, explicit_start=None, explicit_end=None, version=None, tags=None, sort_keys=True)

Bases: yaml.dumper.SafeDumper

ignore_aliases(data)

class c7n.utils.UnicodeWriter(f, dialect=<class 'csv.excel'>, **kwds)

Bases: object

utf8 encoding csv writer.

writerow(row)

writerows(rows)

exception c7n.utils.VarsSubstitutionError

Bases: Exception

c7n.utils.annotation(i, k)

c7n.utils.backoff_delays(start, stop, factor=2.0, jitter=False)

Geometric backoff sequence w/ jitter

c7n.utils.camelResource(obj)

Some sources from apis return lowerCased where as describe calls

always return TitleCase, this function turns the former to the later

c7n.utils.chunks(iterable, size=50)

Break an iterable into lists of size

c7n.utils.dumps(data, fh=None, indent=0)

c7n.utils.filter_empty(d)

c7n.utils.format_event(evt)

c7n.utils.format_string_values(obj, err_fallback=(<class 'IndexError'>, <class 'KeyError'>), *args, **kwargs)

Format all string values in an object. Return the updated object

c7n.utils.generate_arn(service, resource, partition='aws', region=None, account_id=None, resource_type=None, separator='/')

Generate an Amazon Resource Name. See http://docs.aws.amazon.com/general/latest/gr/aws-arns-and-namespaces.html.

c7n.utils.get_account_alias_from_sts(session)

c7n.utils.get_account_id_from_sts(session)

c7n.utils.get_annotation_prefix(s)

c7n.utils.get_retry(codes=(), max_attempts=8, min_delay=1, log_retries=False)

Decorator for retry boto3 api call on transient errors.

https://www.awsarchitectureblog.com/2015/03/backoff.html https://en.wikipedia.org/wiki/Exponential_backoff

Parameters:

• codes – A sequence of retryable error codes.
• max_attempts – The max number of retries, by default the delay time is proportional to the max number of attempts.
• log_retries – Whether we should log retries, if specified specifies the level at which the retry should be logged.
• _max_delay – The maximum delay for any retry interval note this parameter is only exposed for unit testing, as its derived from the number of attempts.

Returns a function for invoking aws client calls that retries on retryable error codes.

c7n.utils.group_by(resources, key)

Return a mapping of key value to resources with the corresponding value.

Key may be specified as dotted form for nested dictionary lookup

c7n.utils.load_file(path, format=None, vars=None)

c7n.utils.loads(body)

c7n.utils.local_session(factory)

Cache a session thread local for up to 45m

c7n.utils.parse_cidr(value)

Process cidr ranges.

c7n.utils.parse_s3(s3_path)

c7n.utils.parse_url_config(url)

c7n.utils.query_instances(session, client=None, **query)

Return a list of ec2 instances for the query.

c7n.utils.reformat_schema(model)

Reformat schema to be in a more displayable format.

c7n.utils.reset_session_cache()

c7n.utils.set_annotation(i, k, v)

>>> x = {}
>>> set_annotation(x, 'marker', 'a')
>>> annotation(x, 'marker')
['a']

c7n.utils.set_value_from_jmespath(source, expression, value, is_first=True)

c7n.utils.snapshot_identifier(prefix, db_identifier)

Return an identifier for a snapshot of a database or cluster.

c7n.utils.type_schema(type_name, inherits=None, rinherit=None, aliases=None, required=None, **props)

jsonschema generation helper

params:

• type_name: name of the type
• inherits: list of document fragments that are required via anyOf[$ref]

• rinherit: use another schema as a base for this, basically work around

  inherits issues with additionalProperties and type enums.

• aliases: additional names this type maybe called
• required: list of required properties, by default ‘type’ is required
• props: additional key value properties

c7n.utils.worker(f)

Generic wrapper to log uncaught exceptions in a function.

When we cross concurrent.futures executor boundaries we lose our traceback information, and when doing bulk operations we may tolerate transient failures on a partial subset. However we still want to have full accounting of the error in the logs, in a format that our error collection (cwl subscription) can still pickup.

c7n.utils.yaml_dump(value)

c7n.utils.yaml_load(value)