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?
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.
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
Any workaround on how to do this at the moment?
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!
Most helpful comment
Any workaround on how to do this at the moment?