mirror-github-bspeice/metrik
1
0
Fork 0
mirror of https://github.com/bspeice/metrik synced 2024-11-23 07:38:09 -05:00
Code Issues Releases Wiki Activity

Another update to the cron scripts

Browse Source
This commit is contained in:
Bradlee Speice 2016-08-24 22:47:11 -04:00
parent 1ba75cab31
commit 8fb5fa9c32
3 changed files with 14 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
metrik/__init__.py
View File

@ -1,2 +1,2 @@
__version__ = '0.3.0' __version__ = '0.3.1'
__release__ = __version__ __release__ = __version__

15
metrik/batch.py
View File

@ -4,6 +4,8 @@ from datetime import datetime
from argparse import ArgumentParser from argparse import ArgumentParser
from dateutil.parser import parse from dateutil.parser import parse
from six import StringIO from six import StringIO
from subprocess import check_output
from os import path
from metrik.conf import get_config from metrik.conf import get_config
from metrik.flows.rates_flow import LiborFlow from metrik.flows.rates_flow import LiborFlow
@ -23,15 +25,24 @@ def run_flow(flow_class, present, live):
def build_cron_file(): def build_cron_file():
EXEC = 'metrik' EXEC = 'metrik'
FLOW_FLAG = '-f' FLOW_FLAG = '-f'
USER = 'root'
metrik_command_location = path.dirname(check_output(['which', 'metrik']))
cron_header = '''# cron.d entries for metrik
SHELL=/bin/sh
PATH={}
'''.format(metrik_command_location)
cron_strings = [] cron_strings = []
for flow_name, flow_class in flows.items(): for flow_name, flow_class in flows.items():
cron_string = flow_class.get_schedule().get_cron_string() cron_string = flow_class.get_schedule().get_cron_string()
cron_strings.append( cron_strings.append(
cron_string + ' ' + EXEC + ' ' + FLOW_FLAG + ' ' + flow_name cron_string + ' ' + USER + ' ' + EXEC + ' ' +
FLOW_FLAG + ' ' + flow_name
) )
return '\n'.join(cron_strings) + '\n' return cron_header + '\n'.join(cron_strings) + '\n'
def list_flows(): def list_flows():

1
test/test_batch.py
View File

@ -14,5 +14,4 @@ class BatchTest(TestCase):
def test_cron_string(self): def test_cron_string(self):
cron_string = build_cron_file() cron_string = build_cron_file()
assert len(cron_string.split('\n')) == len(flows) + 1
assert cron_string[-1] == '\n' assert cron_string[-1] == '\n'