aws / chalice

Python Serverless Microframework for AWS
Apache License 2.0
10.67k stars 1.01k forks source link

Unable to run cron in local #979

Open adsahay opened 6 years ago

adsahay commented 6 years ago

I'm trying to write a chalice app for scheduled events.

import logging
from chalice import Chalice
from chalice.app import Cron

app = Chalice(app_name='import_blogs')
# Enable DEBUG logs.
app.log.setLevel(logging.DEBUG)

@app.schedule(Cron('*/2', '*', '*', '*', '?', '*'))
def import_rss(event):
    app.log.debug('Inside import_rss')
    return {'hello': 'rss'}

I expect the log to be printed every two minutes when running chalice local. However, there is no output.

Is there a way to simulate Cloudwatch scheduled events in local development, without deploying to lambda?

stealthycoin commented 6 years ago

Yea that is a downside of local at the moment is that it lags behind our actual feature set. This would be a feature request since it doesn't exist right now.

sla-shi commented 5 years ago

Wasted half an hour trying to figure out this limitation. Would be good to have a notice in the official documentation https://chalice.readthedocs.io/en/latest/api.html#Chalice.schedule

pablodo commented 5 years ago

Any workaround on how to do this at the moment?

andersoncontreira commented 4 years ago

In my case, my workaround for this situation is work with an unit test to execute the code and test my logic, is not the perfect scenario but is a temporary solution.

Bellow a simple example: aws_lambda_mock.py `

class FakeLambdaContextIdentity(object):
    def __init__(self, cognito_identity_id, cognito_identity_pool_id):
    self.cognito_identity_id = cognito_identity_id
    self.cognito_identity_pool_id = cognito_identity_pool_id

class FakeLambdaContext(object):
    def __init__(self):
    self.function_name = 'test_name'
    self.function_version = 'version'
    self.invoked_function_arn = 'arn'
    self.memory_limit_in_mb = 256
    self.aws_request_id = 'id'
    self.log_group_name = 'log_group_name'
    self.log_stream_name = 'log_stream_name'
    self.identity = FakeLambdaContextIdentity('id', 'id_pool')
    # client_context is set by the mobile SDK and wont be set for chalice
    self.client_context = None

    def get_remaining_time_in_millis(self):
    return 500

    def serialize(self):
    serialized = {}
    serialized.update(vars(self))
    serialized['identity'] = vars(self.identity)
    return serialized

`

test_app.py `

import unittest
from tests.unit.mocks.aws_lambda_mock import FakeLambdaContext

class AppTestCase(unittest.TestCase):
    def test_index(self):

    event = {
        "version": "1",
        "id": "cdc73f9d-aea9-11e3-9d5a-835b769c0d9c",
        "detail-type": "Scheduled Event",
        "source": "aws.events",
        "account": "123456789012",
        "time": "1970-01-01T00:00:00Z",
        "region": "us-east-2",
        "resources": [
            "arn:aws:events:us-east-2:123456789012:rule/ExampleRule"
        ],
        "detail": {}
    }

    lambda_context = FakeLambdaContext()
    response = app.index(event, lambda_context)

    self.assertTrue(response)

if __name__ == '__main__':
    unittest.main()

`

I hope this can help someone!

shtlrs commented 3 years ago

Hello, is there an update on this ? Also, the other question I'd like to ask is: We know it doesn't run locally but will a scheduled job run in production ?

r0lodex commented 2 years ago

The cron won't run locally, but for me, after the app is deployed, I'll check in EventBridge if the event exist or not. It's just a timed execution of our functions, so testing locally would mean just invoking it.

I wrote test with what Chalice has provided, deploy and check at Lambda and EventBridge. Like so:

# app.py
from chalice import Chalice, Cron
from chalice.app import CloudWatchEvent

app = Chalice(app_name='myApp')

@app.schedule(Cron(0, 6, "*", "*", "?", "*"))
def notify(event: CloudWatchEvent):
    return "This should be invoked every day at 6am"

And the test;

# tests/test_notify.py

from chalice.test import Client
from app import app

def test_notify():
    with Client(app) as client:
        response = client.lambda_.invoke(
            "notify",
            client.events.generate_cw_event(
                source="",
                detail_type="",
                detail="",
                resources=""
            ),
        )

        assert response.payload == "This should be invoked every day at 6am"
shayneoneill commented 11 months ago

Any updates. I've spent the better part of the day tearing my code to pieces wondering why the schedules wouldnt work and .... this isn't implented?

Some documentation noting that would help.

Is this still true?

edit: My bad, I just found the note it only works via chalice deploy. I was looking for info on the local deploy thing, but it was under schedule doc

mattwa1sh commented 7 months ago

5 years later and still not implemented. kind of a disgrace. this is an essential component.

griff4594 commented 4 months ago

5 years later and still not implemented. kind of a disgrace. this is an essential component.

There is a work around for this using docker and having it ping it, but that takes time to setup. The best way to implement a work around is to make a test endpoint that runs the exact same code, so it is a manual trigger. That way you know the code behind it works fine, you just need the cron tab to be set.