批处理应用程序（Django）

2 年 ago
清, 扬
4 minutes

首先

我会描述使用Django创建批处理应用程序的步骤。本次示例应用程序已在GitHub上公开。

前提条件

请参考README文件以了解操作环境、设置和批处理应用程序的启动步骤。

Django 是什么

Django（强哥）是用Python实现的Web应用程序框架，它包含了一般Web应用程序开发所需的所有功能。更多详情请参阅Django文档。

通过使用Django丰富的功能，你也可以创建批处理应用程序。

批处理应用程序（django）

我将以Django为基础来解释以下内容。

应用程序配置

アプリケーション構成

设定信息

数据库连接信息、日志级别等在yml文件和django的settings.py文件中进行定义。这些可以根据操作系统和主机名进行更改，因为需要在开发/生产环境之间切换。

設定ファイル切替

設定信息的定義

在开发/正式环境下发生改变的设置值要在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ログレベル開発WindowsDEBUG本番LinuxINFO

应用配置

# 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.namewindowsntlinux or macposixplatform.systemwindowsWindowslinuxLinuxmacDarwin

批处理执行

用命令来运行批处理应用程序。

python /home/pypeach_django/manage.py batch_main create_employees

批量启动

继承django的BaseCommand类并定义批处理启动类，通过指定多个参数，实现灵活的批处理启动。

命令

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)对数据库进行select、insert和update的操作来演示。

员工服务

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数据库事务。

信息 (Messa-ji)

メッセージ

消息定义

将消息的ID和文本设置到本地化文件中。

django.po ：英文 ，汉化为 “Django.汉密”

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ロケールファイル
✖
广告
将在 10 秒后关闭
bannerAds