Skip to content

Instantly share code, notes, and snippets.

@bharath-kotha
Last active June 13, 2021 16:23
Show Gist options
  • Save bharath-kotha/feb589e494c2c29eee0c1dda2cf81f2b to your computer and use it in GitHub Desktop.
Save bharath-kotha/feb589e494c2c29eee0c1dda2cf81f2b to your computer and use it in GitHub Desktop.
celer_mutable_dictionary_issue
from celery import Celery
from datetime import datetime
app = Celery('tasks')
# Setting the task_always_eager to to true for demonstrating the issue
# You don't want to set it to true in production environments
app.conf.task_always_eager = True
def convert_to_iso(timestamp):
''' Convert Unix timestamp to an ISO string
'''
return str(datetime.fromtimestamp(timestamp))
def save_to_db(data):
''' Save data to DB
Usually, this would include a network DB call which could fail.
'''
print(str(data))
@app.task(bind=True, num_retries=2)
def save_travel_details(self, details):
'''
deatils is a dict of the format
{
from: city (string),
to: city (string),
flight_number: flight number (string),
time: timestamp epoch (int)
}
'''
# Conver the timestamp to ISO
details['time'] = convert_to_iso(details['time'])
# After changing
# details_copy = details.copy()
# details_copy['time'] = convert_to_iso(details['time'])
try:
# Sae to database
save_to_db(details)
except Exception as e:
# If save fails, retry saving later
self.retry(exc=e)
if __name__=='__main__':
flight_details = {
'from': 'Hyderabad',
'to': 'Bangalore',
'flight_number': 'PQ126BN',
'time': 1623490200,
}
save_travel_details.delay(flight_details)
from unittest import mock
import tasks
import pytest
@mock.patch('tasks.save_travel_details.signature')
def test_save_travel_details_success(signature_mock):
flight_details = {
'from': 'Hyderabad',
'to': 'Bangalore',
'flight_number': 'PQ126BN',
'time': 1623490200,
}
tasks.save_travel_details.delay(flight_details)
signature_mock.assert_not_called()
@mock.patch('tasks.save_to_db')
@mock.patch('tasks.save_travel_details.signature')
def test_save_travel_details_error(signature_mock, db_mock):
flight_details = {
'from': 'Hyderabad',
'to': 'Bangalore',
'flight_number': 'PQ126BN',
'time': 1623490200,
}
db_mock.side_effect = Exception()
signature_mock.return_value = mock.Mock()
tasks.save_travel_details.delay(flight_details)
signature_mock.assert_called_once()
signature_mock.assert_called_with(flight_details)
#print(signature_mock.mock_calls)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment