I have this method on my views.py
file:
def getHistoricRates():
""" Here we have the function that will retrieve the historical rates from fixer.io, since last month """
rates = {}
response = urlopen('http://data.fixer.io/api/2018-12-31?access_key=c2f5070ad78b0748111281f6475c0bdd')
data = response.read()
rdata = json.loads(data.decode(), parse_float=float)
rates_from_rdata = rdata.get('rates', {})
for rate_symbol in ['USD', 'GBP', 'HKD', 'AUD', 'JPY', 'SEK', 'NOK', 'EUR']:
try:
rates[rate_symbol] = rates_from_rdata[rate_symbol]
except KeyError:
logging.warning('rate for {} not found in rdata'.format(rate_symbol))
pass
return rates
@require_http_methods(['GET', 'POST'])
def historical(request):
date_str = "2018-12-31"
if datetime.datetime.strptime(date_str,"%Y-%m-%d").weekday()<5:
rates = getHistoricRates()
fixerio_rates = [Fixerio_rates(currency=currency, rate=rate)
for currency, rate in rates.items()]
Fixerio_rates.objects.bulk_create(fixerio_rates)
return render(request, 'historical.html')
I want to run historical
at 9am every day, except for weekends.
Now, I can't find any example on how to run an existent method, or how to call it from the cron.py
file whatsoever.
I do have everything configured, for django_cron
but I just can't figure out how to "use" this method from my cron file to run it at specific times.
This is my cron.py
file so far:
from django_cron import CronJobBase, Schedule
from .views import historical
class MyCronJob(CronJobBase):
RUN_AT_TIMES = ['9:00']
schedule = Schedule(run_at_times=RUN_AT_TIMES)
code = 'fixerio.my_cron_job' # a unique code
def do(self):
pass # do your thing here
The name fixerio
is my app's name.
Any ideas on this?
Since you're looking to call the getHistoricRates()
and the bulk_create()
logic from both your historical()
view and also a cron job, it would be better to first refactor that common code from the view into a separate module - for example into helpers.py
that lives alongside the views.py
and cron.py
.
helpers.py
from .models import Fixerio_rates
def create_rates():
rates = getHistoricRates()
fixerio_rates = [Fixerio_rates(currency=currency, rate=rate)
for currency, rate in rates.items()]
Fixerio_rates.objects.bulk_create(fixerio_rates)
def getHistoricRates():
...
You can then invoke that from your cron job:
cron.py
from .helpers import create_rates
class MyCronJob(CronJobBase):
RUN_AT_TIMES = ['9:00']
schedule = Schedule(run_at_times=RUN_AT_TIMES)
code = 'fixerio.my_cron_job' # a unique code
def do(self):
create_rates()
And also from your view:
views.py
from .helpers import create_rates
@require_http_methods(['GET', 'POST'])
def historical(request):
date_str = "2018-12-31"
if datetime.datetime.strptime(date_str,"%Y-%m-%d").weekday()<5:
create_rates()
return render(request, 'historical.html')