AOPの考え方と同じ
★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()