변수에 함수 자체를 할당하거나, 파라미터로 함수를 전달하거나, 다른 함수가 기존 함수를 호출할 수도 있다.f(g(x))
데코레이터 도입 전, classmethod, staticmethod 함수로 기존 메서드의 정의.
def original(...):
....
original = modifier(original)
함수를 동일한 이름으로 다시 할당(original)하고 있다.
이것은 혼란스럽고 오류가 발생하기 쉽고 번거롭다(함수 재할당을 잊어버리거나 함수 정의가 멀리 떨어져 있어서 귀찮음)
데코레이터를 적용하면 이렇게 된다.
@modifier
def original(...):
...
데코레이터는 데코레이터 이후에 나오는 것을 첫 번째 파라미터로 하고, 데코레이터의 결괏값을 반환하게 하는 Syntax sugar다.
한 곳에서 함수의 전체 정의를 파악할 수 있기 때문에 가독성이 향상됐다.
구조를 살펴보자.
modifer는 데코레이터라 하고,
original을 decorated 또는 wrapped 객체라고 한다.
주의) 데코레이터 디자인 패턴과는 다른 용어다.
# decorator_function_1.py
import logging as logger
from functools import wraps
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("%s 재시도", operation.__qualname__)
last_raised = e
return wrapped
@retry
def run_operation(task):
"""실행중 예외가 발생할 것으로 예상되는 특정 작업을 실행"""
return task.run()
class WrongTask:
@classmethod
def run(self):
raise ControlledException()
run_operation(WrongTask)
데코레이터를 적용하지 않은 클래스의 코드를 먼저 살펴보자.
from datetime import datetime
from dataclasses import dataclass
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"),
}
@dataclass
class LoginEvent:
SERIALIZER = LoginEventSerializer
username: str
password: str
ip: str
timestamp: datetime
def serialize(self) -> dict:
return self.SERIALIZER(self).serialize()
def hide_password():
return "**민감한 정보**"
class LoginEventSerializer:
def __init__(self, event):
self.event = event
def serialize(self) -> dict:
return {
"username": self.event.username,
"password": hide_password(),
"ip": self.event.ip,
"timestamp": self.event.timestamp.strftime("%Y-%m-%d %H:%M"),
}
class PasswordChangedEventSerializer:
def __init__(self, event):
self.event = event
def serialize(self) -> dict:
return {
"password": hide_password(),
}
- 표준화 : serialize() 메서드는 모든 이벤트 클래스에 있어야만 한다. <- 고민!
다른 해결 방법
from datetime import datetime
from dataclasses import dataclass
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 field, transformation in self.serialization_fields.items()
}
class Serialization:
def __init__(self, **transformations):
self.serializer = EventSerializer(transformations)
def __call__(self, event_instance):
def serialize_method(event_instance):
return self.serializer.serialize(event_instance)
event_instance.serialize = serialize_method
return event_instance
@Serialization(
username=show_original,
password=hide_field,
ip=show_original,
timestamp=format_time,
)
@dataclass
class LoginEvent:
username: str
password: str
ip: str
timestamp: datetime
res_1 = LoginEvent(
username="hoon",
password="12341234",
ip="where_is_my_ip",
timestamp=datetime.now(),
)
print(res_1) # LoginEvent(username='hoon', password='12341234', ip='where_is_my_ip', timestamp=datetime.datetime(2023, 11, 27, 20, 0, 29, 254364))
@Serialization(
password=hide_field,
)
@dataclass
class PasswordChangedEvent:
password: str
res_2 = PasswordChangedEvent(
password="55555555",
)
print(res_2) # PasswordChangedEvent(password='55555555')
import logging as logger
from collections.abc import Sequence
from typing import Optional
from functools import wraps
_DEFAULT_RETRIES_LIMIT = 3
class ControlledException(Exception):
"""도메인에서 발생하는 일반적인 예외"""
def with_retry(
retries_limit: int = _DEFAULT_RETRIES_LIMIT,
allowed_exceptions: Optional[Sequence[Exception]] = 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.warning("%s 재시도, 원인: %s", operation.__qualname__, e)
last_raise = e
return wrapped
return retry
class NormalTask:
@classmethod
def run(self):
pass
@with_retry
def run_operation(task):
return task.run()
@with_retry(retries_limit=5)
def run_with_custom_retries_limit(task):
return task.run()
@with_retry(allowed_exceptions=(AttributeError,))
def run_with_custom_retries_limit(task):
return task.run()
@with_retry(
retries_limit=5,
allowed_exceptions=(
ZeroDivisionError,
AttributeError,
),
)
def run_with_custom_parameters(task):
return task.run()
import logging as logger
from collections.abc import Any, Sequence
from typing import Optional
from functools import wraps
_DEFAULT_RETRIES_LIMIT = 3
class ControlledException(Exception):
"""도메인에서 발생하는 일반적인 예외"""
class WithRetry:
def __init__(
self,
retries_limit: int = _DEFAULT_RETRIES_LIMIT,
allowed_exceptions: Optional[Sequence[Exception]] = None,
) -> 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(self.retries_limit):
try:
return operation(*args, **kwargs)
except self.allowed_exceptions as e:
logger.warning("%s 재시도, 원인: %s", operation.__qualname__, e)
last_raised = e
raise last_raised
return wrapped
@WithRetry(retries_limit=5)
def run_with_custom_retries_limit(task):
return task.run()
사용자가 깜박 잊고 파라미터를 전달하지 않더라도 동작하도록 하기 위해 필요한 데코레이터 요구 사항이다.
기본값이 있는 경우와 없는 경우 모두 지원하는 데코레이터를 구현해보자.
from functools import wraps
DEFAULT_X = 3
DEFAULT_Y = 4
def decorator(function=None, *, x=DEFAULT_X, y=DEFAULT_Y):
if function is None:
def decorated(function):
@wraps(function)
def wrapped():
return function(x, y)
return wrapped
return decorated
else:
@wraps(function)
def wrapped():
return function(x, y)
return wrapped
@decorator(x=3, y=5)
def my_function(x, y):
return x + y
res = my_function()
print(res)
from functools import wraps, partial
DEFAULT_X = 3
DEFAULT_Y = 4
def decorator(function=None, *, x=DEFAULT_X, y=DEFAULT_Y):
if function is None:
# return partial(decorator, x=x, y=y)
return lambda f: decorator(f, x=x, y=y)
@wraps(function)
def wrapped():
return function(x, y)
return wrapped
@decorator(x=3, y=5)
def my_function(x, y):
return x + y
res = my_function()
print(res)
from functools import wraps
X, Y = 1, 2
def decorator(callable):
"""고정된 X와 Y 값으로 <callable> 호출"""
@wraps(callable)
def wrapped():
return callable(X, Y)
return wrapped
@decorator
def func(x, y):
return x + y
@decorator
async def coro(x, y):
return x + y
def _log(f, *args, **kwargs):
print(f"함수이름: {f.__qualname__!r}, 파라미터: {args=} 와 {kwargs=}")
return f(*args, **kwargs)
@(lambda f: lambda *args, **kwargs: _log(f, *args, **kwargs))
def func(x):
return x + 1
func(3)
함수이름: 'func', 파라미터: args=(3,) 와 kwargs={}
def resolver_function(root, args, context, info):
pass
def resolver_function(root, args, context, info):
helper = DomainObject(root, args, context, info)
...
helper.process()
@DomainArgs
def resolver_function(helper):
helper.process()
추가로, DbC(Design by Contract) 원칙에 따라 사전조건, 사후조건을 강제할 수도 있다.
추적(tracing)이란 다음과 같은 시나리오에서 사용하려는 것으로
모니터링하고자 하는 함수의 실행과 관련한 것이다.