A Python toolkit to build Serverless Event-Driven Applications on AWS.
Project description
SEDA
A Python toolkit to build Serverless Event-Driven Applications on AWS.
Documentation: https://seda.domake.io (pending)
Examples: /templates
What
- Allows to schedule periodic and one-time tasks on EventBridge Scheduler.
- Simplifies creating, executing, and managing asynchronous task using SNS messages and Lambda events.
- Includes
@decorators
in addition to@app.decorators
for reusable apps. - Works well with any framework, interface, toolkit... see /templates.
- Provides Serverless framework support via plugin.
- Other event sources, e.g. CloudWatch, Kinesis, Dynamodb, SQS, S3...
- Easy to read documentation and tests.
- SAM templates and CDK support.
Installation
pip install seda
Example
main.py:
from datetime import datetime
from seda import Seda
seda = Seda()
@seda.schedule("cron(* * * * ? *)", args=("minutes",))
async def myschedule(timespec: str = "auto") -> None:
seda.log.info(f"myschedule: {datetime.now().isoformat(timespec=timespec)}")
@seda.task
async def mytask(timespec: str = "auto") -> None:
seda.log.info(f"mytask: {datetime.now().isoformat(timespec=timespec)}")
main.seda
is an AWS Lambda handler in charge of managing the creation and execution of our tasks.
@schedule
: creates a new periodic "schedule" on EventBridge at deployment time.@task
: creates one-time asynchronous tasks at runtime:- SNS messages: default
- Lambda events:
@task(service="lambda")
- EventBridge one-time schedules:
mytask.at("...")
For reusable apps use @task
and @schedule
decorators that always points to the currently active Seda
instance:
tasks.py:
from datetime import datetime
from seda import task
@task
async def mytask(timespec: str = "auto") -> str:
return datetime.now().isoformat(timespec=timespec)
Tasks
await mytask()
- SNS: This task is executed asynchronously by sending a message to a previously subscribed SNS topic.
- λ: A second option is to directly invoke the Lambda function
InvocationType=Event
by adding the "service" option to the task decorator@task(service="lambda")
. - test: For local and test environments the task is executed synchronously by default.
One-time schedules
from datetime import datetime, timedelta
mytask.at(datetime.now() + timedelta(minutes=5))
The .at(datetime)
method is equivalent to @schedule("at(datetime)")
.
These one-time schedules are created under a second EventBridge Schedule Group (group 1 - N schedules) so after a new deployment we can clean the group of periodic schedules but keeping our one-time schedules.
@schedule
SEDA / AWS | TYPE | EXAMPLE |
---|---|---|
expression ScheduleExpression |
str |
- "rate(5 minutes)" - "cron(*/5 * * * ? *)" - "at(2025-10-26T12:00:00)" |
args - |
Optional[Sequence] |
("a", "b") |
kwargs - |
Optional[Dict] |
{"a": "b"} |
timezone ScheduleExpressionTimezone |
Optional[str] |
"Asia/Saigon" |
time_window FlexibleTimeWindow |
Optional[ScheduleTimeWindow] |
{"Mode": "FLEXIBLE", "MaximumWindowInMinutes": 15} |
retry_policy RetryPolicy |
Optional[ScheduleRetryPolicy] |
{"MaximumEventAgeInSeconds": 60, "MaximumRetryAttempts": 10} |
start_date StartDate |
Optional[datetime] |
datetime.now() + timedelta(minutes=5) |
end_date EndDate |
Optional[datetime] |
datetime.now() + timedelta(days=5) |
dead_letter_arn DeadLetterConfig.Arn |
Optional[str] |
"arn:aws:sqs:..." |
kms_key KmsKeyArn |
Optional[str] |
"arn:aws:kms:..." |
CLI
SEDA CLI provides a list of commands to deploy, remove and debug SEDA resources on an existing Lambda function:
seda deploy --app main.seda -f myfunction
Deploy @schedule
, @task
:
- Creates the Schedule Groups for periodic and one-time tasks
- Creates N periodic schedules
- Creates SNS topic and a Lambda subscription to this topic
- Adds related IAM roles and policies
A second deployment removes the periodic task Schedule Group and creates a new one adding the new schedules.
We can also remove the deployed stack:
seda remove --app main.seda -f myfunction
Remote Function Invocation
Invoke Python interpreter:
seda python 'import sys;print(sys.version)' -f myfunction
Run shell commands:
seda shell env -f myfunction
Serverless Framework
The plugin serverless-seda adds all SEDA CLI commands to Serverless framework CLI:
sls seda deploy
ASGI
SEDA seamlessly integrates with ASGI applications by adding Mangum handler to seda[asgi]
.
pip install seda[asgi]
main.py:
from fastapi import FastAPI
from seda import Seda
app = FastAPI()
seda = Seda(app)
Default handler
Add any callable as a handler to process any non-SEDA events:
def myhandler(event, context):
pass
seda = Seda(default_handler=myhandler)
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.