데코레이터

목표

• 파이썬에서 데코레이터가 동작하는 방식을 이해
• 함수와 클래스에 적용되는 데코레이터를 구현하는 방법을 배운다.
• 일반적인 실수를 피하여 데코레이터를 효과적으로 구현하는 방법을 배운다.
• 데코레이터를 활용한 코드 중복을 회피(DRY 원칙 준수)
• 데코레이터를 활용한 관심사의 분리
• 좋은 데코레이터 사례
• 데코레이터가 좋은 선택이 될 수 있는 일반적인 상황, 관용구, 패턴

파이썬의 데코레이터

• 함수와 메서드의 기능을 쉽게 수정하기 위한 수단으로 나옴.
• original 함수가 있고 그 기능을 약간 수정한 modifier라는 함수가 있다면 원래 아래처럼 썼다.

def original():
    pass

original = modifier(original)

• 위 방법은 혼란스럽고 오류가 발생하기 쉽고 번거롭다.
• 따라서 데코레이터가 도입되었고, 아래처럼 쓴다.

@modifier
def original():
    pass

• 데코레이터는 데코레이터 이후에 나오는 것을 데코레이터의 첫 번째 파라미터로 하고, 데코레이터의 결과값을 반환하게 하는 syntax suger일 뿐이다.

함수 데코레이터

• param 유효성 검사, 사전조건 검사, 기능 전체 새롭게 정의, signature 새로 정의, 원래 함수 결과 캐싱 등을 할 수 있다.
• 특정 예외에 대해 특정 횟수만큼 재시도 하는 데코레이터를 만들어보자.

from functools import wraps

class ControlledException(Exception):
    pass


def retry(operation):
    @wraps(operation)
    def wrapped(*args, **kwargs):
        last_raised = None
        RETRY_LIMIT = 3
        for _ in RETRY_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()

• 위 데코레이터는 operation = retry(operation)의 syntax sugar다.

클래스 데코레이터

• 클래스 데코레이터는 인자로 클래스를 받는다.

• 클래스 데코레이터가 복잡하고 가독성이 떨어진다는 얘기도 많다.

  • 클래스에서 정의한 속성과 메서드를 데코레이터 안에서 완전히 다른 용도로 변경할 수 있기 때문에

• 우선은 클래스 데코레이터의 장점먼저 알아보자

  • 재사용, DRY원칙 지키게 해줌
  • 여러 클래스가 특정 인터페이스나 기준을 따르도록 강제할 수 있음
  • 당장은 작고 간단한 클래스를 만들고, 추후에 데코레이터로 기능 보강 가능
  • 유시보수 시 데코레이터를 사용해 기존 로직을 훨씬 쉽게 변경할 수 있음

• 아래 첫번쨰 예시 방법은 잘 동작하지만, 확장성에 문제가 있다.

  • 클래스가 많아짐. 이벤트 : 직렬화 1대1이므로 이벤트 개수만큼 많아짐
  • 유연성 없음. password숨기는 기능만 쓰고 싶을 떄 불가능
  • 표준화 문제. serializer()메서드가 모든 이벤트 클래스에 있어야 함

import unittest
from datetime import datetime


class LoginEventSerializer:
    def __init__(self, event):
        self.event = event

    def serialize(self) -> dict:
        return {
            "username": self.event.username,
            "password": "**redacted**",
            "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()


class TestLoginEventSerialized(unittest.TestCase):
    def test_serializetion(self):
        event = LoginEvent(
            "username", "password", "127.0.0.1", datetime(2016, 7, 20, 15, 45)
        )
        expected = {
            "username": "username",
            "password": "**redacted**",
            "ip": "127.0.0.1",
            "timestamp": "2016-07-20 15:45",
        }
        self.assertEqual(event.serialize(), expected)


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

개선버전

• 이벤트 인스턴스와 변형 함수를 필터로 받아 동적으로 객체를 만듬.
• 필터를 이벤트 인스턴스의 필드들에 적용해 직렬화 함.

import unittest
from datetime import datetime


def hide_field(field) -> str:
    return "**redacted**"


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:
    """Apply the transformations to an Event object based on its properties and
    the definition of the function to apply to each field.
    """

    def __init__(self, serialization_fields: dict) -> None:
        """Created with a mapping of fields to functions.
        Example::
        >>> serialization_fields = {
        ...    "username": str.upper,
        ...    "name": str.title,
        ... }
        Means that then this object is called with::
        >>> from types import SimpleNamespace
        >>> event = SimpleNamespace(username="usr", name="name")
        >>> result = EventSerializer(serialization_fields).serialize(event)
        Will return a dictionary where::
        >>> result == {
        ...     "username": event.username.upper(),
        ...     "name": event.name.title(),
        ... }
        True
        """
        self.serialization_fields = serialization_fields

    def serialize(self, event) -> dict:
        """Get all the attributes from ``event``, apply the transformations to
        each attribute, and place it in a dictionary to be returned.
        """
        return {
            field: transformation(getattr(event, field))
            for field, transformation in self.serialization_fields.items()
        }


class Serialization:
    """A class decorator created with transformation functions to be applied
    over the fields of the class instance.
    """

    def __init__(self, **transformations):
        """The ``transformations`` dictionary contains the definition of how to
        map the attributes of the instance of the class, at serialization time.
        """
        self.serializer = EventSerializer(transformations)

    def __call__(self, event_class):
        """Called when being applied to ``event_class``, will replace the
        ``serialize`` method of this one by a new version that uses the
        serializer instance.
        """

        def serialize_method(event_instance):
            return self.serializer.serialize(event_instance)

        event_class.serialize = serialize_method
        return event_class


@Serialization(
    username=str.lower,
    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


class TestLoginEventSerialized(unittest.TestCase):
    def test_serialization(self):
        event = LoginEvent(
            "UserName", "password", "127.0.0.1", datetime(2016, 7, 20, 15, 45)
        )
        expected = {
            "username": "username",
            "password": "**redacted**",
            "ip": "127.0.0.1",
            "timestamp": "2016-07-20 15:45",
        }
        self.assertEqual(event.serialize(), expected)


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

another 개선버전

• dataclass를 사용해서 더 간단해진 예제

import sys
import unittest
from datetime import datetime

from decorator_class_2 import (
    Serialization,
    format_time,
    hide_field,
    show_original,
)

try:
    from dataclasses import dataclass
except ImportError:

    def dataclass(cls):
        return cls


@Serialization(
    username=show_original,
    password=hide_field,
    ip=show_original,
    timestamp=format_time,
)
@dataclass
class LoginEvent:
    username: str
    password: str
    ip: str
    timestamp: datetime


class TestLoginEventSerialized(unittest.TestCase):
    @unittest.skipIf(
        sys.version_info[:3] < (3, 7, 0), reason="Requires Python 3.7+ to run"
    )
    def test_serializetion(self):
        event = LoginEvent(
            "username", "password", "127.0.0.1", datetime(2016, 7, 20, 15, 45)
        )
        expected = {
            "username": "username",
            "password": "**redacted**",
            "ip": "127.0.0.1",
            "timestamp": "2016-07-20 15:45",
        }
        self.assertEqual(event.serialize(), expected)


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

다른 유형의 데코레이터

• 제너레이터, 코루틴, 이미 데코레이터 붙은 객체에도 데코레이터 사용이 가능하다.
• 데코레이터의 또 다른 좋은 예는 코루틴으로 사용되는 제너레이터다.
• 새로 생성된 제너레이터에 데이터를 보내기 전에 next()를 호출하여 다음 yield문으로 넘어가야 한다는 것이다.
• 이러한 수작업은 모든 사용자가 기억해야 하는 것으로 에러나가 쉽다.
• 이런 경우 제너레이터를 파라미터로 받아 next()를 호출한 다음 다시 제너레이터를 반환하는 데코레이터를 만들면 쉽게 해결된다.

데코레이터에 인자 전달

• 두 가지 방법이 있다.

1. 간접 참조(indirection)를 통해 새로운 레벨의 중첩 함수를 만들어 데코레이터의 모든 것을 한 단계 더 깊게 만드는 것.
2. 클래스로 데코레이터 정의. (1보다 가독성이 더 좋다.)

중첩 함수 데코레이터

• retry 횟수, allow exception을 인자로 받는 retry 데코레이터 만들어보자.

from functools import wraps

from decorator_function_1 import ControlledException
from log import logger


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.warning(
                        "retrying %s due to %s", operation.__qualname__, e
                    )
                    last_raised = e
            raise last_raised

        return wrapped

    return retry


@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_exceptions(task):
    return task.run()


@with_retry(
    retries_limit=4, allowed_exceptions=(ZeroDivisionError, AttributeError)
)
def run_with_custom_parameters(task):
    return task.run()

데코레이터 객체

from functools import wraps
from unittest import TestCase, main, mock

from decorator_function_1 import (ControlledException, OperationObject,
                                  RunWithFailure)
from log import logger


class Retry:
    def __init__(self, operation):
        self.operation = operation
        wraps(operation)(self)

    def __call__(self, *args, **kwargs):
        last_raised = None
        RETRIES_LIMIT = 3
        for _ in range(RETRIES_LIMIT):
            try:
                return self.operation(*args, **kwargs)
            except ControlledException as e:
                logger.info("retrying %s", self.operation.__qualname__)
                last_raised = e
        raise last_raised


@Retry
def run_operation(task):
    """Run the operation in the task"""
    return task.run()


class RetryDecoratorTest(TestCase):
    def setUp(self):
        self.info = mock.patch("log.logger.info").start()

    def tearDown(self):
        self.info.stop()

    def test_fail_less_than_retry_limit(self):
        """Retry = 3, fail = 2 --> OK"""
        task = OperationObject()
        failing_task = RunWithFailure(task, fail_n_times=2)
        times_run = run_operation(failing_task)

        self.assertEqual(times_run, 3)
        self.assertEqual(task._times_called, 3)

    def test_fail_equal_retry_limit(self):
        """Retry = fail = 3, will fail"""
        task = OperationObject()
        failing_task = RunWithFailure(task, fail_n_times=3)
        with self.assertRaises(ControlledException):
            run_operation(failing_task)

    def test_no_failures(self):
        task = OperationObject()
        failing_task = RunWithFailure(task, fail_n_times=0)
        times_run = run_operation(failing_task)

        self.assertEqual(times_run, 1)
        self.assertEqual(task._times_called, 1)

    def test_doc(self):
        self.assertEqual(
            run_operation.__doc__, "Run the operation in the task"
        )


if __name__ == "__main__":
    main()

데코레이터 활용 우수 사례

• 파라미터 변환
• 코드 추적(파라미터와 함꼐 함수의 실행을 로깅하려는 경우)
• 파라미터 유효성 검사
• 재시도 로직 구현
• 일부 반복 작업을 데코레이터로 이동하여 클래스 단순환

파라미터 변환

• 파라미터 유효성 검사
• 사전조건, 사후조건 강제
• 일반적으로 파라미터 다룰 때 데코레이터 많이 사용함.
• 특히 유사한 객체를 반복적으로 생성하거나 추상화를 위해 유사한 변형을 반복하는 경우

코드 추적

• 실제 함수의 실행 경로 추적
• 함수 지표 모니터링(CPU 사용률, 메모리 사용량 등)
• 함수의 실행 시간 측정
• 언제 함수가 실행되고 전달된 파라미터는 무엇인지 로깅

데코레이터의 활용 - 흔한 실수 피하기

래핑된 원복 객체의 데이터 보존

• 데코레이터를 달면 기존 함수의 docstring, 함수 이름 등등이 데코레이터에 의해 오버라이드 된다.
• 이를 방지하기 위해 functools.wraps 데코레이터를 사용한다.

def decorator(original_function):
  @wraps(original_function):
    def decorated_function(*args, **kwargs):
      # do something
      return original_function(*args, **kwargs)
    return decorated_function

데코레이터를 만들 때는 항상 래핑된 함수 위에 functools.wraps를 사용한다.

데코레이터 부작용 처리

• 데코레이터 함수가 되기 위해 필요한 하나의 조건은 가장 안쪽에 정의된 함수여야 한다는 것이다.
• 그렇지 않으면 임포트에 문제가 될 수 있다.
• 때로는 임포트 시에 실행하기 위해 이러한 부작요이 필요한 경우도 있고, 반대로 있다.

잘못된 예시

• 시작 시간 찍는 부분이 밖으로 나와있기 때문에, 시작시간은 런타임이 아닌 임포트 타임에 찍힌다.

def traced_function_wrong(function):
    """An example of a badly defined decorator."""
    logger.debug("started execution of %s", function)
    start_time = time.time()

    @wraps(function)
    def wrapped(*args, **kwargs):
        result = function(*args, **kwargs)
        logger.info(
            "function %s took %.2fs", function, time.time() - start_time
        )
        return result

    return wrapped


@traced_function_wrong
def process_with_delay(callback, delay=0):
    logger.info("sleep(%d)", delay)
    return callback

올바른 예시

• 위 문제를 아래처럼 해결한다.
• 문제 부분을 가장 안쪽 함수에 넣어주면 된다.

def traced_function(function):
    @wraps(function)
    def wrapped(*args, **kwargs):
        logger.info("started execution of %s", function)
        start_time = time.time()
        result = function(*args, **kwargs)
        logger.info(
            "function %s took %.2fs", function, time.time() - start_time
        )
        return result

    return wrapped


@traced_function
def call_with_delay(callback, delay=0):
    logger.info("sleep(%d)", delay)
    return callback

데코레이터 부작용 활용

• 모듈의 공용 레지스트리에 객체를 등록할 떄, 일부러 임포트타임에 실행되는 방법을 사용한다.
• 각 클래스마다 처리 여부를 flag표시 하는 대신, 아래처럼 처리한다.
• 많은 웹프레임워크, 라이브러리들이 이 원리를 객체를 노출하고 활용한다.

EVENTS_REGISTRY = {}


def register_event(event_cls):
    """Place the class for the event into the registry to make it accessible in
    the module.
    """
    EVENTS_REGISTRY[event_cls.__name__] = event_cls
    return event_cls


class Event:
    """A base event object"""


class UserEvent:
    TYPE = "user"


@register_event
class UserLoginEvent(UserEvent):
    """Represents the event of a user when it has just accessed the system."""


@register_event
class UserLogoutEvent(UserEvent):
    """Event triggered right after a user abandoned the system."""


def test():
    """
    >>> sorted(EVENTS_REGISTRY.keys()) == sorted(('UserLoginEvent', 'UserLogoutEvent'))
    True

어느 곳에서나 동작하는 데코레이터 만들기

• 일반적으로 함수를 위한 데코레이터와 클래스를 위한 데코레이터는 호환이 안된다.
• 디스크립터로 이문제 해결

from functools import wraps
from types import MethodType


class DBDriver:
    def __init__(self, dbstring):
        self.dbstring = dbstring

    def execute(self, query):
        return f"query {query} at {self.dbstring}"


class inject_db_driver:
    """Convert a string to a DBDriver instance and pass this to the wrapped
    function.
    """

    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
        return self.__class__(MethodType(self.function, instance))


@inject_db_driver
def run_query(driver):
    return driver.execute("test_function_2")


class DataHandler:
    @inject_db_driver
    def run_query(self, driver):
        return driver.execute("test_method_2")

데코레이터와 DRY 원칙

• 데코레이터로 DRY원칙 잘 따를 수 있다.
• 하지만 데코레이터 쓰면 코드 복잡성이 증가한다.
• 복잡성을 댓가로 치룰만큼 가치가 있을 때 써야한다.

  처음부터 데코레이터를 만들지 않는다.
  패턴이 생기고 추상화가 명확해지면 그때 만든다.
  적어도 3회 이상 필요한 경우에만 데코레이터를 만든다.
  데코레이터 코드를 최소한으로 유지한다.

데코레이터와 관심사의 분리

• 하나의 데코레이터는 한가지 일만 해야한다.

좋은 데코레이터 분석

• 캡슐화와 관심사의 분리: 내부에서 어떻게 구현했는지 전혀 알 수 없는 블랙박스 모드로 동작해야 한다.

• 독립성: 데코레이팅되는 객체와 최대한 분리되어야 한다.

• 재사용성

• celery의 @app.task, flask의 `@route("/", method=["GET"])은 좋은 예시다.