DbC(Design by Contract)
관계자가 기대하는 바를 암묵적으로 코드에 삽입하는 대신
양측이 동의하는 계약을 먼저 한 다음,
계약을 어겼을 경우는 명시적으로 왜 계속할 수 없는지 예외를 발생시키는 것이다.
일반적으로 까다로운 접근법이 가장 안전하고 견고하며 업계에서도 많이 쓰이는 방식이다.
-> 어떤 방식을 택하든 중복 제거 원칙을 항상 기억해야 한다.
검증 로직을 클라이언트에 또는 함수 각각에 이중으로 두면 안 된다.
함수에 제공된 파라미터의 올바른 데이터 타입만 검사하는 계약은 의미가 없다.
mpy와 같은 도구로 쉽고 효과적으로 자동화할 수 있기 때문이다.
함수에 전달되는 객체의 속성과 반환값을 검사하고 이들이 유지해야 하는 조건을 확인하는 등의 작업을 하는 것이 실질적인 가치가 있다.
DbC와는 다른 접근 방식을 따른다.
함수의 모든 부분을 유효하지 않은 것으로부터 스스로 보호할 수 있게 하는 것이다.
하지만 DbC와 다른 철학을 가졌다는 의미는 아니며, 다른 디자인 원칙과 서로 보완 관계에 있을 수 있다는 것을 뜻한다.
에러 핸들링의 주요 목적은 무엇일까?
에러에 대해 계속 실행할지, 극복할 수 없는 오류여서 프로그램을 중단할지를 결정하는 것이다.
프로그램에서 에러를 처리하는 방법은 다음과 같다.
이 결정을 내리는 것은 견고성과 정확성 간의 트레이드 오프다.
소프트웨어 프로그램은 예상치 못한 상황에서도 실패하지 않아야 견고하다고 할 수 있다. 그러나 무조건 실패하지 않는 것이 항상 옳은 것은 아니다.
>>> db_config = {"port": 3306}
>>> db_config.get("host", "localhost")
'localhost'
>>> db_config.get("port")
3306
>>> import os
>>> os.getenv("DB_PORT", 3306)
3306
잘못된 데이터를 사용하여 계속 실행하는 것보다는 차라리 실행을 멈추는 것이 더 좋을 수 있다.
호출자에게 실패했음을 빨리 알리는 것이 좋은 선택이다.
중요한 개념이 있다.
예외는 대개 호출자에게 잘못을 알려주는 것이므로 캡슐화를 약화시킨다. 새는 추상화
함수가 발생시키는 예외는 함수가 캡슐화하고 있는 로직에 대한 것이어야 한다.
import logging
import time
from dataclasses import dataclass
logger = logging.getLogger()
@dataclass
class Connector:
pass
@dataclass
class Event:
pass
class DataTransport:
"""다양한 수준의 예외를 처리하는 예"""
_READY_BACKOFF: int = 5
_RETRY_TIMES: int = 3
def __init__(self, connector: Connector) -> None:
self._connector = connector
self.connection = None
def deliver_event(self, event: Event):
try:
self.connect()
data = event.decode()
self.send(data)
except ConnectionError as e:
logger.info("커넥션 오류 발견: %s", e)
raise
except ValueError as e:
logger.error("%r 이벤트에 잘못된 데이터 포함: %s", event, e)
raise
def connect(self):
for _ in range(self._RETRY_TIMES):
try:
self.connection = self._connector.connect()
except ConnectionError as e:
logger.info("%s: 새로운 커넥션 시도 %is", e, self._READY_BACKOFF)
time.sleep(self._RETRY_BACKOFF)
else:
return self._connection
raise ConnectionError(f"연결실패 재시도 횟수 {self._RETRY_TIMES} times")
def send(self, data: bytes):
return self.connection.send(data)
ValueError와 ConnectionError는 별 관계가 없다.
다른 유형의 오류가 함께 있는 것을 통해 책임 분산의 필요성을 느끼게 된다.
def connect_with_retry(
connector: Connector, retry_n_times: int, retry_backoff: int = 5
):
"""
<connector>를 사용해 연결을 시도함.
연결에 실패할 경우 <retry_n_times>회 만큼 재시도
재시도 사이에는 <retry_backoff>초 만큼 대기
연결에 성공하면 connection 객체 반환
재시도 횟수를 초과하여 연결에 실패하면 ConnectionError
:param connector: connect 메서드는 갖는 객체
:param retry_n_times: int: 연결 재시도 횟수
:param retry_backoff: int: 재시도 사이의 대기 시간(초)
"""
for _ in range(retry_n_times):
try:
return connector.connect()
except ConnectionError as e:
logger.info("%s: 연결 실패 (%i초 후에 연결 재시도", e, retry_backoff)
time.sleep(retry_backoff)
exc = ConnectionError(f"연결실패 ({retry_backoff}회 재시도")
logger.exception(exc)
raise exc
class DataTransport:
"""추상화 수준에 따른 예외 분리를 한 객체"""
_READY_BACKOFF: int = 5
_RETRY_TIMES: int = 3
def __init__(self, connctor: Connector):
self._connctor = connctor
self.connection = None
def deliver_event(self, event: Event):
self.connection = connect_with_retry(
self._connctor,
self._RETRY_TIMES,
self._READY_BACKOFF,
)
self.send(event)
def send(self, event: Event):
try:
return self.connection.send(evnet.decode())
except ValueError as e:
logger.error("%r 잘못된 데이터 포함: %s", event, e)
raise
TIP
아무것도 하지 않는 예외 블록을 자동으로 탐지할 수 있도록 CI 환경을 구축하자.
너무나 명확한 에러이고 무시해도 괜찮은 에러가 확실하다면 contextlib.suppress
를 활용할 수도 있다.
오류 처리 과정에서 기존 오류와 다른 새로운 오류를 발생시키고 오류 메시지를 변경할 수도 있다.
이런 경우 원래 어떤 오류가 있었는지에 대한 정보를 포함하는 것이 좋다.
Django의 manage.py를 보면 어떻게 활용하는지 확인할 수 있다.
def main():
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "plab.settings")
try:
from django.core.management import execute_from_command_line
except ImportError as exc:
raise ImportError(
"Couldn't import Django. Are you sure it's installed and "
"available on your PYTHONPATH environment variable? Did you "
"forget to activate a virtual environment?"
) from exc
execute_from_command_line(sys.argv)
if __name__ == "__main__":
main()
AssertionError가 발생했다면,
프로그램에서 극복할 수 없는 치명적인 결함이 발견됐다는 뜻이다.
따라서, 어설션을 비즈니스 로직과 섞거나 소프트웨어의 제어 흐름 메커니즘으로 사용해서는 안 된다.
좋지 않은 예시
try:
assert condition.holds()
except AssertionError:
alternative_procedure()
더 이상 처리가 불가능한 상황을 의미하므로
catch 후에 프로그램을 계속 실행하면 안 된다.
함수 호출은 부작용을 가질 수 있으며, 항상 반복 가능하지 않다. 다시 호출했을 때도 같은 결과가 나올지 확신할 수 없다.
또한, 디버거를 사용해 해당 라인에서 중지하여 오류 결과를 편리하게 볼 수 없다.
예외는 예상하지 못한 상황을 처리하기 위한 것이고,
어설션은 정확성을 보장하기 위해 스스로 체크하는 것이다.
책임이 다르면 컴포넌트, 계층 또는 모듈로 분리되어야 한다.
이 섹션의 요점은 좋은 소프트웨어 관행을 약어를 통해 쉽게 기억하자는 것이다.
코드를 변경하려고 할 때 수정이 필요한 곳은 단 한군데만 있어야 한다.
중복은 기존 코드의 지식을 무시함으로써 발생한다.
from dataclasses import dataclass
@dataclass
class Student:
name: str
passed: int
failed: int
years: int
students = [
Student("1번", 10, 2, 4),
Student("2번", 8, 4, 3),
Student("3번", 12, 1, 5),
]
def process_students_list(students):
student_ranking = sorted(
students, key=lambda s: s.passed * 11 - s.failed * 5 - s.years * 2
)
for student in student_ranking:
print(
"이름: {0}, 점수: {1}".format(
student.name,
(student.passed * 11 - student.failed * 5 - student.years * 2),
)
)
process_students_list(students)
결과
이름: 2번, 점수: 62
이름: 1번, 점수: 92
이름: 3번, 점수: 117
개선
def 학생_점수_계산_비즈니스_로직(학생):
return 학생.passed * 11 - 학생.failed * 5 - 학생.years * 2
def process_students_list(students):
student_ranking = sorted(students, key=학생_점수_계산_비즈니스_로직)
for student in student_ranking:
print(
"이름: {0}, 점수: {1}".format(
student.name,
학생_점수_계산_비즈니스_로직(student)
)
)
process_students_list(students)
You Ain't Gonna Need It.
지금 당장의 요구사항을 구현한 클래스를 만들었지만 이것을 공통으로 하는 다른 클래스가 생길 것으로 예상되어 인터페이스로 만들어 버릴 수 있다. 이것은 아래의 이유로 잘못됐다.
Keep It Simple
Easier to Ask Forgivness than Permission
Look Before You Leap
# LBYL 방식
if os.path.exists(filename):
with open(filename) as f:
...
# EAFP 방식
try:
with open(filename) as f:
...
except FileNotFoundError as e:
...
EAFP 방식이 LBYL 보다 가독성이 더 좋다.
특히, LBYL의 if 문이 과연 정밀할까?
두 가지 방법 중 어떤 것을 적용할지 고민이라면 EAFP 방식을 사용하자.
즉, 전문화했다는가?
하위 클래스에서 상속된 대부분의 메서드를 필요하지 않다면,
지금 잘못하고 있는 것이다.
잘못된 예시
import collections
from datetime import datetime
class TransactionPolicy(collections.UserDict):
"""잘못된 상속의 예"""
def change_in_policy(self, customer_id, **new_policy_data):
self[customer_id].update(**new_policy_data)
policy = TransactionPolicy(
{
"client001": {
"fee": 1000.0,
"expiration_date": datetime(2022, 1, 3),
}
}
)
print(policy["client001"]) // {'fee': 1000.0, 'expiration_date': datetime.datetime(2022, 1, 3, 0, 0)}
policy.change_in_policy("client001", expiration_date=datetime(2022, 1, 4))
print(policy["client001"]) // {'fee': 1000.0, 'expiration_date': datetime.datetime(2022, 1, 4, 0, 0)}
두 가지 문제가 있다.
(1) 네이밍 잘못. TransactionPolicy라는 클래스 이름으로 사전 타입이라는 것을 짐작할 수 있을까?
(2) 결합력
아래와 같이 사용하지 않는 (불필요한) 메서드를 포함하고 있다.
'change_in_policy', 'clear', 'copy', 'data', 'fromkeys', 'get', 'items', 'keys', 'pop', 'popitem', 'setdefault', 'update', 'values'
구현 객체를 도메인 객체를 혼합하지 말자.
어떻게 개선할까?
조합을 통해 도메인 객체(TransactionPolicy)가 사전(dict) 자료구조에 사용되도록 하자.
class TransactionalPolicy:
"""컴포지션을 사용한 리팩토링 예제"""
def __init__(self, policy_data, **extra_data):
self._data = {**policy_data, **extra_data}
def change_in_policy(self, customer_id, **new_policy_data):
self._data[customer_id].update(**new_policy_data)
def __getitem__(self, customer_id):
return self._data[customer_id]
def __len__(self):
return len(self._data)
__init__
에서 사전(dict)을 _data
로써, 접근제어자로 사용하고 있음을 알리고 있다.
__getitem__
, __len__
을 함께 구현함으로써, TransactionalPolicy 가 사전의 프록시가 되도록 하고 있다.)
class BaseModule:
module_name = "top"
def __init__(self, module_name):
self.name = module_name
def __str__(self):
return f"{self.module_name}:{self.name}"
class BaseModule1(BaseModule):
module_name = "module-1"
class BaseModule2(BaseModule):
module_name = "module-2"
class BaseModule3(BaseModule):
module_name = "module-3"
class ConCreteModuleA12(BaseModule1, BaseModule2):
"""BaseModule1, BaseModule2 확장"""
class ConCreteModuleB23(BaseModule2, BaseModule3):
"""BaseModule2, BaseModule3 확장"""
print(ConCreteModuleA12("test")) // module-1:test
print([cls.__name__ for cls in ConCreteModuleA12.mro()]) // ['ConCreteModuleA12', 'BaseModule1', 'BaseModule2', 'BaseModule', 'object']
코드를 재사용하기 위해 일반적인 행동을 캡슐화해 놓은 부모 클래스다.
class BaseTokenizer:
def __init__(self, str_token):
self.str_token = str_token
def __iter__(self):
yield from self.str_token.split("-")
token = BaseTokenizer("28a2320b-fd3f-4627-9792-a2b38e3c46b0")
print(list(token)) // ['28a2320b', 'fd3f', '4627', '9792', 'a2b38e3c46b0']
BaseTokenizer를 변경하지 않고 값을 대문자로 변환해보자.
class UpperIterableMixin:
def __iter__(self):
return map(str.upper, super().__iter__())
class Tokenizer(UpperIterableMixin, BaseTokenizer):
pass
token1 = Tokenizer("28a2320b-fd3f-4627-9792-a2b38e3c46b0")
print(list(token1))
가변인자를 사용하려면 해당 인자를 패킹할 변수의 이름 앞에 별표(*)를 사용한다.
def f(first, second, third):
print(first)
print(second)
print(third)
l = [1, 2, 3]
f(*l)
3개의 위치인자를 갖는 함수 f에 가변인자를 전달했다.
패킹되어 전달된 가변 인자가 함수 내에서는 언패킹되어
적절한 위치의 가변인자에 할당되는 모습이다.
a, b, c = [1, 2, 3]
def show(e, rest):
print("요소: {0}, 나머지 {1}".format(e, rest))
first, *rest = [1, 2, 3, 4, 5]
show(first, rest) # 요소: 1, 나머지 [2, 3, 4, 5]
last, *rest = range(6)
show(last, rest) # 요소: 0, 나머지 [1, 2, 3, 4, 5]
first, *middle, last = range(6)
print(first) # 0
print(middle) # [1, 2, 3, 4]
print(last) # 5
first, last, *empty = 1, 2
print(first) # 1
print(last) # 2
print(empty) # []
from dataclasses import dataclass
USERS = [(i, f"first_name_{i}", f"last_name_{i}") for i in range(1_000)]
@dataclass
class User:
user_id: int
first_name: str
last_name: str
# 좋지 않은 예시
def bad_users_from_rows(dbrows) -> list:
"""DB레코드로부터 User를 생성"""
return [User(row[0], row[1], row[2]) for row in dbrows]
# 가독성 개선한 예시
def users_from_rows(dbrows) -> list:
"""DB레코드로부터 User만들기"""
return [
User(user_id, first_name, last_name)
for (user_id, first_name, last_name) in dbrows
]
여기서 모든 위치 인자를 넘기게 하여 리팩터링 하자.
# 위치인자 언패킹을 통한 User 객체 생성
def users_from_rows(dbrows) -> list:
"""DB레코드로부터 User만들기"""
return [User(*row) for row in dbrows]
아래와 같이 키워드 인자로 전달된 인자들이 사전(dict)으로
function 함수 내에서 패킹되고 있다.
def function(**kwargs):
print(kwargs)
function(a="1", b="2", c="3") # {'a': '1', 'b': '2', 'c': '3'}
함수 정의에 이중 별표(**
)를 사용한다는 것은 임의의 키워드 인자를 허용한다는 뜻이다.
이런 경우 파이썬은 우리가 임의로 접근할 수 있는 사전을 만들어준다.
앞의 예시에서 kwargs 인자가 바로 그 사전이다.
다만, 이 사전에서 직접 어떤 값을 추출하는 용도로 사용하지 말자.
def function(**kwargs):
timeout = kwargs.get("timeout", DEFAULT_TIMEOUT)
def function(timeout=DEFAULT_TIMEOUT, **kwargs):
...
즉, 사전에서 특정 키로 인자값을 조회하지 말고, 필요한 경우 함수의 정의에서 직접 꺼내도록 하자.
위 개선된 예에서 timeout은 엄격히 말하면 키워드 전용 인자는 아니다. 위치 인자일 뿐이다. 중요한 점은, 사전을 직접 조작하지 말자는 것이다.(함수의 서명에서 적절히 언패킹 하자)
인수의 값을 함수에 정의된 순서에 따라 차례로 인식된다.
def my_function(x, y):
print(f"{x=}, {y=}")
print(f"x={x}, y={y}")
my_function(1, 2) # x=1, y=2
my_function(x=1, y=2) # x=1, y=2
my_function(y=2, x=1) # x=1, y=2
my_function(1, y=2) # x=1, y=2
키워드로 인자가 이미 전달됐다면, 그 이후 인자도 모두 키워드로 전달돼야 한다.
def my_funcion(x, y, /):
print(f"{x=}, {y=}")
my_funcion(1, y=2)
TypeError: my_funcion() got some positional-only arguments passed as keyword arguments: 'y'
아무튼 특별한 경우가 아니면 키워드 인자 전달 방식으로 가독성을 높이자.
def my_function(x, y, *args, kw1, kw2=0):
print(f"{x=}, {y=}, {kw1=}, {kw2=}")
my_function(1, 2, kw1=3, kw2=5) # x=1, y=2, kw1=3, kw2=5
x=1, y=2, kw1=3, kw2=5 # x=1, y=2, kw1=3, kw2=0
해석 : 두 개의 위치 인자를 받고, 임의의 개수만큼 위치 인자를 추가로 받고, 마지막 2개의 키워드 전용 인자를 받는다.
마지막 kw2는 기본값을 가지고 있으므로 필수 입력 사항은 아니다.)
# 기존
my_function(1, 2)
# 개선 후보 1
my_function(1, 2, True)
# 개선 후보 2
# 이 방법이 더 좋다
my_function(1, 2, use_new_implementation=True)
# 기존 함수 정의
def my_function(x, y):
pass
# 개선 함수 정의
def my_function(x, y, use_new_implementation=True):
pass
너무 많은 인자의 사용은 code smell의 징후다.
f1
-> f2
f2 호출에 많은 정보가 필요하다는 말은, f1이 f2에 대해 많은 것을 알아야 한다는 것.
추상화가 필요한 상태다.
user.name, user.email, user.gender, user.created_at 등을 인자로 넘겨야 한다면, user를 넘기자.
단, 변경 가능한 객체의 부작용을 인지하자.
유사한 기능을 담당하는 응집력이 높은 객체에 모든 파라미터 넘기기.
유연하겠지만, 가독성 엉망이 됨
TIP
그대로 부모 클래스에 전달하는 래퍼 클래스나 데코레이터에만 사용하자.
def calculate_price(base_price: float, tax: float, discount: float) -> float:
return (base_price * (1 + tax)) * (1 - discount)
def show_price(price: float) -> str:
return "$ {0:,.2f}".format(price)
def str_final_price(
base_price: float, tax: float, discount: float, fmt_function=str) -> str:
return fmt_function(calculate_price(base_price, tax, discount))
print(str_final_price(10, 0.2, 0.5)) # 6.0
print(str_final_price(10, 0.2, 0.5, fmt_function=show_price)) # $ 6.00