バッチアプリケーション(django)

はじめに

djangoを使用したバッチアプリケーションを作成する際の手順を記述します
今回のサンプルアプリケーションはgithubに公開しています

前提事項

動作環境やセットアップ及びバッチアプリケーション起動手順はREADMEを参照ください

djangoとは

django（ジャンゴ）はPythonで実装されたWebアプリケーションフレームワークです
一般的なWebアプリケーション開発に必要な機能は一通り揃っています
詳細はDjangoドキュメントを参照ください

djangoの豊富な機能を使用することでバッチアプリケーションも作成できます

バッチアプリケーション(django)

バッチアプリケーション(django)をベースに以降を説明します

アプリケーション構成

アプリケーション構成は以下にしています。

設定情報

DB接続情報やログレベル等はymlファイルとdjangoのsettings.pyに定義します
これらは開発/本番環境の切り替えが必要なためOSやホスト名で変更できるようにします

設定情報の定義

開発/本番で変わる設定値はymlファイルに定義します
djangoで設定値を参照する定義をsettings.pyに行います
djangoのデフォルトに追加や変更した箇所を抜粋します

# DBの接続情報を設定する
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.mysql',
        'NAME': AppConfig.get_properties("database_name"),
        'USER': AppConfig.get_properties("database_user"),
        'PASSWORD': AppConfig.get_properties("database_password"),
        'HOST': AppConfig.get_properties("database_host"),
        'PORT': AppConfig.get_properties("database_port"),
        'OPTIONS': {
            'charset': 'utf8mb4'
        },
        'TEST': {
            'NAME': AppConfig.get_properties("test_database_name"),
        }
    }
}
# Localeのパス設定を行う
LOCALE_PATHS = (
    os.path.join(BASE_DIR, 'locale'),
)

# loggingの設定を行う
LOGGING = {
    'version': 1,
    'disable_existing_loggers': False,
    'handlers': {
        'consoleHandler': {
            'level': 'DEBUG',
            'class': 'logging.StreamHandler',
            'formatter': 'customFormatter'
        },
        'fileRotatingHandler': {
            'level': 'DEBUG',
            'class': 'logging.handlers.TimedRotatingFileHandler',
            'formatter': 'customFormatter',
            'filename': AppConfig.get_log_file(),
            'encoding': 'utf8',
            'when': 'D',
            'interval': 1,
            'backupCount': 7
        }
    },
    'formatters': {
        'customFormatter': {
            'format': '[%(asctime)s] %(levelname)s - %(filename)s#%(funcName)s:%(lineno)d: %(message)s',
            'datefmt': '%Y/%m/%d %H:%M:%S',
        },
    },
    'loggers': {
        '': {
            'handlers': ['consoleHandler', 'fileRotatingHandler'],
            'level': AppConfig.get_properties("app_log_level"),
            'propagate': False,
        },
        'django.db.backends': {
            'handlers': ['consoleHandler'],
            'level': AppConfig.get_properties("sql_log_level"),
            'propagate': False,
        }
    }
}

設定情報の取得

開発/本番環境に応じた設定値の取得を行います
以下の設定を切り替えます

環境	OS	ログレベル
開発	Windows	DEBUG
本番	Linux	INFO

AppConfig

# coding:utf-8
import os
import socket

import yaml

"""
アプリケーションの設定を制御する
"""

class AppConfig:

    @staticmethod
    def get_properties(key):
        """
        設定ファイルからキーに紐づく値を取得する
        """
        base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
        resource_file = base_dir + "/resource/application_" + AppConfig.get_application_config() + ".yml"

        with open(os.path.join(resource_file), 'r', encoding='UTF-8') as f:
            config = f.read()
        return yaml.load(config)[key]

    @staticmethod
    def get_log_file():
        """
        設定ファイルからキーに紐づく値を取得する
        """
        return os.path.join(AppConfig.get_properties("log_path"), AppConfig.get_properties("log_file"))

    @staticmethod
    def get_application_config():
        """
        OS情報やホスト名から設定ファイルの読み込み先を取得する
        """
        application_config = "develop"

        # 実行環境に応じて設定ファイルの読み込み先を切り替える
        # Linux またはホスト名で判定しています
        if os.name == 'posix' or socket.gethostname() == 'example.com':
            application_config = "production"
        return application_config

以下を使用して適宜実行環境の判別を行います

タイプ	OS	値
os.name	windows	nt
	linux or mac	posix
platform.system	windows	Windows
	linux	Linux
	mac	Darwin

バッチ実行

コマンドでバッチアプリケーションを実行します

python /home/pypeach_django/manage.py batch_main create_employees

バッチ起動

djangoのBaseCommand継承してバッチ起動クラスを定義します
複数のparameterを指定可能にすることで柔軟なバッチ起動が実現できます

Command

import gettext
import logging

from django.core.management.base import BaseCommand
from django.db import ProgrammingError
from django.utils.translation import gettext

from app_pypeach_django.application.service.employees_service import EmployeesService

"""
BaseCommandを継承したバッチ起動クラスです。
"""

class Command(BaseCommand):

    def add_arguments(self, parser):
        """
        引数をセットする
        """
        parser.add_argument('parameter', nargs='+', type=str)

    def handle(self, *args, **options):
        """
        コマンド実行時のハンドラ。
        引数に応じて各サービスを実行する
        """
        execute_batch = None

        for index, parameter in enumerate(options['parameter']):
            if index == 0:
                execute_batch = parameter

        logging.info(gettext("I900"), execute_batch)

        try:
            if execute_batch == 'create_employees':
                EmployeesService.create_employees()
            elif execute_batch == 'create_departments':
                EmployeesService.create_departments()
            elif execute_batch == 'update_employees':
                EmployeesService.update_employees()
            elif execute_batch == 'select_employees':
                EmployeesService.select_employees()
            elif execute_batch == 'truncate_employees':
                EmployeesService.truncate_employees()
            elif execute_batch == 'create_scrapy_html':
                ScrapyService.create_scrapy_html()
            elif execute_batch == 'parse_scrapy_html':
                ScrapyService.parse_scrapy_html()
            else:
                logging.info(gettext("E902"), execute_batch)
        except ProgrammingError as e:
            logging.exception(gettext("E903"), e)
        except Exception as e:
            logging.exception(gettext("E990"), e)

        logging.info(gettext("I901"), execute_batch)

サービス実行

バッチ起動クラスからサービスを実行します
モデル(models)を使用してDBにselectとinsert&updateを行うサンプルにしています

EmployeesService

from django.db import transaction, connection
from django.utils import timezone
from django.utils.timezone import localtime

from app_pypeach_django.application.enums.department_type import DepartmentType
from app_pypeach_django.application.enums.gender_type import GenderType
from app_pypeach_django.application.service.app_logic_base import AppLogicBaseService
from app_pypeach_django.models import Employees, Departments

"""
employeesを操作するクラスです。
"""

class EmployeesService(AppLogicBaseService):
    def __init__(self):
        super().__init__()

    @staticmethod
    @transaction.atomic()
    def create_employees():
        """
        Employeesを作成する
        """
        service = EmployeesService()

        for emp_no in range(1, 11):
            if Employees.objects.filter(emp_no=emp_no, delete_flag=0).count() == 0:
                if emp_no <= 5:
                    department_no = DepartmentType.SALES.value
                else:
                    department_no = DepartmentType.MARKETING.value
                select_model = Departments.objects.filter(department_no=department_no).values("id").first()
                # データを登録する
                service._regist_employees(select_model['id'], emp_no)

    def _regist_employees(self, department_no, emp_no):
        """
        employeesを登録する
        """
        self.regist_model = Employees()
        self.regist_model.emp_no = emp_no
        self.regist_model.department_no = department_no
        self.regist_model.gender = GenderType.MAN.value
        self.regist_model.department_date_from = "20190902"
        self.regist_model.delete_flag = 0
        self.regist_model.regist_dt = localtime(timezone.now())
        self.regist_model.update_dt = localtime(timezone.now())
        self.regist_model.save()
        return self.regist_model.id

サービスに関する特記事項は以下のとおり

• トランザクション制御

処理が異常終了した場合にロールバックさせるため@transaction.atomic()を指定します
詳細はdjangoのデータベースのトランザクションを参照ください

メッセージ

バッチにおける開始・終了やエラーのメッセージはdjangoのロケールファイル(django.po)を使用します
djangoのgettextを使用してロケールファイルに定義したメッセージを取得します

メッセージ定義

ロケールファイルにメッセージのIDと文言をセットします

django.po

msgid "I900"
msgstr "処理を開始します:%s"
msgid "I901"
msgstr "処理を終了します:%s"
msgid "I902"
msgstr "トレース情報:%s"

メッセージを定義したらdjango-adminコマンドでコンパイルします

django-admin compilemessages -l ja

メッセージ取得

gettextを使用してメッセージを取得します

logging.info(gettext("I900"), execute_batch)

参考情報

• Djangoドキュメント
• Djangoロケールファイル