Decoratorとwith

★Decorator

メソッド式

def deco1(func):

def overwrite():

...

func()

return overwrite

def deco2(func):

import functools

@functools.wraps(func)

def overwrite(*args, **kwargs):

...

func(*args, **kwargs)

return overwrite

def deco3(param):

def wrapper(func):

import functools

@functools.wraps(func)

def overwrite(*args, **kwargs):

...

func(*args, **kwargs)

return overwrite

return wrapper

@deco1

def test1():

...

test1()

print(test1.__name__) # overwrite

@deco2

def test2():

...

test2()

print(test2.__name__) # test2

@deco3('dummy')

def test3():

...

クラス式

class Foo(object):

def __init__(self, func):

self._func = func

def __call__(self, *args, **kwargs):

print("{func} is running".format(func=self.func.__name__))

# 前処理

self._func(*args, **kwargs)

# 後処理

class Foo2(object):

def __init__(self, level='INFO'):

self._level = level

def __call__(self, func):

def wrapper(*args, **kwargs):

print("[{level}]:{func} is running" \

.format(level=self._level, func=func.__name__)

# 前処理

self._func(*args, **kwargs)

# 後処理

return wrapper

@Foo

def bar():

...

@Foo2(level='WARN')

def bar():

...

複数の場合に処理順番

@a

@b

@c

def f():

pass

f = a(b(c(f))) と同じ

計測

import time

(async) def measure(func):

import functools

@functools.wraps(func)

def overwrite(*args, **kwargs):

start = int( time.time() * 1000 )

result = (await) func(*args, **kwargs)

end = int( time.time() * 1000 )

print("実行時間：{0}".format(end - start) + "[ms]")

return result

return overwrite

※asyncとawaitをつけると、非同期処理に対応

キャッシュ

from functools import lru_cache

@lru_cache()

# @lru_cache(maxsize=128, typed=False)

def fibonacci(n):

if n < 2:

return n

return fibonacci(n-2) + fibonacci(n-1)

Generic化

from functools import singledispatch

from collections import abc

import numbers

import html

@singledispatch

def htmlize(obj)

content = html.escape(repr(obj))

return f'<pre>{content}</pre>'

@htmlize.register(str)

def _(text):

content = html.escape(text).replace('\n', '<br>\n')

return f'<p>{content}</p>'

@htmlize.register(numbers.Integral)

def _(n):

return f'<pre>{n} (0x{0:x})</pre>'

@htmlize.register(tuple)

@htmlize.register(abc.MutableSequence)

def _(seq):

inner = '</li>\n<li>'.join(htmlize(item) for item in seq)

return '<ul>\n<li>' + inner + '</li>\n</ul>'

Strategy Pattern

promos = []

def promotion(func):

promos.append(func)

return func

@promotion

def strategyA(order):

return ...

@promotion

def strategyB(order):

return ...

def best_promo(order):

return max(promo(order) for promo in promos)

同期ロック

import threading

import functools

def synchronized(func):

@functools.wraps(func)

def wrapper(self, *args, **kwargs):

with self.lock:

return func(self, *args, **kwargs)

return wrapper

class MyWorker(object):

def __init__(self):

self.lock = threading.Lock()

self.idx = 0

@synchronized

def test1(self):

self.idx = self.idx + 1

@synchronized

def test2(self):

self.idx = self.idx + 2

worker = MyWorker()

threading.Thread(target=worker.test1).start()

threading.Thread(target=worker.test2).start()

★with

file = open("filename")

try:

print(file.read())

finally:

file.close()

↓

with open("filename") as file:

print(file.read())

with open('file1', 'w') as f1, open('file2', 'w') as f2:

f1.write('Hello')

f2.write('Hello')

方式１

class CustomOpen(object):

def __init__(self, filename):

self.file = open(filename)

def __enter__(self):

return self.file

def __exit__(self, type, value, traceback):

self.file.close()

with CustomOpen('file') as f:

contents = f.read()

方式２

from contextlib import contextmanager

@contextmanager

def custom_open(filename):

f = open(filename)

try:

yield f

finally:

f.close()

with custom_open('file') as f:

contents = f.read()