파이썬 클린코드를 읽으며 정리한 내용입니다.
def original():
pass
def modifier(org):
pass
original = modifier(original)
@modifier
def original():
pass
# 특정 예외에 대해서 일정 횟수 재시도하는 데코레이터 예시
class ControlledException(Exception):
"""도메인에서 발생하는 일반적인 예외"""
def retry(operation):
@wraps(operation)
def wrapped(*args, **kwargs):
last_raised = None
RETRIES_LIMIT = 3
for _ in range(RETRIES_LIMIT):
try:
return operation(*args, **kwargs)
except ControlledException as e:
logger.info("retrying %s", operation.__qualname__)
last_raised = e
raise last_raised
return wrapped
# 사용 예시
@retry
def run_operation(task):
return task.run()
class LoginEventSerializer:
def __init__(self, event):
self.event = event
def serialize(self) -> dict:
return {
"username": self.event.username,
"password": "**민감한 정보 삭제**",
"ip": self.event.ip,
"timestamp": self.event.timestamp.strftime("%Y-%m-%d %H:%M")
}
class LoginEvent:
SERIALIZER = LoginEventSerializer
def __init__(self, username, password, ip, timestamp):
self.username = username
self.password = password
self.ip = ip
self.timestamp = timestamp
def serialize(self) -> dict:
return self.SERIALIZER(self).serialize()
from datetime import datetime
def hide_field(field) -> str:
return "**민감한 정보 삭제**"
def format_time(field_timestamp: datetime) -> str:
return field_timestamp.strftime("%Y-%m-%d %H:%M")
def show_original(event_field):
return event_field
class EventSerializer:
def __init__(self, serialization_fields: dict) -> None:
self.serialization_fields = serialization_fields
def serialize(self, event) -> dict:
return {
field: transformation(getattr(event, field))
for filed, transformation in
self.serialization_fields.items()
}
class Serialization:
def __init__(self, **transformations):
self.serializer = EventSerializer(transformations)
# 인자로 받은 event_class에 serialize를 주입
def __call__(self, event_class):
def serialize_method(event_instance):
return self.serializer.serialize(event_instance)
event_class.serialize = serialize_method
return event_class
@Serialization(
username=show_original,
password=hide_field,
ip=show_original,
timestamp=format_time
)
class LoginEvent:
def __init__(self, username, password, ip, timestamp):
self.username = username
self.password = password
self.ip = ip
self.timestamp = timestamp
from dataclasses import dataclass
@Serialization(
username=show_original,
password=hide_field,
ip=show_original,
timestamp=format_time
)
@dataclass
class LoginEvent:
username: str
password: str
ip: str
timestamp: datetime
데코레이터는 함수, 메서드, 클래스뿐만 아니라 제너레이터, 코루틴, 심지어 이미 데코레이트된 객체에도 사용할 수 있다.
앞의 예시에 나온것처럼, 데코레이터는 스택 형태로 쌓일 수 있다.
데코레이터가 코루틴으로 사용되는 것도 좋은 예시이다. 새로 생성된 제너레이터에 데이터를 보내기 전 next()를 호출하는 작업은 잊어버리기 쉬운데, 이 경우 제너레이터를 파라미터로 받아 next()를 호출한 후 제너레이터를 반환하는 데코레이터를 만들면 쉽게 해결된다.
코루틴이란? cooperative routine를 의미하는데 서로 협력하는 루틴이라는 뜻. 즉, 메인 루틴과 서브 루틴처럼 종속된 관계가 아니라 서로 대등한 관계이며 특정 시점에 상대방의 코드 실행.
중첩 함수의 데코레이터
파라미터를 데코레이터에 전달하려면 세 단계의 중첩 함수가 필요하다.
@retry(arg1, arg2, ...)
이 구문은 다음과 같다. <original_function> = retry(arg1, arg2, ...)(<original_function>)
RETRIES_LIMIT = 3
def with_retry(retries_limit=RETRIES_LIMIT, allowed_exceptions=None):
allowed_exceptions = allowed_exceptions or (ControlledException,)
def retry(operation):
@wraps(operation)
def wrapped(*args, **kwargs):
last_raised = None
for _ in range(retries_limit):
try:
return operation(*args, **kwargs)
except allowed_exceptions as e:
logger.info("retrying %s due to %s", operation, e)
last_raised = e
raise last_raised
return wrapped
return retry
데코레이터 객체
__init__()
메소드에 파라미터를 전달하고, __call__()
에서 데코레이터 로직을 구현하면 된다.class WithRetry:
def __init__(self, retries_limit=RETRIES_LIMIT, allowed_exceptions=None):
self.retries_limit = retries_limit
self.allowed_exceptions = allowed_exceptions or (ControlledException)
def __call__(self, operation):
@wraps(operation)
def wrapped(*args, **kwargs):
last_raised = None
for _ in range(retries_limit):
try:
return operation(*args, **kwargs)
except allowed_exceptions as e:
logger.info("retrying %s due to %s", operation, e)
last_raised = e
raise last_raised
return wrapped
__init__
에서 정해진 로직에 따라 초기화를 진행한다.run_with_custom_retries_limit
함수를 래핑해 __call__
메서드를 호출한다.__call__
은 원본 함수를 래핑하여 새로운 함수를 반환한다.파라미터 변환
코드 추적
def trace_decorator(function):
def wrapped(*args, **kwargs):
logger.info("%s 실행", function.__qualname__)
return function(*args, *kwargs)
return wrapped
@trace_decorator
def process_account(account_id):
"""id별 계정 처리"""
logger.info("%s 계정 처리", account_id)
help(process_account)
>>>
Help on function wrapped in module __main__:
wrapped(*args, **kwargs)
print(process_account.__name__)
>>> wrapped
print(process_account.__qualname__)
>>> trace_decorator.<locals>.wrapped
from functools import wraps
def trace_decorator(function):
@wraps(function)
def wrapped(*args, **kwargs):
logger.info("%s 실행", function.__qualname__)
return function(*args, *kwargs)
return wrapped
@trace_decorator
def process_account(account_id):
"""id별 계정 처리"""
logger.info("%s 계정 처리", account_id)
help(process_account)
>>>
Help on function process_account in module __main__:
process_account(account_id)
id별 계정 처리
print(process_account.__name__)
>>> process_account
print(process_account.__qualname__)
>>> process_account
process_account.__wrapped__
>>> <function __main__.process_account(account_id)>
데코레이터 부작용의 잘못된 처리
def traced_function_wrong(function):
logger.info("%s 실행", function)
start_time = time.time()
@wraps(function)
def wrapped(*args, **kwargs):
result = function(*args, **kwargs)
logger.info("함수 %s의 실행시간: %.2fs", function, time.time() - start_time)
return result
return wrapped
wrapped
에 들어가있지 않은 두 줄이 실행된다.wrapped
안으로 넣어주면 된다.데코레이터 부작용의 활용
EVENTS_REGISTRY = {}
def register_event(event_cls):
"""이벤트 클래스를 레지스트리에 등록"""
EVENTS_REGISTRY[event_cls.__name__] = event_cls
return event_cls
class Event:
"""이벤트 객체"""
class UserEvent:
TYPE = "user"
@register_event
class UserLoginEvent(UserEvent):
"""사용자가 시스템에 접근했을 때 발생"""
@register_event
class UserLogoutEvent(UserEvent):
"""사용자가 시스템에서 나갈 때 발생"""
EVENT_REGISTRY
에 들어가게 된다.args
와 **kwargs
서명을 사용하여 데코레이터를 정의하면 모든 경우에 사용할 수 있다. 하지만 가독성 측면에서나 사용성 측면에서나 원래 함수의 서명과 비슷하게 데코레이터를 정의하는 게 더 좋을 수 있다.import logging
from functools import wraps
logger = logging.getLogger(__name__)
class DBDriver:
def __init__(self, dbstring):
self.dbstring = dbstring
def execute(self, query):
return f"{self.dbstring}에서 쿼리 {query} 실행"
def inject_db_driver(function):
"""데이터베이스 dns 문자열을 받아 DBDriver 인스턴스를 생성하는 데코레이터
"""
@wraps(function)
def wrapped(dbstring):
return function(DBDriver(dbstring))
return wrapped
@inject_db_driver
def run_query(driver):
return driver.execute("test_function")
run_query("test_db")
>>> 'test_db에서 쿼리 test_function 실행'
self
가 있어 동작하지 않는다.class DataHandler:
@inject_db_driver
def run_query(self, driver):
return driver.execute(self.__class__.__name__)
class inject_db_driver:
"""데이터베이스 dns 문자열을 받아 DBDriver 인스턴스를 생성하는 데코레이터
"""
def __init__(self, function):
self.function = function
wraps(self.function)(self)
def __call__(self, dbstring):
return self.function(DBDriver(dbstring))
def __get__(self, instance, owner):
if instance is None:
return self
# instance에 function을 동적으로 bind하여 리턴한 결과를 __class__로 다시 래핑
# 이것을 사용하려는 클래스에 클래스 변수로 넣어주면 호출될 때 __get__호출
return self.__class__(MethodType(self.function, instance))
__get__
메서드를 사용하지 않기 때문에 여전히 잘 동작한다.@app.task
: 이 데코레이터는 많은 로직과 코드를 래핑하지만 그 중 어떤것도 래핑하는 함수와 관련이 없다. 또한 아무도 데코레이터가 하는 일을 살펴볼 필요가 없다.@route
: 이것을 이용해 데코레이팅된 함수는 url로 등록된다. 이 또한 래핑하는 함수와 관련이 없으며, 함수를 매퍼에 등록하여 url에 연결하거나 원래 함수 서명을 변경하여 http 요청 객체를 수신하게 해준다. 이를 통해 더 깔끔한 인터페이스를 제공한다.