Processor

Core class for other components to inherit.

class sosw.app.LambdaGlobals[source]

Global placeholder for global_vars that we want to preserve in the lifetime of the Lambda Container. e.g. once initiailised the given Processor, we keep it alive in the container to minimize warm-run time.

This namespace also contains the lambda_context which should be reset by get_lambda_handler method. See Worker examples in documentation for more info.

class sosw.app.Processor(custom_config=None, **kwargs)[source]

Core Processor class template. All the main components (Worker, Orchestrator and Scheduler) inherit from this one. You can also use this class as parent for some of your standalone Lambdas, but we strictly encourage you to use Worker class in case you are running functions under sosw orchestration.

get_ddbc(prefix: str) -> DynamoDbClient:

Lazily initializes and retrieves a DynamoDB client configured for a specific table and schema validation.

This method initializes custom DynamoDB clients based on the provided prefix. If a client with the specified prefix has already been initialized, it returns the existing client. If not, it looks in the Processor config for a prefixed dynamodb config (e.g. for prefix project_a -> project_a_dynamo_db_config). The dynamo_db client will be initialized as self.project_a_dynamo_db_client.

This is particularly useful for scenarios requiring schema validation, transformation of DynamoDB syntax to dictionary format, and other operations beyond the capabilities of the raw boto3 client.

Parameters:: prefix (str) – The prefix for the DynamoDB client configuration and naming.
Raises:: ValueError – If the provided prefix is not supported by the available configuration.

property _account

Get current AWS Account to construct different ARNs.

We don’t have this parameter in Environmental variables, only can parse from Context. It is stored in global_vars and is supposed to be passed by your lambda_handler during initialization.

As a fallback for cases when we use Processor not in the Lambda environment, we have a lazy autodetection mechanism using STS, but it is pretty heavy (~0.3 seconds).

Some things to note:

We store this value in class variable for fast access
It uses Lazy initialization.
We first try from context and only if not provided - use the autodetection.

property _region: Property fetched from AWS Lambda Environmental variables.

die(message='Unknown Failure')[source]

Logs current Processor stats and message. Then raises RuntimeError with message.

If there is access to publish SNS messages, the method will also try to publish to the topic configured as dead_sns_topic or ‘SoswWorkerErrors’.

Parameters:: message (str) – Description of failure.

static get_config(name)[source]

Returns config by name from DynamoDB config or SSM. Override this to provide your config handling method.

Parameters:: name – Name of the config
Return type:: dict

get_stats(recursive: bool = True)[source]

Return statistics of operations performed by current instance of the Class.

Statistics of custom clients existing in the Processor is also aggregated by default. Clients must be initialized as self.some_client ending with _client suffix (e.g. self.dynamo_db_client). Clients must also have their own get_stats() methods implemented.

Be careful about circular get_stats() calls from child classes. If required overwrite get_stats() with recursive = False.

Parameters:: recursive – Merge stats from self.***_client.
Return type:: dict
Returns:: Statistics counter of current Processor instance.

init_config(custom_config: Dict | None = None)[source]

By default, tries to initialize config from DEFAULT_CONFIG or as an empty dictionary. After that, a specific custom config of the Lambda will recursively update the existing one. The last step is update config recursively with a passed custom_config.

Overwrite this method if custom logic of recursive updates in configs is required.

Note