Skip to content

Instantly share code, notes, and snippets.

@Shinichi-Nakagawa
Last active May 7, 2017 06:52
Show Gist options
  • Save Shinichi-Nakagawa/4ec8c81f0d60cc44326f to your computer and use it in GitHub Desktop.
Save Shinichi-Nakagawa/4ec8c81f0d60cc44326f to your computer and use it in GitHub Desktop.
DockerのPython Imageとparse-crontabを使ってシンプルな定時処理batchを作る ref: http://qiita.com/shinyorke/items/77cd54bc836227b2416a
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import logging
from multiprocessing import Pool
from scheduler.job import JobController
__author__ = 'Shinichi Nakagawa'
# Docker ImageのTimezoneがUTCなので注意!
@JobController.run("20 15 * * 5")
def notice_tmr_club():
"""
タモリ倶楽部の時間だお(東京)
:return: None
"""
logging.info("タモリ倶楽部はじまるよ!!!")
# Docker ImageのTimezoneがUTCなので注意!(大切なので2回言いました)
@JobController.run("00 9 * * *")
def notice_baseball():
"""
やきうの時間を教えるお
:return: None
"""
logging.info("やきうの時間だあああ!!!!")
def main():
"""
crontabを動かすmethod
:return: None
"""
# ログ設定(Infoレベル、フォーマット、タイムスタンプ)
logging.basicConfig(
level=logging.INFO,
format="time:%(asctime)s.%(msecs)03d\tprocess:%(process)d" + "\tmessage:%(message)s",
datefmt="%Y-%m-%d %H:%M:%S"
)
# crontabで実行したいジョブを登録
jobs = [notice_tmr_club, notice_baseball]
# multi process running
p = Pool(len(jobs))
try:
for job in jobs:
p.apply_async(job)
p.close()
p.join()
except KeyboardInterrupt:
logging.info("exit")
if __name__ == '__main__':
main()
batch:
build: .
dockerfile: ./Dockerfile
command: python batch.py
container_name: python_crontab_example
# Python crontab sample
FROM python:3.5.1
MAINTAINER Shinichi Nakagawa <spirits.is.my.rader@gmail.com>
# add to application
RUN mkdir /app
WORKDIR /app
ADD requirements.txt /app/
RUN pip install -r requirements.txt
ADD ./scheduler /app/scheduler/
ADD *.py /app/
$ docker-compose up
Creating python_crontab_example
Attaching to python_crontab_example
python_crontab_example | time:2015-12-13 13:45:09.463 process:9 message:->- Process Start
python_crontab_example | time:2015-12-13 13:45:09.464 process:8 message:->- Process Start
python_crontab_example | time:2015-12-13 13:45:09.465 process:9 message:-?- next running schedule:2015-12-18 15:20:00
python_crontab_example | time:2015-12-13 13:45:09.465 process:8 message:-?- next running schedule:2015-12-14 09:00:00
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import time
import functools
import logging
from crontab import CronTab
from datetime import datetime, timedelta
import math
__author__ = 'Shinichi Nakagawa'
class JobController(object):
"""
ジョブ実行Controller
"""
@classmethod
def run(cls, crontab):
"""
処理実行
:param crontab: job schedule
"""
def receive_func(job):
@functools.wraps(job)
def wrapper():
job_settings = JobSettings(CronTab(crontab))
logging.info("->- Process Start")
while True:
try:
logging.info(
"-?- next running\tschedule:%s" %
job_settings.schedule().strftime("%Y-%m-%d %H:%M:%S")
)
time.sleep(job_settings.interval())
logging.info("->- Job Start")
job()
logging.info("-<- Job Done")
except KeyboardInterrupt:
break
logging.info("-<- Process Done.")
return wrapper
return receive_func
class JobSettings(object):
"""
出力設定
"""
def __init__(self, crontab):
"""
:param crontab: crontab.CronTab
"""
self._crontab = crontab
def schedule(self):
"""
次回実行
:return: datetime
"""
crontab = self._crontab
return datetime.now() + timedelta(seconds=math.ceil(crontab.next()))
def interval(self):
"""
次回実行までの時間
:return: seconds
"""
crontab = self._crontab
return math.ceil(crontab.next())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment