A Python toolkit to schedule periodic and one-time tasks on AWS Lambda
BSD-3-CLAUSE License
@decorators
in addition to @app.decorators
for reusable apps.pip install seda
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:
@task(service="lambda")
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)
await mytask()
InvocationType=Event
by adding the "service" option to the task decorator @task(service="lambda")
.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 |
---|---|---|
expressionScheduleExpression | str |
- "rate(5 minutes)" - "cron(*/5 * * * ? *)" - "at(2025-10-26T12:00:00)"
|
args- | Optional[Sequence] |
("a", "b") |
kwargs- | Optional[Dict] |
{"a": "b"} |
timezoneScheduleExpressionTimezone | Optional[str] |
"Asia/Saigon" |
time_windowFlexibleTimeWindow | Optional[ScheduleTimeWindow] |
{"Mode": "FLEXIBLE", "MaximumWindowInMinutes": 15} |
retry_policyRetryPolicy | Optional[ScheduleRetryPolicy] |
{"MaximumEventAgeInSeconds": 60, "MaximumRetryAttempts": 10} |
start_dateStartDate | Optional[datetime] |
datetime.now() + timedelta(minutes=5) |
end_dateEndDate | Optional[datetime] |
datetime.now() + timedelta(days=5) |
dead_letter_arnDeadLetterConfig.Arn | Optional[str] |
"arn:aws:sqs:..." |
kms_keyKmsKeyArn | Optional[str] |
"arn:aws:kms:..." |
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
:
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
The plugin serverless-seda adds all SEDA CLI commands to Serverless framework CLI:
sls seda deploy
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)
Add any callable as a handler to process any non-SEDA events:
def myhandler(event, context):
pass
seda = Seda(default_handler=myhandler)