leadtime.py

from flask import Flask, request, redirect, url_for
import mysql.connector
import numpy as np
import json
import datetime
from datetime import timedelta
import sys
# pip install python-dateutil
from dateutil.relativedelta import relativedelta

app = Flask(__name__)

@app.route('/')
def leadtime():
    name  = "service1"
    d     = "1d"
    d_op  = d[-1]
    d     = int(d[0]) * 7 if d[1] == "w" else int(d[0]) 
    start = '2017/11/20 00:00:00'
    end   = '2018/3/10 00:00:00'
    # notice: 後述する str"ft"time ではなく str"pt"time
    start_datetime = datetime.datetime.strptime(start, "%Y/%m/%d %H:%M:%S")
    end_datetime   = datetime.datetime.strptime(end, "%Y/%m/%d %H:%M:%S")

    return leadtime_daily(name, d, d_op, start_datetime, end_datetime)

def leadtime_daily(name, d, d_op, start_datetime, end_datetime):
    ret  = []
    conn = db_connect()
    cur  = conn.cursor(dictionary=True)
    sql   = "select * from sample where name = %s and start_time >= %s and start_time < %s and end_time < %s"
    try:
        while start_datetime < end_datetime:
            # get (start ~ start + [d]day) data
            # if end_time = null, we skip such line
            if(d_op == "d" or d_op == "w"):
                start_datetime_add_delta = start_datetime + timedelta(d)
            elif(d_op == "m"):
                start_datetime_add_delta = start_datetime + relativedelta(months = d)
            else:
                raise 
            # SQLの実行
            cur.execute(sql, (
                name,
                start_datetime.strftime("%Y/%m/%d %H:%M:%S"),
                start_datetime_add_delta.strftime("%Y/%m/%d %H:%M:%S"),
                end_datetime.strftime("%Y/%m/%d %H:%M:%S")
                ))
            # 平均の計算
            leadtime_ave = calc_leadtime_ave(cur.fetchall())
 
            # JSON形式に整形
            ret.append({"date": start_datetime.strftime("%Y/%m/%d"), "leadtime": round(leadtime_ave, 3)})
            start_datetime = start_datetime_add_delta
        return json.dumps({
            "status": "ok",
            "result": {
                "type": "leadtimeAve",
                "data": ret
                }
            })
    except Exception as e:
        return json.dumps({
            "status": "ng",
            "result": {
                "type": "error",
                "data": ""
                }
            })
    finally:
        cur.close()
        conn.close()

def calc_leadtime_ave(lines):
    leadtime_sum = timedelta(0) # 初期化
    count = 0
    for line in lines:
        # notice: start=2/25 12:00:00 end=2/26 00:00:00 の場合 end-start = 0.5(日)となる  
        leadtime_sum += line["end_time"] - line["start_time"]
        count += 1
    return (leadtime_sum / timedelta(count)) if count != 0 else 0

def db_connect():
    conn = mysql.connector.connect(
        host = 'localhost',
        port = 3306,
        user = 'root',
        password = 'hogehoge',
        database = "practice"
    )
    return conn

if __name__ == '__main__':
    #app.debug = True
    app.run(host='0.0.0.0') # host='0.0.0.0'は重要

leadtime_test.py

import sys, os
sys.path.append(os.path.dirname(os.path.abspath(__file__)) + '/../')
import index
import unittest
import tempfile
import datetime
from datetime import timedelta

class IndexTestCase(unittest.TestCase):

    def setUp(self):
        #self.db_fd, flaskr.DATABASE = tempfile.mkstemp()
        self.app = index.app.test_client()
        #flaskr.init_db()

    #def tearDown(self):
        #os.close(self.db_fd)
        #os.unlink(flaskr.DATABASE)

    def test_response(self):
        response_goodreq = self.app.get('/')
        self.assertEqual(response_goodreq.status_code, 200)
        response_badreq = self.app.get('/hoge')
        self.assertEqual(response_badreq.status_code, 404)

    def test_calc_leadtime_ave(self):
        lines = []
        lines = self.create_testcase_calc_leadtime_ave('2018/3/5 00:00:00', '2018/3/10 00:00:00', lines)
        result = index.calc_leadtime_ave(lines)
        self.assertEqual(result, 5.0)
        lines = self.create_testcase_calc_leadtime_ave('2018/2/28 00:00:00', '2018/3/10 00:00:00', lines)
        result = index.calc_leadtime_ave(lines)
        self.assertEqual(result, 7.5)

    # testcase生成
    # 第一引数はselfでないといけない
    # 呼び出し時にはselfは必要ない
    def create_testcase_calc_leadtime_ave(self, start, end, lines):
        start_datetime = datetime.datetime.strptime(start, "%Y/%m/%d %H:%M:%S")
        end_datetime   = datetime.datetime.strptime(end, "%Y/%m/%d %H:%M:%S")
        lines.append({"start_time": start_datetime, "end_time": end_datetime})
        return lines

if __name__ == '__main__':
    unittest.main()