Fabric でタスク実行前後にタスクの内容を通知する

Fabric 使っていて、自分の用途が結構タスク量多かったので、タスク実行前と完了時に何のタスクを行うのか、行ったのかを出力してほしいなと思い CustomTask クラス作った。

report.py

# -*- coding: utf-8 -*-

import logging
import time

from fabric.tasks import Task
from fabric.operations import *
from fabric.colors import *

class ReportStatus(Task):
    """
    タスク実行前後にそのタスクをこれから実行する旨と実行終えた旨を伝える CustomTask クラス

    @task(task_class=ReportStatus, task_name="実行前後表示テスト")
    def something_task():
        print green("タスク実行")
    """
    def __init__(self, func, task_name, *args, **kwargs):
        super(ReportStatus, self).__init__(*args, **kwargs)
        self.func = func
        self.task_name = task_name

        if hasattr(func, '__doc__') and func.__doc__:
            self.__doc__ = func.__doc__
        else:
            self.__doc__ = task_name

        if hasattr(callable, '__module__'):
            self.__module__ = callable.__module__

    def run(self, *args, **kwargs):
        print cyan("[%s] [%s] を実行します" % (self.task_name, self.func.__name__))
        logging.info("[%s][%s] executing" % (self.func.__name__, self.task_name))

        result = self.func(*args, **kwargs)

        print green("[%s] [%s] を完了しました" % (self.task_name, self.func.__name__), bold=True)
        logging.info("[%s][%s] done" % (self.func.__name__, self.task_name))
        print "\n"

        return result

    def __call__(self, *args, **kwargs):
        return self.run(*args, **kwargs)

    def __getattr__(self, k):
        return getattr(self.func, k)

    def __details__(self):
        return get_task_details(self.func)

logging 使って log も吐くようにしてるので適宜設定してあげるとログ出力もする

fabfile.py

# -*- coding: utf-8 -*-

import datetime
import logging
from fabric.colors import *
from fabric.api import *

from report import ReportStatus

log_file_name = "%s.log" % (datetime.datetime.today().strftime("%Y%m%d"))
logging.basicConfig(
    format='[%(asctime)s][%(levelname)s] %(message)s',
    filename=log_file_name, datefmt='%Y/%m/%d %I:%M:%S',
    level=logging.INFO
)

@task(task_class=ReportStatus, task_name="実行前後表示テスト")
def something_task():
    print green("タスク実行")

実行するとこんな感じ。

log は指定したフォーマットで出力される

[2017/01/19 01:20:18][INFO] [something_task][実行前後表示テスト] executing
[2017/01/19 01:20:18][INFO] [something_task][実行前後表示テスト] done