5.5 Shared helpers (aws_services/_base.py)¶
Module functions:
| Function | Purpose |
|---|---|
rel(target, edge, relationship=None, *, reverse=False, description=None, **properties) |
Build a relation object (reference); None for an empty target. |
tag_dict(tags) |
Normalise [{"Key","Value"}], [{"key","value"}] and {k: v} tag shapes. |
policy_document(doc) |
Decode a policy given as a dict, JSON or URL-encoded JSON (IAM returns the latter). |
as_list(value) |
None → [], scalar → [scalar]. |
policy_statements(doc) |
Statement list of a policy, tolerant of a single statement object. |
policy_principals(statement) |
{"AWS": [...], "Service": [...], "Federated": [...]}; "*" becomes {"AWS": ["*"]}. |
condition_values(statement, *keys) |
Values of condition keys across all operators, case-insensitive. |
account_principal(p) / principal_ref(p) |
Account IDs and :root ARNs → canonical arn:aws:iam::<id>:root. |
arns_in(value, limit=200) |
Every ARN inside a nested value (state machine definitions, pipeline configurations). |
identifier_refs(env) |
Identifier-shaped values (ARNs, SQS queue URLs verified on the parsed host) of an environment map. Everything else is dropped, so secrets kept in environment variables never reach the map. |
resource_policy_relations(policy, account_id) |
Relations from a resource policy: service principals with an aws:SourceArn condition become INVOKES / TRIGGERED_BY from the source (reverse); AWS principals become GRANTS_ACCESS / POLICY_ALLOWS_ACTION (reverse, cross_account flagged). Returns (relations, public); public is true for an unconditional *. |
gather_limited(factories, limit=8) |
Run coroutine factories with bounded concurrency; a failing item becomes None. Use it for per-item describe calls. |
error_code(exc) |
The AWS error code of an exception, or its class name. |
AWSServiceMixin methods:
| Method | Purpose |
|---|---|
_client(service, region=None) |
Async client context manager with the collector's retry config. |
_paginate(client, operation, result_key, **kwargs) |
Async iterator over a botocore paginator. |
_pages(call, result_key, cursor_param="NextToken", cursor_key=None, max_pages=1000, **kwargs) |
Manual cursor pagination for operations without a paginator (_paginate would raise OperationNotPageableError). cursor_param is the request parameter, cursor_key the response key when it differs. |
_arn(service, resource, region=None) |
arn:aws:<service>:<region>:<account>:<resource>; pass region="" for regionless ARNs. |
_account_ref() |
arn:aws:iam::<account>:root. |
_asset(*, arn, name, asset_type, region=None, tags=None, metadata=None, relations=(), raw=None, exposed=False, aliases=()) |
Build an AWS CloudAsset: drops empty relations and aliases, normalises tags, fills provider / region / account. Use it for every new asset. |
_disabled_asset(service, asset_type, label, reason="not enabled") |
The not-enabled placeholder for a security service (5.10). |
_security_alias(service) |
aws-security:<service>:<region>:<account>. |
Each package also has a _common.py with shared helpers for its modules (for example
application/_common.py: _host, _clean_url which strips credentials and query strings from
repository URLs, _bucket_arn, _ecr_repo, _env_names, _vpc_config, _role_rel,
ApplicationBase._secret_or_param_ref which reduces a Secrets Manager reference to the secret ARN
without its JSON key / version suffix).