Memoru

System Engineering and Programming and IT

watchdog

tree /f

PS G:\workspace\py\sandbox\watchdog_sample> tree /f
Folder PATH listing for volume Users
Volume serial number is 04A1-7D2F
G:.
│  app.py
│  apps.log
│  configbase.py
│  logging_debug.conf
│  watchdog_sample.py
│
├─configs
│  │  config.py
│  │  config_debug.py
│  │  default.py
│  │
│  └─__pycache__
│
├─plugins
│  │  a.py
│  │  b.py
│  │  c.py
│  │
│  └─__pycache__
│
└─__pycache__

watchdog

#! /usr/bin/env python
"""{fname}
# sample code for module : watchdog
Usage:
"""

__author__ = 'Memoru'
__version__ = '0.0.1'
__date__ = '2018-06-23'

import sys
import os
__doc__ = __doc__.format(fname=os.path.basename(__file__))

import logging.config
import configbase as base
CONF = base.CONF
ENV = base.ENV


import sys
import time
from importlib import import_module, reload
from pathlib import Path

from watchdog.events import FileSystemEvent, PatternMatchingEventHandler
from watchdog.observers import Observer


class PluginManager:

    class Handler(PatternMatchingEventHandler):
        def __init__(self, manager: 'PluginManager', *args, **kwargs):
            super().__init__(*args, **kwargs)
            self.manager = manager

        def on_created(self, event: FileSystemEvent):
            print(event)
            if event.src_path.endswith('.py'):
                self.manager.load_plugin(Path(event.src_path))

        def on_modified(self, event):
            print(event)

    def __init__(self, path: str):
        self.plugins = {}
        self.path = path
        self.observer = Observer()
        sys.path.append(self.path)

    def start(self):
        self.scan_plugin()
        self.observer.schedule(self.Handler(self, patterns='*.py'), self.path)
        self.observer.start()

    def stop(self):
        self.observer.stop()
        self.observer.join()

    def scan_plugin(self):
        for file_path in Path(self.path).glob('*.py'):
            self.load_plugin(file_path)

    def load_plugin(self, file_path):
        module_name = file_path.stem
        if module_name not in self.plugins:
            self.plugins[module_name] = import_module(module_name)
            print('{} loaded.'.format(module_name))
        else:
            self.plugins[module_name] = reload(self.plugins[module_name])
            print('{} reloaded.'.format(module_name))


def main():

    plugin_manager = PluginManager('plugins/')
    plugin_manager.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        plugin_manager.stop()


#  Debug ---------------------------------------  #

def process():
    try:
        logger = logging.getLogger(__name__)
        logger.debug('◇◇--Begin sample -----------------◇◇')
        logger.debug(__doc__)
        main()
        logger.debug('◆◆--End   sample -----------------◆◆')
    except Exception as ex:
        logger.exception(ex)
        raise ex
        

if __name__ == '__main__':
    logging.config.fileConfig('logging_debug.conf')
    process()

execution

(sandbox) PS G:\workspace\py\sandbox\watchdog_sample> python .\watchdog_sample.py
[2018/06/27 07:10:43][DEBUG](watchdog_sample.py:91) ◇◇--Begin sample -----------------◇◇
[2018/06/27 07:10:43][DEBUG](watchdog_sample.py:92) watchdog_sample.py
# sample code for module : watchdog
Usage:

a loaded.
b loaded.
c loaded.

json.dump

json_sample.py

#! /usr/bin/env python
"""{fname}
# sample code for csv
Usage:
"""

__author__ = 'Memoru'
__version__ = '0.0.1'
__date__ = '2018-06-26'

import sys
import os
__doc__ = __doc__.format(fname=os.path.basename(__file__))

import json
from pathlib import Path

import logging.config
import configbase as base
CONF = base.CONF
ENV = base.ENV



#  Debug ---------------------------------------  #

def process(args={'json_file' : 'sample.json'}):
    try:
        logger = logging.getLogger(__name__)
        logger.debug('◇◇--Begin sample -----------------◇◇')
        logger.debug(__doc__)
        logger.debug(f'CONF.MEDIA_ROOT={CONF.MEDIA_ROOT}')
        csv_root = CONF.MEDIA_ROOT
        p = Path(csv_root)
        jsonf = args['json_file']
        built_p = p / jsonf
        logger.debug(built_p)
        
        obj = { 'id' : '123456789_1',
                'contact_id' : '123456789',
                'c_no' : '1',
                'top_class' : '挨拶',
                'text' : 'こんにちは'
              }
        f = open(built_p, 'w', encoding='utf-8')
        json.dump(obj, f ,ensure_ascii=False, indent=4)
        
        
        logger.debug('◆◆--End   sample -----------------◆◆')
    except Exception as ex:
        logger.exception(ex)
        raise ex
        

if __name__ == '__main__':
    logging.config.fileConfig('logging_debug.conf')
    #process()
    args = {'json_file' : 'sample2.json'}
    process(args)

csv

csv_sample.py

#! /usr/bin/env python
"""{fname}
# sample code for csv
Usage:
"""

__author__ = 'Memoru'
__version__ = '0.0.1'
__date__ = '2018-06-26'

import sys
import os
__doc__ = __doc__.format(fname=os.path.basename(__file__))

import csv
from pathlib import Path

import logging.config
import configbase as base
CONF = base.CONF
ENV = base.ENV



#  Debug ---------------------------------------  #

def process(args={'csv_file' : 'sample.csv'}):
    try:
        logger = logging.getLogger(__name__)
        logger.debug('◇◇--Begin sample -----------------◇◇')
        logger.debug(__doc__)
        logger.debug(f'CONF.MEDIA_ROOT={CONF.MEDIA_ROOT}')
        csv_root = CONF.MEDIA_ROOT
        p = Path(csv_root)
        csvf = args['csv_file']
        built_p = p / csvf
        logger.debug(built_p)
        
        if built_p.exists() :
            with open(built_p, 'r', encoding='shift_jis') as f:
                reader = csv.reader(f)
                head = next(reader)
                logger.debug(head)
                if(len(head)==2):
                    pass
                else:
                    raise ValueError('header error')
            
        
        logger.debug('◆◆--End   sample -----------------◆◆')
    except Exception as ex:
        logger.exception(ex)
        raise ex
        

if __name__ == '__main__':
    logging.config.fileConfig('logging_debug.conf')
    #process()
    args = {'csv_file' : 'sample2.csv'}
    process(args)

configbase.py

#!/usr/bin/env python
# -*- coding:utf-8 -*-

"""
Base Config setting
 - 共通のconfig定数をconfigs.default.pyに設定する。
   CONFに読み込まれる
 - マシン環境(本番、デバック, etc)によって異なる定数の場合
   configs.config_<env.EXECUTION_MODE>.pyに、値を設定する。
   ENVに読み込まれる。
"""
__version__ = '0.0.2'

import importlib
from os import environ

# module statement ------------------------------
import configs.default as CONF  ## Common Config

class Config():
    """ configs配下のconfig_xxx.pyよりConfigを読み込むClass"""
    def __init__(self,env=None):
        if env is None:
            if environ.get("EXECUTION_MODE") is None:
                suffix = ""
            else:
                suffix = "_" + environ.get("EXECUTION_MODE")
        else:
            suffix = "_" + env

        self.ENV = _get_conf('configs.config' + suffix) ## Environment Config
        self.LOG_CONF = 'logging' + suffix +".conf"     ## Logging Config

def _get_conf(config_name):
    conf = importlib.import_module(config_name)
    return conf

ENV = Config().ENV
LOG_CONF = Config().LOG_CONF


###### debug ---------------------------------###
# configs/config_debug.pyのMODULE_FILEを確認
def process(env):
    print(" 1  : get config from {}".format(Config(env).ENV.MODULE_FILE))
    print(" 11 : get config from {}".format(Config(env).LOG_CONF))
    print(" 2  : get config from {}".format(Config().ENV.MODULE_FILE))
    print(" 22 : get config from {}".format(Config().LOG_CONF))
    print(" 3  : get config from '{}' in default".format(CONF.PROJECT_NAME))

if __name__ == '__main__' :
    process("debug")

logging_debug.conf

[loggers]
keys=root,functions

[handlers]
keys=fileHandler,stderrHandler

[formatters]
keys=simpleFormatter

[logger_root]
level=DEBUG
handlers=fileHandler,stderrHandler

[logger_functions]
level=INFO
handlers=fileHandler,stderrHandler
qualname=functions
propagate=0

[handler_fileHandler]
class=handlers.TimedRotatingFileHandler
formatter=simpleFormatter
args=('apps.log','D')

[handler_stderrHandler]
class=StreamHandler
formatter=simpleFormatter
args=(sys.stderr,)

[formatter_simpleFormatter]
format=[%(asctime)s][%(levelname)s](%(filename)s:%(lineno)s) %(message)s
datefmt=%Y/%m/%d %H:%M:%S

unittest

unittest

  • unittest.main()は、unittest.TestCaseを継承した全てのクラスのメソッドのうちtestで始まるテストをテストケースとして実行する。
import unittest
import calc
from add_sample import add

class TestCalc(unittest.TestCase):
    
    def test_add2(self):
        expected = 5
        actual = add(1,4)
        print(expected, actual)
        self.assertEqual(expected, actual)
        
    def test_add(self):
        c = calc.Calc()
        expected = 5
        actual = c.add(2,3)
        print(expected, actual)
        self.assertEqual(expected, actual)
    
    def test_sub(self):
        c = calc.Calc()
        expected = -1
        actual = c.sub(2,3)
        print(expected, actual)
        self.assertEqual(expected, actual)

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

shuitl

(sandbox) PS G:\workspace\py\sandbox\standards> python
Python 3.6.5 (v3.6.5:f59c0932b4, Mar 28 2018, 16:07:46) [MSC v.1900 32 bit (Intel)] on win32
Type "help", "copyright", "credits" or "license" for more information.
>>> import shutil
>>> for i in dir(shutil):
...     print(i)
...
Error
ExecError
ReadError
RegistryError
SameFileError
SpecialFileError
_ARCHIVE_FORMATS
_BZ2_SUPPORTED
_LZMA_SUPPORTED
_UNPACK_FORMATS
_ZLIB_SUPPORTED
__all__
__builtins__
__cached__
__doc__
__file__
__loader__
__name__
__package__
__spec__
_basename
_check_unpack_options
_copyxattr
_destinsrc
_ensure_directory
_find_unpack_format
_get_gid
_get_uid
_make_tarball
_make_zipfile
_ntuple_diskusage
_rmtree_safe_fd
_rmtree_unsafe
_samefile
_unpack_tarfile
_unpack_zipfile
_use_fd_functions
chown
collections
copy
copy2
copyfile
copyfileobj
copymode
copystat
copytree
disk_usage
errno
fnmatch
get_archive_formats
get_terminal_size
get_unpack_formats
getgrnam
getpwnam
ignore_patterns
make_archive
move
nt
os
register_archive_format
register_unpack_format
rmtree
stat
sys
unpack_archive
unregister_archive_format
unregister_unpack_format
which

method on dict

(sandbox) PS G:\workspace\py\sandbox\standards> python
Python 3.6.5 (v3.6.5:f59c0932b4, Mar 28 2018, 16:07:46) [MSC v.1900 32 bit (Intel)] on win32
Type "help", "copyright", "credits" or "license" for more information.
>>> dir({})
['__class__', '__contains__', '__delattr__', '__delitem__', '__dir__', '__doc__', '__eq__', '__format__', '__ge__', '__getattribute__', '__getitem__', '__gt__', '__hash__', '__init__', '__init_subclass__', '__iter__', '__le__', '__len__', '__lt__', '__ne__', '__new__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__setitem__', '__sizeof__', '__str__', '__subclasshook__', 'clear', 'copy', 'fromkeys', 'get', 'items', 'keys', 'pop', 'popitem', 'setdefault', 'update', 'values']
>>> for item in dir({}):
...     print(item)
...
__class__
__contains__
__delattr__
__delitem__
__dir__
__doc__
__eq__
__format__
__ge__
__getattribute__
__getitem__
__gt__
__hash__
__init__
__init_subclass__
__iter__
__le__
__len__
__lt__
__ne__
__new__
__reduce__
__reduce_ex__
__repr__
__setattr__
__setitem__
__sizeof__
__str__
__subclasshook__
clear
copy
fromkeys
get
items
keys
pop
popitem
setdefault
update
values
>>>