Non-HTTP Invocation
You don't need Uvicorn and Starlette to use Smyth. Testing your Lambdas that handle events such as SQS, DynamoDB, etc., is also possible with Smyth. You can achieve this by creating an entrypoint script similar to the one in Custom Entrypoint.
Example
In your etc
directory, create a smyth_run.py
file.
Skip to the full file?
my_project/etc/smyth_run.py |
---|
| import asyncio
from smyth.runner.strategy import round_robin
from smyth.smyth import Smyth
def my_handler(event, context):
print(event["Records"][0]["body"], context.smyth["process"]["name"])
EVENT_DATA = {
"Records": [
{
"messageId": "19dd0b57-b21e-4ac1-bd88-01bbb068cb78",
"receiptHandle": "MessageReceiptHandle",
"body": "Hello from SQS!",
"attributes": {
"ApproximateReceiveCount": "1",
"SentTimestamp": "1523232000000",
"SenderId": "123456789012",
"ApproximateFirstReceiveTimestamp": "1523232000001",
},
"messageAttributes": {},
"md5OfBody": "7b270e59b47ff90a553787216d55d91d",
"eventSource": "aws:sqs",
"eventSourceARN": "arn:aws:sqs:us-east-1:123456789012:MyQueue",
"awsRegion": "us-east-1",
}
]
}
smyth = Smyth()
smyth.add_handler(
name="hello",
path="/hello",
lambda_handler=my_handler,
timeout=60,
concurrency=10,
strategy_generator=round_robin,
)
async def main():
handler = smyth.get_handler_for_name("hello")
await asyncio.gather(
*[
smyth.invoke(
handler=handler,
event_data=EVENT_DATA,
)
for _ in range(20)
],
return_exceptions=True,
)
if __name__ == "__main__":
with smyth:
asyncio.run(main())
|
Import and Declare the Basics
Import asyncio
, smyth
, and your Lambda handler function.
my_project/etc/smyth_run.py |
---|
| import asyncio
from smyth.runner.strategy import round_robin
from smyth.smyth import Smyth
def my_handler(event, context):
print(event["Records"][0]["body"], context.smyth["process"]["name"])
|
Create the Event Data
This example comes from sam local generate-event sqs receive-message
. Yours might be more complex.
my_project/etc/smyth_run.py |
---|
| EVENT_DATA = {
"Records": [
{
"messageId": "19dd0b57-b21e-4ac1-bd88-01bbb068cb78",
"receiptHandle": "MessageReceiptHandle",
"body": "Hello from SQS!",
"attributes": {
"ApproximateReceiveCount": "1",
"SentTimestamp": "1523232000000",
"SenderId": "123456789012",
"ApproximateFirstReceiveTimestamp": "1523232000001",
},
"messageAttributes": {},
"md5OfBody": "7b270e59b47ff90a553787216d55d91d",
"eventSource": "aws:sqs",
"eventSourceARN": "arn:aws:sqs:us-east-1:123456789012:MyQueue",
"awsRegion": "us-east-1",
}
]
}
|
Instantiate Smyth and Add Handlers
Initialize Smyth and add your handlers.
my_project/etc/smyth_run.py |
---|
| smyth = Smyth()
smyth.add_handler(
name="hello",
path="/hello",
lambda_handler=my_handler,
timeout=60,
concurrency=10,
strategy_generator=round_robin,
)
|
Prepare the Main Function
This part depends on what you want to achieve. Here we are trying to simulate a bunch of SQS messages quickly triggering a handler.
my_project/etc/smyth_run.py |
---|
| async def main():
handler = smyth.get_handler_for_name("hello")
await asyncio.gather(
*[
smyth.invoke(
handler=handler,
event_data=EVENT_DATA,
)
for _ in range(20)
],
return_exceptions=True,
)
if __name__ == "__main__":
with smyth:
asyncio.run(main())
|
Run It
python etc/smyth_run.py
Hello from SQS! hello:8
Hello from SQS! hello:4
Hello from SQS! hello:5
Hello from SQS! hello:3
Hello from SQS! hello:1
Hello from SQS! hello:2
Hello from SQS! hello:6
Hello from SQS! hello:4
Hello from SQS! hello:3
Hello from SQS! hello:1
Hello from SQS! hello:0
Hello from SQS! hello:5
Hello from SQS! hello:8
Hello from SQS! hello:2
Hello from SQS! hello:0
Hello from SQS! hello:6
Hello from SQS! hello:9
Hello from SQS! hello:9
Hello from SQS! hello:7
Hello from SQS! hello:7