seda 0.0.6

A Python toolkit to build Serverless Event-Driven Applications on AWS.

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)