5. 데코레이터를 사용한 코드 개선

hyuckhoon.ko·2023년 11월 27일
0

1. 파이썬의 데코레이터

- PEP-318에서 기존 함수와 메서드의 기능을 쉽게 수정하기 위한 수단으로 소개 됨

- 파이썬에서 함수는 일반적인 객체일 뿐이다.

변수에 함수 자체를 할당하거나, 파라미터로 함수를 전달하거나, 다른 함수가 기존 함수를 호출할 수도 있다.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)

클래스 데코레이터

  • 함수 테코레이터와 유사하다
  • 단점은 '데코레이터와 관심사의 분리'에서 확인하자
  • 장점
    • DRY 원칙 준수
    • 당장은 작고 간단한 클래스를 생성하고, 나중에 데코레이터로 기능 보강 가능
    • 클래스 유지보수 시, 데코레이터로 기존 로직을 쉽게 변경 가능

데코레이터를 적용하지 않은 클래스의 코드를 먼저 살펴보자.

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()
  • LoginEvent가 직접 매핑할 시리얼라이저 클래스를 선언한다.
  • 비밀번호필드를 숨기고, timestamp 필드를 포매팅한다.

문제점

  • 클래스가 많아진다.
    • 이벤트 클래스와 직렬화 클래스가 1대1 관계
    • password를 가진 다른 이벤트 클래스가 생기면, 구조 변경 리팩터링이 필요
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')


2. 고급 데코레이터

1. 중첩 함수를 사용한 데코레이터

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()

2. 데코레이터 객체

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)
  • 데코레이터의 파라미터는 키워드 전용이다. x, y를 넘기는 경우, function의 값이 None이 됨.

대안

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)

코루틴(coroutine)을 위한 데코레이터

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)이란 다음과 같은 시나리오에서 사용하려는 것으로
모니터링하고자 하는 함수의 실행과 관련한 것이다.

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

0개의 댓글