-
Notifications
You must be signed in to change notification settings - Fork 429
feat: Idempotency helper utility #245
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from 14 commits
d503fb0
e564c63
c4d19ba
45d384b
13e5b09
1efc27d
546e879
60fd336
ee46124
acab091
2a364fd
4f5d52b
a19d955
0ef52f9
d128b0a
4caa52c
d89fcee
ed9e0c2
7000927
c4856fd
aed4a7b
3b6c2e3
2047d34
8a054cb
523535f
834db1c
24f6187
41d559e
dca02ee
b4490b9
43b72e7
88e983e
d17275d
83d78ce
4bdfdf6
9de6e29
a4cc61a
978a6bb
1fd8b5a
022739e
8a2d4fe
e6f2d98
4aa8145
c54952c
7832e56
fed58fe
b079387
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,10 @@ | ||
""" | ||
Utility for adding idempotency to lambda functions | ||
""" | ||
|
||
from aws_lambda_powertools.utilities.idempotency.persistence.base import BasePersistenceLayer | ||
from aws_lambda_powertools.utilities.idempotency.persistence.dynamodb import DynamoDBPersistenceLayer | ||
|
||
from .idempotency import idempotent | ||
|
||
__all__ = ("DynamoDBPersistenceLayer", "BasePersistenceLayer", "idempotent") |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,26 @@ | ||
from collections import OrderedDict | ||
|
||
|
||
class LRUDict(OrderedDict): | ||
def __init__(self, max_size=1024, *args, **kwds): | ||
self.max_size = max_size | ||
super().__init__(*args, **kwds) | ||
|
||
def __getitem__(self, key): | ||
value = super().__getitem__(key) | ||
self.move_to_end(key) | ||
return value | ||
|
||
def __setitem__(self, key, value): | ||
if key in self: | ||
self.move_to_end(key) | ||
super().__setitem__(key, value) | ||
if len(self) > self.max_size: | ||
oldest = next(iter(self)) | ||
del self[oldest] | ||
|
||
def get(self, key, *args, **kwargs): | ||
item = super(LRUDict, self).get(key, *args, **kwargs) | ||
if item: | ||
self.move_to_end(key=key) | ||
return item |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,33 @@ | ||
""" | ||
Idempotency errors | ||
""" | ||
|
||
|
||
class ItemAlreadyExistsError(Exception): | ||
""" | ||
Item attempting to be inserted into persistence store already exists | ||
""" | ||
|
||
|
||
class ItemNotFoundError(Exception): | ||
""" | ||
Item does not exist in persistence store | ||
""" | ||
|
||
|
||
class AlreadyInProgressError(Exception): | ||
""" | ||
Execution with idempotency key is already in progress | ||
""" | ||
|
||
|
||
class InvalidStatusError(Exception): | ||
""" | ||
An invalid status was provided | ||
""" | ||
|
||
|
||
class IdempotencyValidationerror(Exception): | ||
""" | ||
Payload does not match stored idempotency record | ||
""" |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,101 @@ | ||
""" | ||
Primary interface for idempotent Lambda functions utility | ||
""" | ||
import logging | ||
from typing import Any, Callable, Dict | ||
|
||
from aws_lambda_powertools.middleware_factory import lambda_handler_decorator | ||
from aws_lambda_powertools.utilities.idempotency.persistence.base import STATUS_CONSTANTS, BasePersistenceLayer | ||
|
||
from ..typing import LambdaContext | ||
from .exceptions import AlreadyInProgressError, ItemAlreadyExistsError, ItemNotFoundError | ||
|
||
logger = logging.getLogger(__name__) | ||
|
||
|
||
def default_error_callback(): | ||
raise | ||
|
||
|
||
@lambda_handler_decorator | ||
def idempotent( | ||
handler: Callable[[Any, LambdaContext], Any], | ||
event: Dict[str, Any], | ||
context: LambdaContext, | ||
persistence_store: BasePersistenceLayer, | ||
) -> Any: | ||
""" | ||
Middleware to handle idempotency | ||
|
||
Parameters | ||
---------- | ||
handler: Callable | ||
Lambda's handler | ||
event: Dict | ||
Lambda's Event | ||
context: Dict | ||
Lambda's Context | ||
persistence_store: BasePersistenceLayer | ||
Instance of BasePersistenceLayer to store data | ||
|
||
Examples | ||
-------- | ||
**Processes Lambda's event in an idempotent manner** | ||
>>> from aws_lambda_powertools.utilities.idempotency import idempotent, DynamoDBPersistenceLayer | ||
>>> | ||
>>> persistence_store = DynamoDBPersistenceLayer(event_key="body", table_name="idempotency_store") | ||
>>> | ||
>>> @idempotent(persistence_store=persistence_store) | ||
>>> def handler(event, context): | ||
>>> return {"StatusCode": 200} | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Question: Should we explain how it'll work under the hood here? This could help clarify how often this will call DynamoDB, what parameters it'll look at to decide if it's an idempotent request or not, etc. Question: Should we mention how this stores the event into DynamoDB? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I think if we address this it should be in the documentation. Happy to make changes to the docs based on feedback! |
||
""" | ||
|
||
try: | ||
to-mc marked this conversation as resolved.
Show resolved
Hide resolved
|
||
# We call save_inprogress first as an optimization for the most common case where no idempotent record already | ||
# exists. If it succeeds, there's no need to call get_record. | ||
persistence_store.save_inprogress(event=event) | ||
except ItemAlreadyExistsError: | ||
try: | ||
event_record = persistence_store.get_record(event) | ||
except ItemNotFoundError: | ||
return _call_lambda(handler=handler, persistence_store=persistence_store, event=event, context=context) | ||
to-mc marked this conversation as resolved.
Show resolved
Hide resolved
|
||
|
||
if event_record.status == STATUS_CONSTANTS["EXPIRED"]: | ||
return _call_lambda(handler=handler, persistence_store=persistence_store, event=event, context=context) | ||
to-mc marked this conversation as resolved.
Show resolved
Hide resolved
|
||
|
||
if event_record.status == STATUS_CONSTANTS["INPROGRESS"]: | ||
raise AlreadyInProgressError( | ||
f"Execution already in progress with idempotency key: " | ||
f"{persistence_store.event_key}={event_record.idempotency_key}" | ||
) | ||
|
||
if event_record.status == STATUS_CONSTANTS["COMPLETED"]: | ||
return event_record.response_json_as_dict() | ||
|
||
return _call_lambda(handler=handler, persistence_store=persistence_store, event=event, context=context) | ||
|
||
|
||
def _call_lambda( | ||
handler: Callable, persistence_store: BasePersistenceLayer, event: Dict[str, Any], context: LambdaContext | ||
) -> Any: | ||
""" | ||
|
||
Parameters | ||
---------- | ||
handler: Callable | ||
Lambda handler | ||
persistence_store: BasePersistenceLayer | ||
Instance of persistence layer | ||
event | ||
Lambda event | ||
context | ||
Lambda context | ||
""" | ||
try: | ||
handler_response = handler(event, context) | ||
to-mc marked this conversation as resolved.
Show resolved
Hide resolved
|
||
except Exception as ex: | ||
persistence_store.save_error(event=event, exception=ex) | ||
raise | ||
else: | ||
persistence_store.save_success(event=event, result=handler_response) | ||
return handler_response |
Uh oh!
There was an error while loading. Please reload this page.