Effective Python 번역 요약

Chapter 1. Pythonic Thinking

1. Know Which Version of Python You’re Using

python 3 써라

2. Follow the PEP 8 Style Guide

• tab 보단 4 spaces
• 한줄은 79 이하
• 여러줄에 걸쳐진 표현식의 둘째 줄 부터는 4 space 들여쓰기
• 함수, 클래스는 2줄 띄어쓰기
• 클래스 내 메소드는 한줄 띄어쓰기
• 배열 인덱스나, 함수 호출, 아규먼트 등에 space 두지 말기
• 변수 선언시 띄어쓰기는 하나만
• functions, variables, and attributes: lowercase_underscore
• protected instance attributes: _leading_underscore
• Private instance attributes: __double_leading_underscore
• Classes, exceptions: CapitalizedWord
• module-level constants: ALL_CAPS
• instance methods에는 self가 첫 파라메터
• class methods에는 cls가 첫 파라메터
• if not a is b < if a is not b
• if len(somelist) == 0, if somelist == [], if value == '' < if not something, if something
• 싱글라인 if, for, while, except 쓰지 말고 나눠서
• import는 파일의 제일 위에
• bar 패키지 안에서 foo를 임포트할 때, 절대경로 명시. import foo < from bar import foo
• 상대경로를 써야 할때에도 명시. import foo < from . import foo
• 스탠다드 라이브러리, 써드파티, 커스텀 순서대로 알파벳 순으로 임포트

3. Know the Differences Between bytes, str and unicode

Python3:

• bytes: 8-bit 값
• str: unicode 문자

Python2:

• str: 8-bit 값
• unicode: unicode 문자

utf-8은 unicode 문자를 바이너리 데이터로 표현하는 방식 중 가장 대표적인 애다. Python3 의 str, Python2의 unicode 인스턴스는 해당하는 바이너리 인코딩이 없다. 따라서 unicode 문자를 바이너리 데이터로 바꾸기 위해서는 encode 메소드를 써야 한다. 반대로 바이너리 데이터를 unicode 문자로 바꾸기 위해서는 decode 메소드를 써야 한다.

인코딩(디코딩)은 처음에 한번 제대로 해두고, 일관되게 unicode 타입(Python3: str, Python2: unicode)을 사용하라.

다음 함수들을 사용하라.

• Python3

def to_str(bytes_or_str):
	if isinstance(bytes_or_str, bytes):
		value = bytes_or_str.decode('utf-8')
	else:
		value = bytes_or_str
	return value  # 항상 str

def to_bytes(bytes_or_str):
	if isinstance(bytes_or_str, str):
		value = bytes_or_str.encode('utf-8')
	else:
		value = bytes_or_str
	return value  # 항상 bytes

• Python2

def to_unicode(unicode_or_str):
    if isinstance(unicode_or_str, str):
        value = unicode_or_str.decode('utf-8')
    else:
        value = unicode_or_str
    return value  # 항상 unicode

def to_str(unicode_or_str):
    if isinstance(unicode_or_str, unicode):
        value = unicode_or_str.encode('utf-8')
    else:
        value = unicode_or_str
    return value  # 항상 str

그런데 Python2에서는 unicode와 str의 비교, 병합 등의 연산이 가능하지만, Python3에서는 공백 문자열 마저도 완전히 다른걸로 취급한다.

파일을 열고 닫을때는 기본적으로 utf-8 인코딩을 사용하여 unicode문자를 다루므로, Python3에서 bytes를 다루고자 할때는 명시적으로 rb, wb를 파라메터를 사용해야한다.

4. Write Helper Functions Instead of Complex Expressions

my_values = {
	'red': ['5'],
	'green': [''],
	'blue': ['0']
}

# 1번 방법
red = int(my_values.get('red', [''])[0] or 0)

# 2번 방법
red = my_values.get('red', [''])
red = int(red[0]) if red[0] else 0

# helper 함수를 사용하는 방법
def get_first_int(values, key, default=0):
    found = values.get(key, [''])
    if found[0]:
        found = int(found[0])
    else:
        found = default
    return default
red = get_first_int(my_values, 'red')

• 1, 2번 읽기 어렵다. 반복해서 쓸 일이 있으면 3번 방법 써라.
• if/else가 or/and 등 보다 읽기 좋다.

5. Know How to Slice Sequences

• 0번부터 또는 마지막까지 자르는 경우, 생략하라. arr[0:len(arr)] < arr[:]

• 음수 인덱스도 적절히 활용하라. ex) arr[-20:]

• slice를 할당하면 새로운 인스턴스를 생성하고, slice에 할당하면 메모리 값을 바꾼다.

  a = [1, 2, 3, 4, 5, 6]
  
  b = a[-3:]
  b[1] = -1  # a[4]의 값을 바꾸지 않는다. 새로운 list가 b에 할당된다.
  print a  # [1, 2, 3, 4, 5, 6]
  print b  # [4, -1, 6]
  
  a[1:4] = [10, 20]  # a의 값을 바꾼다.
  print a  # [1, 10, 20, 5, 6]
  
  b = a[:]  # slice를 사용해 할당하면 새로운 list를 만든다.
  assert b == a and b is not a  # 내용은 같지만 다른 인스턴스이다.
  
  b = a  # 반면 변수 자체를 할당하면 메모리를 공유한다.
  a[:] = [10, 20]
  assert a is b # 같은 인스턴스이다.
  print b  # [10, 20] 값도 바뀐다.
  

6. Avoid Using start, end, and stride in a Single Slice

w = '가나다라'
x = w.encode('utf-8')  # unicode 문자를 8-bit 값으로 바꾸고
y = x[::-1]  # stride로 reverse 하면
z = y.decode('utf-8')  # UnicodeDecodeError: 'utf-8' codec can't not decode byte ...

• start, end, stride 한번에 다 쓰면 너무 복잡하다.
• stride는 따로 쓰고, 가능한 양수만 쓴다.
• 만약 정말 세 파라메터 모두 써야 한다면 slice와 stride를 나눠서 따로 하라. 아니면 itertools의 islice를 쓰라.

7. Use List Comprehensions Instead of map and filter

• list 표현은 map과 filter기능을 동시에 수행하면서 lambda가 필요없다.
• dictionary, set 에도 사용가능하다.

8. Avoid More Than Two Expressions in List Comprehensions

• 복잡한 중첩은 피하라.

  [x for sublist1 in mylist for sublist2 in sublist1 for x in sublist2]
  # 보다는,
  flat = []
  for sublist1 in mylist:
      for sublist2 in sublist1:
          flat.extend(sublist2)
  # 과 같이 하라.
  

9. Consider Generator Expressions for Large Comprehensions

• list comprehension은 메모리를 많이 사용한다. generator를 써라.

  value = [len(line) for line in open('./file.txt')]
  # 보다는,
  it = (len(line) for line in open('./file.txt'))
  # 를 써라.
  

• 중첩도 된다.

10. Prefer enumerate Over range

• range로 index를 순회하기보다는 enumerate를 써라.

• 둘째 파라메터를 주면 시작 지점을 설정할 수 있다.

  for i in range(3, len(l)):
      v = l[i]
      print('l[%d]: %s' % (i+1, v))
  # 보다는
  for i, v in enumerate(l, 3):
      print('l[%d]: %s' % (i+1, v))
  # 를 써라.
  

11. Use zip to Process Iterators in Parallel

• Python3의 zip은 generator지만, Python2는 아니다. itertools의 izip을 사용하라.

• 여러 반복을 한번에 처리할 수 있다.

  longest_name = None
  max_length = 0
  
  names = ['sungmin', 'james', 'peter']
  letters = [len(n) for n in names]
  for name, length in zip(names, letters):
      if length > max_length:
          longest_name = name
          max_length = length
  

• 길이가 다르면 짧은 쪽에 맞춘다.

  names.append('john')
  for name, length in zip(names, letters):
      print(name)
  # sungmin, james, peter
  

• Python3 에서는 itertools의 zip_longest를 쓰면 긴쪽에 맞출 수 있다. 짧은 쪽은 None으로 채워진다.

12. Avoid else Blocks After for and while Loops

• loop문 뒤에 else는 break되지 않고 끝났을 때 수행된다.

• 헷갈리니까 왠만하면 쓰지 마라.

  a = 4
  b = 9
  for i in range(2, min(a, b) + 1):
      if a % i == 0 and b % i == 0:
          print('Not coprime')
          break
  else:
      print('Coprime')
  # 보다는,
  def coprime(a, b):
      for i in range(2, min(a, b) + 1):
          if a % i == 0 and b % i == 0:
              return False
      return True
  # 또는,
  def coprime2(a, b):
      is_coprime = True
      for i in range(2, min(a, b) + 1):
          if a % i == 0 and b % i == 0:
              is_coprime = False
              break;
      return is_coprime
  # 처럼 하라.
  

13. Take Advantage of Each Block in try/except/else/finally

• 각각을 적절히 활용하는 예제 코드

UNDEFINED = object()

def divide_json(path):
    handle = open(path, 'r+')   # IOError가 날 수 있다.
    try:
        data = handle.read()    # UnicodeDecodeError가 날 수 있다.
        op = json.loads(data)   # ValueError가 날 수 있다.
        value = (
            op['numerator'] /
            op['denominator'])  # ZeroDivisionError가 날 수 있다.
    except ZeroDivisionError as e:
        return UNDEFINED
    else:
        op['result'] = value
        result = json.dumps(op)
        handle.seek(0)
        handle.write(result)    # IOError가 날 수 있다.
        return value
    finally:
        handle.close()          # 항상 수행된다.

• try는 예외 잡을 때, except는 예외처리, else는 성공 시, finally는 마지막에 수행된다.

Chapter 2. Functions

14. Prefer Exceptions to Returning None

• None을 리턴하고 if not result과 같은 식으로 확인하면, 리턴값이 0, [], '' 등일 때 잘못될 수 있다.

  def divide(a, b):
      try:
          return a / b
      except ZeroDivisionError:
          return None
  
  x, y = 0, 5
  result = devide(x, y)
  if not result:               # result = 0
      print('Invalid inputs')  # 잘못되었다.
  
  # 성공 실패를 같이 반환할 수도 있지만,
  def divide2(a, b):
      try:
          return True, a / b
      except ZeroDivisionError:
          return False, None
  # 보기도 불편하고
  _, result = divide(x, y)
  # 이렇게 무시할 가능성이 높다.
  
  # 이럴 땐, 인풋이 적절치 않다는 ValueError를 내라.
  def divide3(a, b):
      try:
          return a / b
      except ZeroDivisionError as e:
          raise ValueError('Invalid inputs') from e
  # 그러면,
  try:
      result = divide(x, y)
  except ValueError:
      print('Invalid inputs')
  else:
      print('Result is %.1f' % result)
  # 이렇게 경우에 따라 처리할 수 있다.
  

15. Know How Closures Interact with Variable Scope

다음 함수가 잘 동작할까?

def sort_priority(numbers, group):
    '''group에 속한 숫자들끼리 비교해서 앞에 두고, 나머지를 비교한다.'''
    found = False
    def helper(x):
        if x in group:
            found = True
            return (0, x)
        return (1, x)
    numbers.sort(key=helper)
    return found

numbers = [8, 3, 1, 2, 5, 4, 7, 8]
group = {2, 3, 5, 7}
found = sort_priority(numbers, group)
print('Found: ', found)  # Found: False <- 잘못되었다.
print(numbers)           # [2, 3, 5, 7, 1, 4, 6, 8]

• 함수는 바깥 스코프에서 선언된 변수를 쓸 수 있다.
• 함수는 first_class 오브젝트이다. 변수에 할당가능하고, 비교할 수 있으며, 파라메터로 넘길 수 있다.
• tuple을 대소 비교할 때에는 앞에 것부터 순서대로 비교한다.
• 변수를 사용하고자 할 때, 그 값을 찾는 스코프 순서는 다음과 같다. 모두 없으면 NameError가 난다.
  • 현재 함수 스코프
  • 그 바깥 함수 스코프
  • 모듈 스코프 (= 글로벌 스코프)
  • built-in 스코프 (len, str 등의 함수를 포함)
• 변수에 값을 할당할 때, 현재 스코프에 있으면 값을 바꾸고, 없으면 바깥 스코프를 사용하지 않고 새로 메모리에 올린다.

어떻게 함수 내에서 바깥 스코프의 변수를 바꿀 수 있을까.

# Python3
def sort_priority2(numbers, group):
    '''group에 속한 숫자들끼리 비교해서 앞에 두고, 나머지를 비교한다.'''
    found = False
    def helper(x):
        nonlocal found
        if x in group:
            found = True
            return (0, x)
        return (1, x)
    numbers.sort(key=helper)
    return found

# Python2
def sort_priority3(numbers, group):
    '''group에 속한 숫자들끼리 비교해서 앞에 두고, 나머지를 비교한다.'''
    found = [False]
    def helper(x):
        if x in group:
            found[0] = True
            return (0, x)
        return (1, x)
    numbers.sort(key=helper)
    return found[0]

• global <변수명> 을 명시하면 모듈 스코프에서 변수를 찾는다.

• Python3에는 nonlocal <변수명>이 있어, 바깥 함수 스코프로 제한할 수도 있다.

• Python2에는 애초에 변수를 mutable한 list 등으로 선언하고 내부에서는 found[0] = True와 같이 바꾼다.

• 하지만 함수가 길어지면 헷갈릴 수 있기 때문에, 복잡한 함수는 다음과 같이 별도의 helper 클래스로 만드는 것이 읽기 좋다.

  class Sorter(object):
      def __init__(self, group):
          self.group = group
          self.found = False
  
      def __call__(self, x):
          if x in self.group:
              self.cound = True
              return (0, x)
          return (1, x)
  
  sorter = Sorter(group)
  numbers.sort(key=sorter)
  assert sorter.found is True
  

16. Consider Generators Instead of Returning Lists

• ret = []를 선언 한 후 ret.append(x)를 반복하고 나중에 return ret하는 것보다, 바로 yield하는 generator로 구성하는게 읽기에도 좋고, 메모리관리에도 좋다.
• 사용자는 반드시 generator의 특성들(인덱스 참조 가 안되고 재사용 안되는 등)을 인지해야한다.

17. Be Defensive When Iterating Over Arguments

함수 내에서 인풋을 여러번 순회할 때는 타입을 신경써야한다.

다음 함수는 어떤 종류의 파라메터만 받을 수 있는가?

def normalize(numbers):
    '''전체 리스트에서 각 값의 비율을 계산한다'''
    total = sum(numbers)
    result = []
    for value in numbers:
        percent = 100 * value / total
        result.append(percent)
    return result

만약 list가 아니라 iterator가 들어가게 되면, sum(numbers)에서 다 사용되고, for문은 더 이상 돌지 않는다.

함수 앞에 numbers = list(numbers)를 추가하여, iterator를 list로 만들어 복사하거나, 아에 generator를 인자로 넘긴다.

def normalize_copy(numbers):
    '''전체 리스트에서 각 값의 비율을 계산한다'''
    numbers = list(numbers)  # iterator의 값 복사
    total = sum(numbers)
    result = []
    for value in numbers:
        percent = 100 * value / total
        result.append(percent)
    return result

def normalize_func(get_iter):
    '''전체 리스트에서 각 값의 비율을 계산한다'''
    total = sum(get_iter())
    result = []
    for value in get_iter():
        percent = 100 * value / total
        result.append(percent)
    return result

percentages = normalize_func(lambda: read_file(path))

• for는 사실 iter(foo)이다. iter함수는 foo.__iter__를 호출하는데, 이는 __next__메소드를 가진 iterator 오브젝트를 반환한다. for루프는 StopIteration이 날 때까지 next(<iterator>)를 수행한다.

• __iter__메소드를 generator로 오버라이드한 클래스를 만들어, 매번 iterator를 생성할 수 있다.

  class ReadFile(object):
      def __init__(self, path):
          self.path = path
  
      def __iter__(self):
          with open(self.path) as f:
              for line in f:
                  yield int(line)
  
  numbers = ReadFile(path)
  percentages = normalize(numbers)
  

• iter 함수는 인자가 interator일 경우에는 그 자체를, container인 경우에는 iterator를 만들어 반환한다. 이를 이용해 인풋의 타입을 체크할 수 있다.

  def normalize_defensive(numbers):
      if iter(numbers) is iter(numbers)  # numbers가 iterator가 아니라면 False
      	raise TypeError('Must supply a container')
      total = sum(numbers)
      result = []
      for value in numbers:
          percent = 100 * value / total
          result.append(percent)
      return result
  
  numbers = [15, 35, 80]
  normalize_defensive(numbers)  # No Error
  numbers = ReadFile(path)
  normalize_defensive(numbers)  # No Error
  

18. Reduce Visual Noise with Variable Positional Arguments

• *args를 이용해서 0 - n개의 인자를 받도록 할 수 있다.
• *args는 인풋들을 tuple로 묶는다. 따라서 generator를 넘기게 되면, 이를 모두 메모리에 올리게 된다. 그러므로 들어오는 인풋의 개수가 적을때에만 사용하도록 한다.
• *args 뒤에 다른 인풋을 받도록 선언할 수 없다.
• 파라메터가 꼬이는 경우가 쉽게 생기기 떄문에, keyword-only argument를 사용하는 것도 좋다.

19. Provide Optional Behavior with Keyword Arguments

def remainder(number, divisor):
    return number % divisor

remainder(20, 7)
remainder(20, divisor=7)
remainder(number=20, divisor=7)
remainder(divisor=7, number=20)

• 파라메터의 이름과 함께 주어지는 인자를 keyword argument라 하고, 위치에 맞게 주어지는 인자를 positional argument라 한다.
• positional argument는 keyword argument 앞에 와야한다.
• keyword argument를 사용하면 가독성이 좋다.

def flow_rate(weight_diff, time_diff, period=1, units_per_kg=1):
    return ((weight_diff / units_per_kg) / time_diff) * period

flow_rate(weight_diff, time_diff, 3600, 2.2)  # 헷갈린다.
flow_rate(weight_diff, time_diff, period=3600, units_per_kg=2.2)  # 이렇게 쓰라.

• 함수를 선언할 때 keyword 파라메터를 여러개 둘 수 있다. positional argument로도 동작하지만 keyword를 명시하라.

20. Use None and Docstrings to Specify Dynamic Default Arguments

• keyword argument의 기본값으로 주어진 함수는 모듈이 로드될 때 한번만 수행되므로 동적인 값을 주지 않는다. 동적인 기본값이 필요할 때에는 None을 기본값으로 주고 그 동작을 docstring으로 설명하라.

  def log(message, when=datetime.now()):
      print('%s: %s' % (when, message))
  
  # 이 함수는 datetime.now()가 한번 실행되어 바인딩 되므로, 동적인 값을 줄 수 없다.
  log('Hi there!')  # 2014-11-15 21:10:10.371432: Hi there!
  sleep(0.1)
  log('Hi again!')  # 2014-11-15 21:10:10.371432: Hi again!
  
  def log_dynamic(message, when=None):
      ''' Log a message with a timestamp.
  
      Args:
      	message: Message to print.
      	when: datetime of when the message occured.
      		Defaults to the present time.
      '''
      when = datetime.now() if when is None else when
      print('%s: %s' % (when, message))
  
  # 이 함수는 제대로 동작한다.
  

• 함수 뿐 아니라 call by reference인 오브젝트도 문제다. 기본값으로 주어진 오브젝트는 모듈 로드 때 한번만 메모리에 올라간다.

  def decode(data, default={}):
      try:
          return json.loads(data)
      except ValueError:
          return default
  
  foo = decode('bad data')
  foo['stuff'] = 5
  bar = decode('also bad')
  bar['mepp'] = 1
  # foo와 bar는 같은 dictionary 오브젝트를 공유한다.
  print('foo: ', foo)  # foo: {'stuff': 5, 'meep': 1}
  print('bar: ', bar)  # bar: {'stuff': 5, 'meep': 1}
  

21. Enforce Clarify with Keyword-Only Arguments

• keyword argument를 써서 함수를 정의했다 하더라도 positional argument처럼 호출할 수 있기때문에 헷갈릴 수 있다.

• Python3 에서는 *를 positional argument와 keyword argument사이에 두어 keyword argument를 강제한다.

  def safe_division(number, divisor, *,
                    ignore_overflow=False,
                    ignore_zero_division=False):
      try:
          return number / divisor
      except OverflowError:
          if ignore_overflow:
              return 0
          else:
              raise
      except ZeroDivisionError:
          if ignore_zero_division:
              return float('int')
          else:
              raise
  
  safe_division(1, 10*500, True, False)
  # TypeError: safe_division() takes 2 positional arguments
  

• Python2에서는 그런 기능이 없기 때문에 ** arguments를 활용하고, 잘못된 인풋에는 TypeError를 낸다.

  def safe_division(number, divisor, **kwargs):
      ignore_overflow = kwargs.pop('ignore_overflow', False)
      ignore_zero_division = kwargs.pop('ignore_zero_division', False)
      if kwargs:  # pop으로 값을 뽑아내고, 비어있는지 체크해서 파라메터가 잘못되었는지 확인한다.
          raise TypeError('Unexpected **kwargs: %r' % kwargs)
      # ...
  

Chapter 3. Classes and Inheritance

22. Prefer Helper Classes Over Bookkeeping with Dictionary and Tuples

• dictionary의 값으로 dictionary나 긴 tuple이 들어가는 것을 지양하라.
• tuple은 요구사항에 따라 길어질 수 있다. 가벼운 immutable한 데이터 컨테이너가 필요할 때, collections의 namedtuple을 사용하면 이름으로 관리할 수 있고, 나중에 class로의 확장도 쉽다.
  • 하지만 기본값을 설정할 수 없다. class를 이용하는게 더 나을 수 있다.
  • 여전히 인덱스와 순회로 값에 접근할 수 있어, 의도치 않은 오작동을 일으킬 수 있다.

class WeightedGradeBook(object):
    def __init__(self):
        self._grades = {}

    def add_student(self, name):
        self._grades[name] = {}

    def report_grade(self, name, subject, score, weight):
        by_subject = self._grades[name]
        grade_list = by_subject.setdefault(subject, [])
        grade_list.append((score, weight))

    def average_grade(self, name):
        by_subject = self._grades[name]
        score_sum, score_count = 0, 0
        for subject, scores in by_subject.items():
            subject_avg, total_weight = 0, 0
            for score, weight in scores:
                subject_avg += score * weight
                total_weight += weight
            score_sum += subject_avg / total_weight
            score_count += 1
        return score_sum / score_count

# 만약 여기에서 note라는 변수가 추가되어 (score, weight, note)가 되면 수정하기 복잡하다.

from collections import namedtuple
Grade = namedtuple('Grade', ('score', 'weight'))

class Subject(object):
    def __init__(self):
        self._grades = []

    def report_grade(self, score, weight):
        self._grades.append(Grade(score, weight))

    def average_grade(self):
        total, total_weight = 0, 0
        for grade in self._grades:
            total += grade.score * grade.weight
            total_weight += grade.weight
        return total / total_weight


class Student(object):
    def __init__(self):
        self._subjects = {}

    def subject(self, name):
        if name not in self._subjects:
            self._subjects[name] = Subject()
        return self._subjects[name]

    def average_grade(self):
        total, count = 0, 0
        for subject in self._subjects.values():
            total += subject.average_grade()
            count += 1
        return total / count


class GradeBook(object):
    def __init__(self):
        self._students = {}

    def student(self, name):
        if name not in self._students:
            self._students[name] = Student()
        return self._students[anme]


# 이처럼 길더라도 여러개의 helper class를 이용하는 것이 더 깔끔하다.

book = GradeBook()
albert = book.student('Albert Einstein')
math = albert.subject('student')
math.report_grade(80, 0.10)
print(albert.average_grade)  # 123

23. Accept Functions for Simple Interfaces Instead of Classes

• defaultdict를 쓰는데, 없는 키의 카운트를 세고 싶다. stateful hook을 위해서는 helper 함수로 closure를 정의하라.

  def increment_with_report(current, increments):
      added_count = 0  # 없는 키를 세기 위한 state
    
      def missing():   # defaultdict의 hook. 키가 없을때 호출되며, state를 바꾼다.
          nonlocal added_count
          added_count += 1
          return 0
    
      result = defaultdict(missing, current)
      for key, amount in increments:
          result[key] += amount
    
      return result, added_count
  

  result, count = increment_with_report(current, increments

• 하지만 이런 예는 stateless 함수보다 읽기 어렵다. state를 감싼 작은 class는, stateful closure의 동작을 더 깔끔하게 보여준다. __call__ 메소드를 구현하여 인스턴스 자체를 hook으로 쓴다.

  class CountMissing(object):
      def __int__(self):
  		    self.added = 0
    
      def __call__(self):
          self.added += 1
          return 0
    
  counter = CountMissing()
  result = defaultdict(counter, current)
    
  for key, amount in increments:
      result[key] += amount
  

• 간단한 interface와 같은 함수도 있다. (eg. hook을 받는 defaultdict)
• Python의 함수나 메소드는 first class이다. 즉 다른 표현에 사용되거나 변수로 쓰일 수 있다.
• __call__ 메소드는 인스턴스를 함수처럼 쓸 수 있게 한다. 상태를 유지하는 함수가 필요할 때에는 stateful closure를 정의하기보단 __call__을 구현한 클래스를 만들자.

24. Use @classmethod Polymorphism to Construct Objects Generically

Python에서는 오브젝트 뿐 아니라 클래스도 다형성을 지원한다. 다형성은 계층관계의 클래스들이 각자 별도의 메소드를 가질 수 있는 것을 뜻한다. 이로 인해, 공통된 interface나 abstract base class를 가지고 여러 기능을 수행하는 클래스를 만들 수 있다.

여러 데이터 소스의 라인 수의 총 합을 세어주는 MapReduce를 구현한다고 해보자. 인터페이스가 되는 InputData클래스가 필요하고, 데이터 소스의 종류별로 read 메소드를 오버라이드한 subclass들을 구현한다. 마찬가지로 map과 reduce함수를 가진 Worker추상 클래스가 필요하고, 이를 상속하여 필요한 작업에 맞게 오버라이드한 subclass들을 구현한다.

class InputData(object):
    '''input data는 이 class를 상속하여 read 메소드를 구현한다.'''
    def read(self):
        raise NotImplementedError


class PathInputData(InputData):
    '''file을 열어 읽는 subclass이다.'''
    def __init__(self, path):
        super().__init__()  # Python2에서는 super(PathInputData, self).__init__()
        self.path = path

    def read(self):
        return open(self.path).read()

# 이와 비슷하게 network, decompress 등으로부터 데이터를 읽는 subclass를 만들 수 있다.


class Worker(object):
    '''MapReduce 작업을 수행하는 워커 클래스이다. 작업에 따라 subclass를 구현한다.'''
    def __init__(self, input_data):
        self.input_data = input_data
        self.result = None

    def map(self):
        raise NotImplementedError

    def reduce(self, other):
        raise NotImplementedError


class LineCounterWorker(Worker):
    '''데이터 소스의 라인 수를 세어주는 워커 클래스이다.'''
    def map(self):
        data = self.input_data.read()
        self.result = data.count('\n')

    def reduce(self, other):
        self.result += other.result

# 다른 종류의 작업을 하는 subclass를 만들 수 있다.

여러개의 Worker가 여러개의 InputData를 멀티쓰레드로 처리하게 한 후, Worker.reduce로 결과를 종합하고 싶다. 이를 관리하기 위해 각 오브젝트를 연결해 주는 다음과 같은 helper 함수가 필요하다.

def generate_inputs(data_dir):
    '''data_dir하위의 파일들에 대해 각각 PathInputData 오브젝트를 만든다.'''
    for name in os.listdir(data_dir):
        yield PathInputData(os.path.join(data_dir, name))


def create_workers(input_list):
    '''각각의 InputData 오브젝트에 대해, LineCountWorker를 만든다.'''
    workers = []
    for input_data in input_list:
        workers.append(LineCountWorker(input_data))
    return workers


def execute(workers):
    '''멀티쓰레드를 이용하여 각 Worker들의 map 메소드를 수행하고,
    첫번째 Worker를 기준으로 reduce하여 최종 값을 구한다.'''
    threads = [Thread(target=w.map) for w in workers]
    for thread in threads: thread.start()
    for thread in threads: thread.join()
    first, rest = workers[0], workers[1:]
    for worker in rest:
        first.reduce(worker)
    return first.result


def mapreduce(data_dir):
    inputs = generate_inputs(data_dir)
    workers = create_workers(inputs)
    return execute(workers)

하지만 문제가 있다. 이와 같은 generate_inputs, create_workers, mapreduce 함수는 내부에서 PathInputData, LineCountWorker를 사용하기 때문에 generic하지 못하다. 때문에, 다른 InputData, Worker를 위해서는 사용될 수 없다.

다른 언어에서는 생성자의 다형성을 이용해 generic한 오브젝트 생성을 하지만, Python에는 __init__이라는 하나의 생성자만 지원한다.

이러한 문제는 @classmethod다형성을 이용해 해결할 수 있다. 이 데코레이션이 달린 메소드는 생성된 오브젝트만이 아니라, 전체 클래스에 영향을 미친다.

class GenericInputData(object):
    def read(self):
        raise NotImplementedError

    @classmethod
    def generate_inputs(cls, config):
        raise NotImplementedError


class PathInputData(GenericInputData):
	# ...
    @classmethod
    def generate_inputs(cls, config):
        data_dir = config['data_dir']
        for name in os.list(data_dir):
            yield cls(os.path.join(data_dir, name))


class GenericWorker(object):
    # ...
    def map(self):
        raise NotImplementedError

    def reduce(self):
        raise NotImplementedError

    @classmethod
    def create_workers(cls, input_class, config):
        workers = []
        for input_data in input_class.generate_inputs(config):
            workers.append(cls(input_data))  # __init__을 바로 사용하지 않고 cls를 이용한다.
        return workers


class LineCountWorker(GenericWorker):
    # ...

create_workers메소드는 input_class라는 인자를 받고, input_class.generate_inputs메소드를 호출하여 GenericInputData의 subclass를 생성한다. 즉 PathInputData가 인자로 들어오든 다른 subclass가 인자로 들어오든 상관없이 적절한 인스턴스를 생성하고, 이를 이용해 Worker인스턴스를 생성한다.

이제 mapreduce 함수는 다음과 같이 generic하게 된다.

def mapreduce(worker_class, input_class, config):
    workers = worker_class.create_workers(input_class, config)
    return execute(workers)


# 다음과 같이 사용할 수 있다.
from tempfile import TemporaryDirectory

def write_test_files(tmpdir):
    # ...

with TemporaryDirectory() as tmpdir:
    write_test_files(tmpdir)
    config = {'data_dir': tmpdir}
    result = mapreduce(LineCountWorer, PathInputData, config)

• Python 은 __init__이라는 하나의 생성자만 지원한다.
• 하지만 @classmethod를 이용하면 대용할 수 있는 생성자를 만들 수 있다.
• subclass를 생성하고 관리하는 generic한 방법이 필요할 땐, class method 다형성을 이용하라.

25. Initialize Parent Classes with super

다이아몬트 형태의 상속관계를 생각해보자. 슈퍼클래스가 있고, 이를 상속한 두개의 서브클래스가 있고, 이 두 서브클래스를 모두 상속하는 최종 클래스가 있다고 하자.

생성자를 다음과 같이 작성하면 어떻게 될까.

class First(object):
    def __init__(self):
        print 1


class Second(First):
    def __init__(self):
        First.__init__(self)
        print 2


class Third(First):
    def __init__(self):
        First.__init__(self)
        print 3


class Fourth(Second, Third):
    def __init__(self):
        Second.__init__(self)
        Third.__init__(self)


Fourth()  # 1, 2, 1, 3 순서대로 출력된다.

즉, 최종 클래스의 생성자에서 호출하는 순서대로 부모 클래스의 생성자가 호출되는데, 부모 클래스들이 또 다시 부모 클래스를 공유할 경우, 최상위 클래스의 생성자는 여러번 불리게 된다.

최상위 클래스는 값을 할당하고 그 하위 클래스들은 값을 바꾸는 동작을 한다면, 이는 잘못된 결과를 낳을 수 있다. 값 할당 → 변경 → 변경 순서로 동작해야 하는데, 값할당 → 변경 → 값할당 → 변경 순서로 동작하기 때문이다.

이런 문제를 방지하기 위해 MRO(method resolution Order)를 정의하는 super라는 메소드가 있다. 이는 어떤 순서로 초기화 함수를 수행할지를 표준화하고. 각 생성자는 중복 없이 한번만 실행된다.

# Python2
class Second(First):
    def __init__(self):
        super(Second, self).__init__()
        print 2


class Third(First):
    def __init__(self):
        super(Third, self).__init__()
        print 3


class Fourth(Second, Third):
    def __init__(self):
        super(Fourth, self).__init__()
        print 4


from pprint import pprint

pprint(Fourth.mro())
'''
[<class '__main__.Fourth'>,
 <class '__main__.Second'>,
 <class '__main__.Third'>,
 <class '__main__.First'>,
 <type 'object'>]
'''

Fourth()  # 1, 3, 2, 4 순서대로 출력된다.

'''
Python3
__class__라는 변수가 있어서 supser(__class__, self)로 쓸수 있다.
더 간단하게는 super()라고만 해도된다.
클래스명을 쓰지 않기 때문에, 클래스명 변경에 더 자유롭다.
'''

• 상속관계가 있을 때엔 super(<class name>, self).__init__()으로 부모의 생성자를 호출한다.
• 부모 클래스가 여러개인 경우, 나중에 주어진 클래스부터, 가장 최상위 클래스부터 초기화된다. 즉 먼저 주어졌을수록, 최하위 일수록 우선순위를 가진다. (값을 오버라이드함)

26. Use Multiple Inheritance Only for Mix-in Utility Classes

• 다중 상속이 가능하지만, 왠만하면 하나만 상속하라. 만약 다중 상속을 해야 한다면, mix-in 클래스를 써라.
• mix-in 클래스는 추가적인 메소드를 정의한다.
• 하지만 추가적인 인스턴스 어트리뷰트를 정의하지 않는다.
• 뿐만 아니라 __init__생성자가 호출될 일이 없다.

클래스를 serializable한 dictionary로 만드는 함수를 generic하게 작석하고 싶다면 아래와 같이 할 수 있다.

class ToDictMixin(object):
    def to_dict(self):
        return self._traverse(None, self.__dict__)

    def _traverse(self, key, value):  # key 파라메터는 필요없어 보이겠지만 ...
        if isinstance(value, ToDictMixin):
            return value.to_dict()
        elif isinstance(value, dict):
            return {k: self._traverse(k, v) for k, v in value.items()}
        elif isinstance(value, list):
            return [self._traverse(key, v) for v in value]
        elif hasattr(value, '__dict__'):
            return self._traverse(key, value.__dict__)
        else:
            return value

이 클래스를 상속한 클래스는 to_dict메소드를 호출 할 수 있다. 이 메소드는 재귀적으로 동작한다. 따라서 싸이클이 있는 경우에는 무한루프에 빠질 수 있다. 따라서 다음과 같이 무한루프에 빠지는 케이스를 따로 처리하도록 오버라이드한다.

class BinaryTree(ToDictMixin):
    def __init__(self, value, left=None, right=None):
        self.value = value
        self.left = left
        self.right = right


class BinaryTreeWithParent(BinaryTree):
    def __init__(self, value, left=None, right=None, parent=None):
        super().__init__(value, left, right)
        self.parent = parent  # 이 어트리뷰트 때문에 cycle이 생기고, to_dict는 무한루프에 빠진다.

    def _traverse(self, key, value):  # 이렇게 오버라이드 해서 무한루프를 방지한다.
        if isinstance(value, BinaryTreeWithParent) and key == 'parent':  # 키는 여기서 쓴다.
            return value.value  # parent는 _traverse로 순회하지 않고, 값만 반환한다.
        else:
            return super()._traverse(key, value)

root = BinaryTreeWithParent(10)
root.left = BinaryTreeWithParent(7, parent=root)
root.left.right = BinaryTreeWithParent(9, parent=root.left)
print(root.to_dict())

다른 mix-in클래스도 필요할 수 있다. 오브젝트를 json으로 만들고, json으로 오브젝트를 만드는 mix-in클래스를 구현해보자.

class JsonMixin(object):
    @classmethod
    def from_json(cls, data):
        kwargs = json.loads(data)
        return cls(**kwargs)

    def to_json(self):
        return json.dumps(self.to_dict())

이 클래스는 ToDictMixin.to_dict 메소드를 사용한다. 따라서 JsonMixin클래스는 ToDictMixin클래스를 상속한 클래스에서만 제대로 동작할 것이다.

class DatacenterRack(ToDictMixin, JsonMixin):
    def __init__(self, switch=None, machines=None):
        self.switch = Switch(**switch)
        self.machines = [Machine(**kwargs) for kwargs in machines]


class Switch(ToDictMixin, JsonMixin):
    def __init__(self, ports, speed):
        self.ports = ports
        self.speed = speed


class Machine(ToDictMixin, JsonMixin):
    def __init__(self, cores, ram, disk):
        self.cores = cores
        self.ram = ram
        self.disk = disk


serialized = '''{
    "switch": {"ports": 5, "speed": 1e9},
    "machines": [
        {"cores": 8, "ram": 32e9, "disk": 5e12},
        {"cores": 4, "ram": 16e9, "disk": 1e12},
        {"cores": 2, "ram": 4e9, "disk": 500e12}
    ]
}'''

deserialized = DatacenterRack.from_json(serialized)  # json 스트링을 오브젝트로 만든다.
roundtrip = deserialized.to_json()  # 오브젝트를 json 스트링으로 만든다.
assert json.loads(serialized) == json.loads(roundtrip)

• mix-in 클래스의 메소드를 오버라이드할 수 있다.
• 여러 mix-in 클래스들을 함께 사용할 수 있다.

27. Prefer Public Attributes Over Private Ones

• __private_field는 내부적으로 _MyClass__private_field로 저장된다. 따라서 self.__private_field는 내부적으로 self._MyClass__private_field로 동작한다.

  print(child_object.__dict__)  # {'_MyParentClass__private_field': 27}
  

• subclass는 클래스명이 다르므로 parent의 private 어트리뷰트에 접근하지 못한다. 대신 child_object._MyParentClass__private_field 처럼 하면 접근이 가능하다.

• “We are all consenting adults here.”. Python의 모토다. private이더라도 완전히 가려지지 않는다.

• 때문에 다른 개발자가 private을 쓸 수 있으며, 상속구조가 바뀌거나 하면 문제가 될 수 있다.

• 차라리 다큐먼트로 설명을 하고, _protedted_field를 쓰는게 낫다.

• 하지만 _protedted_field는 subclass에서 같은 이름을 쓰면 덮어쓰게 되므로 이에 유의해야한다.

28. Inherit from collections.abc for Custom Container Types

기존 자료구조를 이용하여 새로운 커스텀 클래스를 만들 수 있다.

class FrequencyList(list):
    def __init__(self, numbers):
        super().__init__(numbers)

    def frequency(self):
        counts = {}
        for item in self:
            counts.setdefault(item, 0)
            counts[item] += 1
        return counts


foo = FrequencyList(['a', 'b', 'a', 'c', 'b', 'a', 'd'])
print(len(foo))         # 7
foo.pop()
print(repr(foo))        # ['a', 'b', 'a', 'c', 'b', 'a']
print(foo.frequency())  # {'a': 3, 'c': 1, 'b': 2}

좀 더 복잡하게, 커스텀 클래스를 기존 자료구조처럼 이용하고 싶은 경우를 생각해보자.

class BinaryNode(object):
    def __init__(self, value, left=None, right=None):
        self.value = value
        self.left = left
        self.right = right


class IndexableNode(BinaryNode):
    def _search(self, count, index):
        found = None
        if self.left:
            found, count = self.left._search(count, index)
        if not found:
            if count == index:
                found = self
            else:
                count += 1
                if self.right:
                    found, count = self.right._search(count, index)
                else:
                    found = None
        return found, count

    def __getitem__(self, index):
        found, _ = self._search(0, index)
        if not found:
            raise IndexError('Index out of range')
        return found.value


tree = IndexableNode(
    10,
    left=IndexableNode(
        5,
        left=IndexableNode(2),
        right=IndexableNode(
            6,
            right=IndexableNode(7))),
    right=IndexableNode(
        15,
        left=IndexableNode(11)))

print(list(tree))  # [2, 5, 6, 7, 10, 11, 15]
print(tree[2])     # 6
print(11 in tree)  # True
print(4 in tree)   # False

• __getitem__을 구현하면, [index]로 접근할 수 있으며, list함수나 in 오퍼레이터도 쓸 수 있다.

하지만 진짜 list 클래스처럼 쓰려면 __len__도 구현해야하고, count나 index 메소드까지 쓰려면 일이 많다. 기본 컨테이너를 직접 구현하는건 쉬운 일이 아니다. 많은 메소드들을 다 구현해야하기 때문이다.

class SequenceNode(IndexableNode):
    def __len__(self):  # 우선 len을 구현한다.
        _, count = self._search(0, None)
        return count

• collections.abc모듈은 각 컨테이너 타입에 맞는 메소드들을 미리 구현해 놓은 추상 클래스를 제공한다.

from collections.abc import Sequence

class BadType(Sequence):
    pass

bad = BadType()
# Type Error: Can't instantiate abstract class BadType with abstract methods __getitem__, __len__

class BetterNode(SequenceNode, Sequence):  # __getitem__, __len__만 구현해서 넣으면,
    pass                                   # index, count는 자동으로 구현된다.

• collections.abc의 Sequence는 __getitem__, __len__만 구현하면 index, count 등의 디펜던시가 있는 다른 메소드는 자동으로 사용할 수 있다.
• 특별한 메소드를 더 많이 가진 타입인 Set이나 MutableMapping을 만들 때는 특히 더 유용하다.

Chapter 4. Metaclasses and Attributes

29. Use Plain Attributes Instead of Get and Set Methods

• getter와 setter를 두는것은 Pythoinc하지 않다.

• getter와 setter는 interface를 제공하고, 기능을 encapsulate하고, 사용의 한계를 정의하는 면에서 중요하지만, 파이썬에서는 쓰지 않고 그냥 private, public 어트리뷰트로 구현한다.

• 어트리뷰트의 값이 바뀌었을 때, 어떤 추가적인 동작을 하기 원한다면 @property와 @<attribute>.setter를 쓴다.

  class Resistor(object):
      def __init__(self, ohms):
          self.ohms = ohms
          print('init voltage, ohms')
          self.voltage = 0  # 하위 클래스의 voltage.setter가 불린다.
          self.current = 0
  
  class VoltageResistance(Resistor):
      def __init__(self, ohms):
          super().__init__(ohms)
          print('init _voltage')
          self._voltage = 0  # super().__init()에서 이미 세팅되었기 때문에, 없어도 된다.
  
      @property
      def voltage(self):
          print('get voltage')
          return self._voltage
  
      @voltage.setter
      def voltage(self, voltage):
          print('set voltage')
          self._voltage = voltage
          # self.voltage, self._voltage 모두 없다가 이때 _voltage 어트리뷰트가 생긴다.
          self.current = voltage / self.ohms
  
  r = VoltageResistance(1e3)
  # init voltage, ohms
  # set voltage
  # init _voltage
  
  print(r.current)  # 0
  r.voltage = 10
  print(r.current)  # 0.01
  

• 새로 할당하려는 값의 유효성을 검사하거나, immutable하도록 제한하는데 쓸 수 도 있다.

  class BoundedResistance(Resistor):
      def __init__(self, ohms):
          super().__init__(ohms)
  
      @property
      def ohms(self):
          return self._ohms
  
      @ohms.setter
      def ohms(self, ohms):
          if hasattr(self, '_ohms'):
              raise AttributeError("Can't set attribute")
          if ohms <= 0:
              raise ValueError('%f ohms must be > 0' % ohms)
          self._ohms = ohms
  
  r = BoundedResistance(1e3)
  r.ohms = 0  # ValueError: 0.000000 ohmes must be > 0
  BoundedResistance(-5)  # ValueError: -5.000000 ohmes must be > 0
  
  print(r.__dict__)  # {'_ohms': 1000.0, 'voltage': 0, 'current': 0}
  

• BoundedResistance.__init__(-5)는 Resistor.__init__(-5)를 호출하고, 이는 self.ohms = -5이기 때문에 결국 BoundedResistance의 ohms.setter가 수행된다. ohms라는 변수는 생기지 않고 실제로는 _ohms만 있다.

• @property는 subclass로는 공유되지만, 다른 클래스는 구현을 공유할 수 없다. descriptor를 쓰면 이를 해결할 수 있다.

• @property로 만든 getter에서 다른 어트리뷰트을 바꾸지 마라.

• @property 메소드는 빠르게 유지하고, 복잡하거나 느린 작업은 별도의 메소드를 두어라.

30. Consider @property Instead of Refactoring Attributes

아래 예제를 보자.

def fill(bucket, amount):
    now = datetime.now()
    if now - bucket.reset_time > bucket.period_delta:
        bucket.quota = 0
        bucket.reset_time = now
    bucket.quota += amount

def deduct(bucket, amount):
    now = datetime.now()
    if now - bucket.reset_time > bucket.period_delta:
        return False
    if bucket.quota - amount < 0:
        return False
    bucket.quota -= amount
    return True

class Bucket(object):
    def __init__(self, period):
        self.period_delta = timedelta(seconds=period)
        self.reset_time = datetime.now()
        self.max_quota = 0       # 버켓에 들어있던 최대치
        self.quota_consumed = 0  # 사용한 양

    def __repr__(self):
        return ('Bucket(max_quota=%d, quota_consumed=%d)' %
               (self.max_quota, self.quota_consumed))

    @property
    def quota(self):  # @property 를 이용해 현재 남은 양을 조회할 수 있다.
        return self.max_quota - self.quota_consumed

    @quota.setter
    def quota(self, amount):  # setter는 적절하게 max_quota와 quota_consumed를 조절한다.
        delta = self.max_quota - amount
        if amount == 0:
            self.quota_consumed = 0
            self.max_quota = 0
        elif delta < 0:
            assert self.quota_consumed == 0
            self.max_quota = amount
        else:
            assert self.max_quota >= self.quota_consumed
            self.quota_consumed = delta

• @property는 기존의 어트리뷰트에 새로운 기능을 추가할 수 있다.
• @property로 위 예의 quota와 같은 증감이 있는 더 나은 데이터 모델을 만들 수 있다.
• @proprty를 너무 많이 쓴다 싶으면 리팩토링을 고려하라.

31. Use Descriptors for Reusable @property Methods

만약 여러 어트리뷰트 모두가 비슷한 동작을 하는 @property가 필요하다면 어떻게 해야할까?

class Exam(object):
    def __init__(self):
        self._writing_grade = 0
        self._math_grade = 0

    @staticmethod
    def _check_grade(value):
        if not (0 <= value <= 100):
            raise ValueError('Grade must be between 0 and 100')

    @property
    def writing_grade(self):
        return self._write_grade

    @writing_grade.setter
    def writing_grade(self, value):
        self._check_grade(value)
        eslf._writing_grade = value

    # ... _math_grade에 대해서도 반복

• 발리데이션과 같은 상황에서 같은 내용의 @property를 반복해서 써야 할 때에는, descriptor 클래스를 사용하라.

• decriptor는 __get__과 __set__메소드를 제공한다.
• 코드 재사용 측면에서 mix-in보다 낫다.

class Grade(object):
    def __init__(self):
        self._value = 0

    def __get__(self, instance, instance_type):
        return self._value

    def __set__(self, instance, value):
        if not (0 <= value <= 100):
            raise ValueError('Grade must be between 0 and 100')
        self._value = value


class Exam(object):
    # 클래스 어트리뷰트
    math_grade = Grade()
    writing_grade = Grade()
    science_grade = Grade()


exam = Exam()
exam.writing_grade = 40
# Exam.__dict__['writing_grade'].__set__(exam, 40) 이렇게 인터프리트 된다.
print(exam.writing_grade)
# print(Exam.__dict__['writing_grade'].__get__(exam, Exam)) 이렇게 인터프리트 된다.

인스턴스의 writing_grade에 접근하려고 할때, 인스턴스에 해당 어트리뷰트가 없으면 클래스 어트리뷰트를 찾는다. 이 클래스 어트리뷰트는 __get__, __set__ 메소드를 가진 오브젝트이고, 이제 descriptor 프로토콜이 적용된다.

그런데 위 코드에서는Exam 클래스가 정의될 때 Grade 인스턴스들이 한번만 생성되기 때문에, 모든 Exam 인스턴스들이 Grade를 공유한다는 문제가 있다.

따라서 Grade의 _value는 각 인스턴스의 값을 딕셔너리로 저장할 수 있어야 하는데, 일반적인 딕셔너리는 Exam 인스턴스가 지워지더라도 레퍼런스 카운트가 0이 되지 않고, 가비지 컬렉팅이 이루어지지 않아서 메모리 누수가 발생한다.

• desciptor에서 메모리를 효과적으로 관리하기 위해서는, weakref 모듈의 WeakKeyDictionary 모듈을 사용하라. 이 딕셔너리는 가지고 있는 키가 마지막 레퍼런스 인 경우에, 더이상 키를 유지하지 않는 특성이 있다.

class Grade(object):
    def __init__(self):
        self._value = WeakKeyDictionary()

    def __get__(self, instance, instance_type):
        if instance is None: return self
        return self._values.get(instance, 0)

    def __set__(self, instance, value):
        if not (0 <= value <= 100):
            raise ValueError('Grade must be between 0 and 100')
        self._values[instance] = value

32. Use __getattr__, __getattribute__, and __setattr__ for Lazy Attributes

데이터베이스와 오브젝트를 연결하려고 할 때, 오브젝트는 데이터베이스의 스키마를 알 필요가 없다. 이를 구현하기 위해 @property나 descriptor를 떠올릴 수 있겠지만, 이들은 미리 정의되어야 하므로 이런 상황에는 적합하지 않다. 이런 상황에서는 __getattr__ 메소드를 사용하면 된다. 만약 클래스가 이 메소드를 구현하고 있다면, 존재하지 않는 어트리뷰트를 호출하려고 할때, 이 메소드가 수행되어 동적으로 어트리뷰트를 생성할 수 있다.

class LazyDB(object):
    def __init__(self):
        self.exists = 5

    def __getattr__(self, name):
        value = 'Value for %s' % name
        setattr(self, name, value)
        return value


data = LazyDB()
print(data.__dict__)  # {'exists': 5}
print(data.foo)       # Value for foo
print(data.__dict__)  # {'exists': 5, 'foo': 'Value for foo'}

• __getattr__는 존재하지 않는 어트리뷰트에 접근하려고 했을 때 수행된다. 존재하는 어트리뷰트에 접근할때는 수행되지 않는다.

  class LoggingLazyDB(LazyDB):
      def __getattr__(self, name):
          print('Called __getattr__(%s)' % name)
          return super().__getattr__(name)  # 무한재귀에 빠지지 않도록 super를 쓴다.
  
  data = LoggingLazyDB()
  print(data.exists)  # 5
  print(data.foo)     # Called __getattr__(foo)
                      # Value for foo
  print(data.foo)     # Value for foo
  

• __getattribute__는 어트리뷰트에 접근할 때 마다 호출된다. __getattr__보다 우선순위가 높다.

  class ValidationDB(object):
      def __init__(self):
          self.exists = 5
  
      def __getattribute__(self, name):
          print('Called __getattribute__(%s)' % name)
          try:
              return super().__getattribute__(name)  # 기본 어트리뷰트를 바꾸지 않도록 super 먼저
          except AttributeError:  # 어트리뷰트가 없을 때 발생.
              value = 'Value for %s' % name
              setattr(self, name, value)
              return value
    
  data = ValidationDB()
  print(data.exists)  # 5
  print(data.foo)     # Called __getattribute__(foo)
                      # Value for foo
  print(data.foo)     # Called __getattribute__(foo)
                      # Value for foo
  

• 어트리뷰트를 제한하고 싶을 때애는 __getattr__과 __getattribute__모두에 AttributeError를 내도록 한다.

  class MissingPropertyDB(object):
      def __getattr__(self, name):
          if name == 'bad_name':
              raise AttributeError('%s is missing' % name)
          # ...
    
      def __getattribute__(self, name):
          if name == 'bad_name':
              raise AttributeError('%s is missing' % name)
          # ...
    
  

  __getattr__에서만 제한하면, __getattribute__보다 우선순위가 낮으므로 제한되지 않는다. __getattribute__에서만 제한하면 __getattr__이 호출되어 제한되지 않는다.

• __hasattr__는 어트리뷰트가 있는지 알려주는데, 내부적으로 __getattr__또는 __getattribute__를 호출한다.

• __setattr__는 두개로 나누어지지 않고, 값이 할당 될 때마다 호출된다.

  class LoggingSavingDB(object):
      def __setattr__(self, name, value):
          print('Called __setattr__(%s, %s)' % (name, value))
          super().__setattr__(name, value)
  
  data = LoggingSavingDB()
  data.foo = 5  # Called __setattr__(foo, 5)
  data.foo = 6  # Called __setattr__(foo, 6)
  
  

  setattr(self, name, value)는 self.__setattr__(name, value)와 같다.

• __getattribute__는 어트리뷰트에 접근할 때마다 호출되므로, 이 메소드 내에서 변수에 접근하면 무한루프에 빠진다.

  class BrokenDictionaryDB(object):
      def __init__(self, data):
          self._data = {}
  
      def __getattribute__(self, name):
          print('Called __getattribute__(%s)' % name)
          return self._data[name]  # <- __getattribute__(_data)를 호출한다.
          # 다음과 같이 되어야 한다.
          data_dict = super().__getattribute__('_data')
          return data_
  

33. Validate Subclasses with Metaclasses

클래스의 유효성은 __init__에서 확인할 수 있는데, 이렇게 하면 인스턴스가 생성될 때 에러를 출력하게 된다. 메타클레스를 활용하면 클래스가 정의될 때부터 유효성을 검증할 수 있다.

• 메타클래스는 type을 상속한다. __new__메소드는 클래스의 내용을 파라메터로 받으며, 클래스가 선언될 때 호출된다.

  class Meta(type):
      def __new__(meta, name, bases, class_dict):
          print((meta, name, bases, class_dict))
          return type.__new__(meta, name, bases, class_dict)
  
  class MyClass(object, metaclass=Meta):
      # __metaclass__ = Meta  <-- python2에서는 이렇게 메타클래스를 적용한다.
      stuff = 123
  
      def foo(self):
          pass
  
  # 다음이 출력된다.
  # (<class '__main__.Meta'>,
  #  'MyClass',
  #  (<class 'object'>, ),
  #  {'__module__': '__main__',
  #   '__qualname__': 'MyClass',
  #   'foo': <function MyClass.foo at 0x102c7dd08,
  #   'stuff': 123})
  

• __new__메소드에서 class_dict를 통해 클래스 어트리뷰트에 접근함으로써 유효성을 검증할 수 있다. 클래스 선언이 완료된 시점에서 호출된다.

  class ValidatePolygon(type):
      def __new__(meta, name, bases, class_dict):
          if bases != (object, ):
              if class_dict['sides'] < 3:
                  raise ValueError('Polygons need 3+ sides')
          return type.__new__(meta, name, bases, class_dict)
  
  class Polygon(object, metaclass=ValidatePolygon):
      sides = None
  
      @classmethod
      def inferior_angles(cls):
          return (cls.sides - 2) * 180
  
  class Triangle(Polygon):
      sides = 3
  
  print('Before class')      # Before class
  class Line(Polygon):
      print('Before sides')  # Before sides
      sides = 1
      print('After sides')   # After sides
                             # ValueError: Polygons need 3+ sides
  print('After class')       # 출력되지 않는다.
  

34. Register Class Existence with Metaclasses

메타클래스를 이용해, 새로 생성된 타입(클래스)들에 등록하는 등의 작업을 추가할 수 있다. 클래스를 시리얼라이즈 하고, 디시리얼라이즈 하는 예를 생각해보자.

class Serializable(object):
    def __init__(self, *args):
        self.args. = args

    def serialize(self):
        return json.dumps({'args': self.args})


class Deserializable(Serializable):
    @classmethod
    def deserialize(cls, json_data):
        params = json.loads(json_data)
        return cls(*params['args'])

point = Point2D(5, 3)
data = point.serialize  # {'args': [5, 3]}
point = Point2D(data)

이렇게 Deserializable까지 구현하고 나면, 이를 상속한 클래스는 시리얼라이즈, 디시리얼라이즈가 가능하다. 하지만 이런 구현은 json 데이터의 원래 클래스를 알지 못하면 적절한 클래스로 디시리얼라이즈 할 수 없다는 문제가 있다.

이를 해결하기 위해서는 시리얼라이즈 된 데이터에 클래스 명을 넣고, 나중에 이 클래스 명을 이용해 생성자를 가져오면 된다. 클래스 명으로 생성자를 가져오기 위해서는 저장소를 만들고 클래스를 정의할 때마다 등록한다.

class BetterSerializable(object):
    def __init__(self, *args):
        self.args. = args

    def serialize(self):
        return json.dumps({
            'class': self.__class__.__name__,
            'args': self.args
        })


registry = {}

def register_class(target_class):
    registry[target_class.__name__] = target_class

def deserialize(data):
    params = json.loads(data)
    name = params['class']
    target_class = registry[name]
    return target_class(*params['args'])

class Point2D(BetterSerializable):
    def __init__(self, x, y):
        super().__init__(x, y)
        self.x = x
        self.y = y

register_class(Point2D)  # 이렇게 등록을 꼭 해줘야한다.

새로 생긴 문제는 클래스를 정의할 때마다 register_class함수를 호출해 등록을 해줘야 한다는 점이다.

메타클래스를 이용해 클래스 정의 시점에 유효성 검사를 했던 것을생각해보면, 메타클래스로 이 작업을 클래스 정의 시점에 자동으로 적용되게 할 수 있다.

class Meta(type):
    def __new__(meta, name, bases, class_dict):
        cls = type.__new__(meta, name, bases, class_dict)
        register_class(cls)
        return cls

class RegisteredSerializable(BetterSerializable, metclass=Meta):
    pass

이제 RegisteredSerializable을 상속받은 클래스는, 정의 시점에서 register_class를 호출하게 되고, 별도의 과정 없이 시리얼라이즈, 디시리얼라이즈가 가능하다.

• class registration은 모듈러 프로그램을 만드는데 효과적인 패턴이다.
• 메타클래스를 이용해 서브클래스가 생성될 때 까먹지 않고 자동으로 등록할 수 있다.

35. Annotate Class Attributes with Metaclasses

메타클래스는 어노테이션이 달린 프로퍼티를 클래스 생성 후 실제로 사용되기 전에 수정할 수 있다. 데이터베이스에 있는 값을 가져오는 상황을 생각해보자.

class Field(object):
    def __init__(self, name):
        self.name = name
        self.internal_name = '_' + self.name

    def __get__(self, instance, instance_type):
        if instance is None: return self
        return getattr(instance, self.internal_name, '')
    
    def __set__(self, instance, value):
        setattr(instance, self,internal_name, value)


class Customer(object):
    first_name = Field('first_name')
    last_name = Field('last_name')  # 어트리뷰트의 이름을 넣어 주어야 한다.


foo = Customer()
print(repr(foo.first_name), foo.__dict__)  # '' {}
foo.first_name = 'Euclid'
print(repr(foo.first_name), foo.__dict__)  # 'Euclid' {'_first_name': 'Euclid'}.

위와 같은 구현은 어트리뷰트의 이름을 Field의 파라메터로 또 한번 써줘야 한다는 것이다. 메타클래스를 활용하면 이 반복을 줄일 수 있다.

class Field(object):
    pass  # 메타클래스에서 어트리뷰트가 할당되므로 초기화가 필요 없다.
  
class Meta(type):
    def __new__(meta, name, bases, class_dict):
        for key, value in class_dict.items():
            if isinstance(value, Field):
                value.name = key
                value.internal_name = '_' + key
        cls = type.__new__(meta, name, bases, class_dict)
        return cls

class DatabaseRow(object, metaclass=Meta):
    pass

class Customer(DatabaseRow):
    first_name = Field()
    last_name = Field()

• 메타클래스에서 어트리뷰트를 마저 할당해서 클래스를 완성할 수 있다.1
• 런타임에서 어트리뷰트를 할당하기 위해서는 descriptor와 메타클래스를 사용할 수 있다.
• 메타클래스와 descriptor 모두 weakref를 사용하지 않으면서도 메모리 누수를 잡을 수 있다.

Chapter 5. Concurrency and Parallelism

parallel 프로그램은 병렬수행으로 전체 작업 시간을 단축시키지만, concurrent 프로그램은 병렬적으로 수행되는 듯 하지만 전체 속도가 빨라지지는 않는다.

Python은 concurrent 프로그램을 짜기 쉽고, 시스템콜, 서브프로세스, C-extensions으로 병렬 처리도 할 수 있다. 하지만, concurrent 프로그램이, 진짜 병렬적으로 동작하도록 짜는기는 아주 어렵다.

36. Use subprocess to Manage Child Processes

Python에서 만든 child 프로세스는 CPU 코어를 모두 사용하게끔 병렬로 수행될 수 있다.

• popen, popen2, os.exec* 등이 있지만, 서브프로세스를 다루는데에는 subprocess 모듈이 최고다.

  proc = subprocess.Popen(['echo', 'Hello from the child!'], stdout=subprocess.PIPE)
  out, err = proc.communicate()
  print(out.decode('utf-8'))  # Hello from the child!
  

• child 프로세스는 parent 프로세스(Python)와 독립적으로 수행된다. 그 사이에 Python으로 다른 작업을 수행할 수 있다.

  proc = subprocess.Popen(['sleep', '0.3'])
  while proc.poll() is None:
      print('Working...')
  
  print('Exit status', proc.poll())
  
  # Working...
  # Working...
  # Exit status 0
  

• 여러 child 프로세스를 병렬적으로 돌릴 수 있다. communicate 메소드로 프로세스가 완료될 때까지 기다릴 수 있다.

  def run_sleep(period):
      proc = subprocess.Popen(['sleep', str(period)])
      return proc
  
  start = time()
  procs = []
  for _ in range(10):
      proc = run_sleep(0.1)
      procs.append(proc)
  
  for proc in procs:
      proc.communicate()
  end = time()
  print('Finished in %.3f seconds' % (end - start))  # Finished in 0.117 seconds
  

• stdin 을 입력할 수 있다.

  def run_openssl(data):
      env = os.environ.copy()
      env['password'] = b'\xe24U\n\xd0Ql3S\x11'
      proc = subprocess.Popen(['openssl', 'enc', '-des3', '-pass', 'env:password'],
                             env=env,
                             stdin=subprocess.PIPE,
                             stdout=subprocess.PIPE)
      proc.stdin.write(data)
      proc.stdin.flush()  # child가 인풋을 받게 한다.
      return proc
  
  procs = []
  for _ in range(3):
      data = os.urandom(10)  # 랜덤 바이트. 암호화 할 네트워크 소켓, 파일, 유저인풋 등이 될 수 있다.
      proc = run_openssl(data)
      procs.append(proc)
  
  for proc in procs:
      out, err = proc.communicate()
      print(out[-10:])
  
  # b']~\x87G\x9d\xaa2\xa4=:'
  # b'\xac\xce\x15\x1d\nDy\xa5Z\x89'
  # b'\xcb\x88\x02Xw\x08P-\xf9\xb6'
  

• PIPE로 여러 쉘 스크립트를 연결해서 사용할 수 있다.

  def run_md5(input_stdin):  # hashlib 모듈에서 md5를 제공하므로 이런 함수를 쓸 일은 사실 없다.
      proc = subprocess.Popen(['md5'],
                             stdin=input_stdin,
                             stdout=subprocess.PIPE)
      return proc
  
  input_procs = []
  hash_procs = []
  for _ in range(3):
      data = os.urandom(10)
      proc = run_openssl(data)E
      input_procs.append(proc)
      hash_proc = run_md5(proc.stdout)  # .stdout를 인자로 넣어 연결한다.
      hash_procs.append(hash_proc)
  
  for proc in input_procs:
      proc.communicate()
  for proc in hash_procs:
      out, err = proc.communicate()
      print(out.strip())
  

• Python3에서는 타임아웃을 걸 수 있다. Python2에서는 proc.stdin, proc.stdout, proc.stderr에 select 모듈을 쓰면 된다.

  proc = run_sleep(10)
  try:
      proc.communicate(timeout=0.1)
  except subprocess.TimeoutExpired:
      proc.terminate()
      proc.wait()
  
  print(proc.poll())  # -15
  

37. Use Threads for Blocking I/O, Avoid for Parallelism

기본 Python은 CPython이다. CPython은 우선 바이트코드로 컴파일되고, 스택기반 인터프리터로 수행된다. 바이트코드 인터프리터는 수행되는동안 상태를 일관되게 유지해야한다. Python은 GIL(global interpreter lock)으로 일관성을 보장한다.

GIL은 CPython이 preemptive 멀티쓰레딩으로부터 영향받지 않도록 하는 mutex이다. 쓰레드 인터럽트는 인터프리터의 상태를 망가뜨릴 수 있다. GIL은 이런 인터럽트를 막고, 모든 바이트코드 명령어들이 제대로 동작하게 한다.

GIL은 치명적인 문제가 있는데, Python자체가 멀티쓰레드를 지원함에도 불구하고 멀티쓰레드를 이용하는 C++나 Java코드를 병렬처리 할 수 없다는 점이다.

인수들을 구하는 코드를 단순히 순서대로 돌렸을 경우에 걸리는 시간을 보자

def factorize(number):
    for i in range(1, number + 1):
        if number % i == 0:
            yield i

numbers = [24134124, 3412334, 12434124, 343414124]
start = time()
for number in numbers:
    list(factorize(number))
end = time()
print('Took %.3f seconds' % (end - start))  # Took 31.118 seconds

멀티쓰레딩을 사용해 처리하면 다음과 같다.

from threading import Thread

class FactorizeThread(Thread):
    def __init__(self, number):
        super().__init__()
        self.number = number

    def run(self):
        self.factors = list(factorize(self.number))

start = time()
threads = []
for number in numbers:
    thread = FactorizeThread(number)
    thread.start()
    threads.append(thread)
for thread in threads:
    thread.join()
end = time()
print('Took %.3f seconds' % (end - start))  # Took 31.706 seconds

• 심지어 더 많이 걸린다. 이처럼 일반적인 CPython 인터프리터에서는, GIL 때문에 여러 바이트코드를 병렬로 돌릴 수 없다.

CPython으로 멀티코어를 활용할 수는 있으나, 사실 Thread 모듈로는 할 수 없고 추가적인 노력이 필요하다. 그럼에도 불구하고 쓰레드를 사용하는 이유는 크게 두가지가 있다.

첫째로, 실제로는 동시에 수행되지 않더라도, 겉으로 보기에 프로그램이 병렬적인 작업을 수행한다는것을 명시할 수 있다.

둘째로, 파일 읽기쓰기, 네트워크 통신, 하드웨어 장치 조작 등의 blocking I/O를 다루기 위해서다. Python은 몇가지 외부 환경과 상호작용할 수 있도록 OS에 시스템콜을 할 수 있는데, 이때 쓰레드는 프로그램을 OS의 작업과 분리시킴으로써 blocking I/O를 다루는 것을 도와준다.

예를들어 원격 헬리콥터에 시리얼포트를 통해 신호를 보낸다고 해보자. 이 작업을 위한 프록시로 비교적 느린 시스템콜(select)를 사용할텐데, 아래 함수는 싱크로너스 시리얼포트를 사용하는 것 처럼 OS로 하여금 0.1초 정도 블락 한 후 프로그램에 권한을 넘겨주도록한다.

import select

def slow_systemcall():
    select.select([], [], [], 0.1)

start = time()
for _ in range(5):
    slow_systemcall()
end = time()
print('Took %.3f seconds' % (end - start))  # Took 0.516 seconds

이 코드의 문제는 프로그램이 slow_systemcall을 돌리는 동안 다른 작업을 하지 못한다는 것이다. 프로그램의 메인 쓰레드는 select 시스템 콜을 하는 쓰레드에 의해 블락된다. 헬기에 명령을 내리고, 그 와중에 계속 계산 작업을 수행하려면 어떻게 해야할까. blocking I/O와 계산을 동시에 수행하고 싶다면, 시스템콜을 쓰레드로 옮겨야한다.

아래 코드는 동시에 여러 시리얼포트의 헬기에 명령을 내리면서, 메인 쓰레드는 계산작업을 수행할 수 있게 한다.

start = time()
threads = []
for _ in range(5):
    thread = Thread(target=slow_systemcall)
    thread.start()
    threads.append(thread)

def compute_helicopter_location(index):
    # ...

for i in range(5):
    compute_helicopter_location(i)
for thread in threads:
    thread.join()
end = time()

print('Took %.3f seconds' % (end - start))  # Took 0.106 seconds

이처럼 시스템콜은 CIL이 있더라도 병렬적으로 수행된다. GIL은 Python코드가 병렬로 동작하는 것은 막지만, 시스템콜에는 영향을 미치지 않는다. 이는 Python이 시스템콜을 호출하기 직전에 GIL을 놓고, 명령을 내린 후 발고 GIL을 다시 얻기 때문에 가능하다.

• 다시말해 쓰레드는 시스템콜을 병렬적으로 수행할 수 있으며, blocking I/O와 계산을 동시에 할수 있도록 한다.

  asyncio같은 모듈로 blocking I/O를 처리하는 다른 방법들이 있고 장점도 있지만, 이들은 다른 모델에 적용하고자 할 때 리팩토링이 필요하다. 쓰레드는 프로그램의 변화를 최소로 하면서 blocking I/O를 할 수 있는 가장 간단하고 좋은 방법이다.

38. Use Lock to Prevent Data Races in Threads

GIL이 멀티쓰레드가 병렬적으로 수행되는 것을 막기 때문에 별도의 mutex 락이 필요 없다고 생각하기 쉬지만 그렇지 않다. 한번에 하나의 Python 쓰레드만 동작하기는 하지만, 자료구조에 쓰는 작업은 다른 바이트코드 명령에 의해 영향 받을 수 있다. 따라서 여러 쓰레드에서 동시에 한 오브젝트를 사용하는 것은, 자료구조의 불변성을 해칠 수 있고, 프로그램의 상태를 망칠 수 있다.

• 아래와 같이 여러 쓰레드가 한 오브젝트의 값을 수정하는 경우 데이터가 어긋날 수 있다.

class Counter(object):  # 카운트를 위한 클래스이다.
    def __init__(self):
        self.count = 0

    def increment(self, offset):
        self.count += offset


def worker(how_many, counter):  # how_many만큼 카운터를 증가시킨다.
    for _ in range(how_many):
        counter.increment(1)


def run_threads(func, number_of_thread, *args):  # 여러 쓰레드에서 func를 수행한다.
    threads = []
    for _ in range(number_of_thread):
        thread = Thread(target=func, args=args)
        threads.append(thread)
        thread.start()
    for thread in threads:
        thread.join()

number_of_thread = 5
how_many = 10**5
counter = Counter()  # 여러 쓰레드는 한 카운터를 공유한다.
run_threads(worker, number_of_thread, how_many, counter)
print('Counter should be %d, found %d' % (number_of_thread * how_many, counter.count))
# Counter should be 500000, found 383202

• GIL에 의해 한번에 한 쓰레드만 수행하긴 하지만, 여러 쓰레드에 계산 시간을 동등하게 배분하면서 빠르게 스위치한다. 기존 쓰레드가 중지되고, 다른 쓰레드로 CPU 자원을 넘기는 일은 atomic한 연산 중간에도 벌어질 수 있다. +=과 같은 연산은 실제로 getattr, +, setattr의 세가지 연산으로 이루어져 있기 때문에 위와 같은 일이 벌어진다.

• threading 모듈의 Lock을 이용하면 mutex를 보장하여, 작업 중간에 쓰레드가 바뀌는 일을 막을 수 있다.

class LockingCounter(object):
    def __init__(self):
        self.lock = Lock()
        self.count = 0

    def increment(self, offset):
        with self.lock:  # with 구문을 사용하여 lock을 얻고, 작업이 끝나면 리리즈 하도록 한다.
            self.count += offset

counter = Counter()  # 여러 쓰레드는 한 카운터를 공유한다.
run_threads(worker, number_of_thread, how_many, counter)
print('Counter should be %d, found %d' % (number_of_thread * how_many, counter.count))
# Counter should be 500000, found 500000

39. Use Queue to Coordinate Work Between Threads

concurrent 잡들의 결과를 연결하는 파이프라인 작업을구성해야 할 때가 있다. 특히 병렬적으로 수행될 수 있는 blocking I/O나, subprocess를 포함하는 경우 유용하다.

이미지를 다운받아 사이즈를 조절하고 다시 업로드 하는 작업을 concurrent하게 연결한다고 해보자. 파이프라인의 각 단계간에 작업결과물을 넘겨줘야 하는데, 이는 thread-safe한 producer-consumer queue로 구현할 수 있다. 프로듀서는 이미지를 대기열의 마지막에 추가하고, 컨슈머는 대기열의 앞에서부터 이미지를 처리한다.

class MyQueue(object):
    def __init__(self):
        self.items = deque()
        self.lock = Lock

    def put(self, item):
        with self.lock:
            self.items.append(item)

    def get(self):
        with self.lock:
            return self.items.popleft()

워커는 큐에서 작업을 빼서 쓰레드로 수행하고 그 결과를 다른 큐에 넣는다.

class Worker(Thread):
    def __init__(self, func, in_queue, out_queue):
        super().__init__()
        self.func = func
        self.in_queue = in_queue    # 작업할 내용을 받을 큐를 지정한다.
        self.out_queue = out_queue  # 작업 결과를 보낼 큐를 지정한다.
        self.polled_count = 0
        self.work_done = 0

    def run(self):
        while True:  # 무한루프를 돌며 큐를 확인하고 처리한다.
            self.polled_count += 1
            try:
                item = self.in_queue.get()
            except IndexError:
                sleep(0.01)  # 남은 작업이 없는 경우 IndexError를 낸다.
            else:
                result = self.func(item)
                self.out_queue.put(result)
                self.work_done += 1

이제 작업에 맞게 큐를 생성하고, 워커에게 할당한다.

download_queue = MyQueue()
resize_queue = MyQueue()
upload_queue = MyQueue()
done_queue = MyQueue()
threads = [
  Worker(download, download_queue, resize_queue),
  Worker(resize, resize_queue, upload_queue),
  Worker(upload, upload_queue, done_queue),
]

for thread in threads:
    thread.start()
for _ in range(1000):  # 테스트를 위해 큐에 더미 오브젝트를 넣는다.
    download_queue.put(object())
while len(done_queue.items) < 1000:  # 작업이 끝나길 busy waiting한다.
    # 작업을 하는 동안 다른 일을 할 수 있다.
    # ...
processed = len(done_queue.items)
polled = sum(t.polled_count for t in threads)
print('Processed', processed, 'items after polling', polled, 'times')
# Processed 1000 items after polling 3030 items

큐를 확인하기 위해 반복적으로 폴링하고, IndexError를 내는 경우가 있기 때문에 폴링 횟수는 3000번보다 더 많게 된다. 만일 파이프라인에서 빠른 작업이 느린 작업 뒤에 위치한다면, 큐를 확인하고 에러를 내는 작업이 더 많아지고 결국 CPU자원을 낭비하게된다.

위 구현엔 잘못된 점이 몇가지 더 있다. 작업이 다 되었는지 확인하는 busy waiting이 있다는 점, 그리고 워커 쓰레드에게 작업종료를 알리는 시그널이 없어 이 busy 루프 안에서 Worker의 run 메소드가 영원히 수행된다는 점이다. 무엇보다 가장 문제는, 나중 작업이 느릴 경우에 앞 큐는 한없이 길어지고 메모리부족으로 프로그램이 죽을 수도 있다는 점이다.

queue모듈의 Queue클레스에는 이런 문제들을 해결할 수 있는 기능이 있다.

• Queue의 get 메소드는 새로운 데이터가 있을때까지 block시킴으로써, 워커가 busy waiting 하지 않게끔 한다.

  from queue import Queue
  queue = Queue()
  
  def consumer():
      print('Consumer waiting')
      queue.get()  # 아래의 put 이후에 수행된다.
      print('Consumer done')
  
  thread = Thread(target=consumer)
  thread.start()
  
  print('Producer putting')
  queue.put(object())  # 위의 get 이전에 수행된다.
  thread.join()
  print('Producer done')
  
  # 출력
  # Consumer waiting
  # Producer putting
  # Consumer done
  # Producer done
  

  위에서 보는 것처럼, thread가 먼저 시작되었더라도, put되기 전에 get을 수행하지 않는다.

• Queue(number)과 같이 인스턴스를 생성하여, 버퍼 사이즈를 지정할 수 있다.

  queue = Queue(1)  # 버퍼사이즈를 1로 설정한다.
  
  def consumer():
      time.sleep(0.1)          # 대기
      queue.get()              # 두번째로 수행된다.
      print('Consumer got 1')
      queue.get()              # 네번째로 수행된다.
      print('Consumer got 2')
  
  thread = Thread(target=consumer)
  thread.start()
  queue.put(object())          # 처음으로 수행된다.
  print('Producer put 1')
  queue.put(object())          # 세번째로 수행된다.
  print('Producer put 2')
  thread.join()
  print('Producer done')
  
  # 출력
  # Producer put 1
  # Consumer got 1
  # Producer put 2
  # Consumer got 2
  # Producer done
  

• task_done 메소드로 작업의 진행상황을 추적할 수 있다. 이로써 done_queue를 폴링하는 작업을 없앨 수 있다.

  in_queue = Queue()
  
  def consumer():
      print('Consumer waiting')
      work = in_queue.get()      # 두번째로 완료된다.
      print('Consumer working')
      # Doing work ...
      print('Consumer done')
      in_queue.task_done()       # 세번째로 완료된다.
  
  Thread(target=consumer).start()
  in_queue.put(object())         # 처음으로 완료된다.
  print('Producer waiting')
  in_queue.join()                # 마지막으로 완료된다.
  
  # 출력
  # Consumer waiting
  # Producer waiting
  # Consumer working
  # Consumer done
  # Producer done
  

아래와 같이, 작업종료를 알리는 작업을 큐에 넣어 워커들을 종료시키는 큐를 만들 수 있다. __iter__ 메소드에서 작업종료를 확인하고 이터레이션을 종료시키며, task_done로 작업 진행을 알수 있게 한다.

class ClosableQueue(Queue):
    SENTINEL = object()

    def close(self):
        self.put(self.SENTINEL)

    def __iter__(self):
        while True:
            item = self.get()
            try:
                if item = self.SENTINEL:
                    return  # iteration을 중지하고, thread가 종료되게한다.
                yield item
            finally:
                self.task_done()

이제 새로 만든 큐에 맞게 워커를 다시 작성하는데, 이 워커는 이터레이션이 끝나면 종료된다.

class StoppableWorker(Thread):
    def __init__(self, func, in_queue, out_queue):
        # ...

    def run(self):
        for item in self.in_queue:
            result = self.func(item)
            self.out_queue.put(result)

큐를 join해서 작업이 끝나길 기다리고, 다음 큐를 종료한다.

download_queue = ClosableQueue()
resize_queue = ClosableQueue()
upload_queue = ClosableQueue()
done_queue = ClosableQueue()
threads = [
  StoppableWorker(download, download_queue, resize_queue),
  StoppableWorker(resize, resize_queue, upload_queue),
  StoppableWorker(upload, upload_queue, done_queue),
]

for thread in threads:
    thread.start()
for _ in range(1000):  # 테스트를 위해 큐에 더미 오브젝트를 넣는다.
    download_queue.put(object())
download_queue.close()  # 작업을 종료한다.
download_queue.join()
resize_queue.close()
resize_queue.join()
upload_queue.close()
upload_queue.join()
print(done_queue.qsize(), 'items finished')
# 1000 items finished

• 파이프라인으로 멀티 쓰레드를 이요한 concurrent 작업 시퀀스를 처리할 수 있다.
• busy waiting, 워커 종료, 메모리 관리를 신경쓰라.
• Queue 클래스엔 blocking, 버퍼사이즈, join, 등 파이프라인을 만드는데 필요한 유용한 기능들이 있다.

40. Consider Coroutines to Run Many Functions Concurrent

쓰레드가 여러 작업을 동시에 수행할 수 있게 하긴 하지만, 몇가지 문제가 있다.

• 쓰레드간 작업공유를 위해 Lock, Queue와 같은 별도의 자료구조들이 필요하기에, 코드가 복잡해진다.
• 쓰레드당 대략 8mb의 메모리가 필요하다. 수천 수만 유저의 리퀘스트를 처리해야하는 서버나, 파티클 시뮬레이션 등과 같은 상황에서, 각 동작에 별도의 쓰레드를 할당하는 것은 불가능하다.
• 쓰레드를 계속해서 생성하고 종료하면 오버헤드가 쌓여서 전체 시스템이 느려진다.

coroutine을 쓰면 이런 문제를 피할 수 있다. coroutine은 제너레이터의 확장으로, 오버헤드가 거의 없고, 1kb미만의 메모리를 사용한다.

coroutine은 각 yield 이후에 값을 함수로 다시 돌려받는 제너레이터라고 볼 수 있다. 제너레이터 함수는 yield의 결과로, send함수로 보내진 값을 받는다.

def my_coroutine():
    while True:
        receive = yield
        print('Received:', received)

it = my_coroutine()
next(it)             # coroutine을 준비시킨다.
it.send('First')     # Receive: First
it.sned('Second')    # Receive: Second

next는 제너레이터가 yield함수까지 진행해서 send로 값을 받을 수 있게 준비시킨다. yield와 send는 제너레이터가 yield된 값을 바꿔가며 동작할 수 있게 한다.

예를들어 받은 값들 중 최소 값을 반환하는 coroutine을 구현한다면 다음과 같다.

def minimize():
    current = yield  # 단순한 yield로 coroutine을 준비하고 초기값을 받는다.
    while True:
        value = yield current  # 현재까지의 최소값을 반환하며, 다음 값을 받는다.
        current = min(value, current)

it = minimize()
next(it)
print(it.send(10))  # 10
print(it.send(4))   # 4
print(it.send(22))  # 4
print(it.send(-1))  # -1

The Game of Life

coroutine으로 그리드 위의 셀들이 죽었다 살았다 변하는 콘웨이의 라이프 게임을 만들어보자.

ALIVE = '*'
EMPTY = '-'

Query = namedtuple('Query', ('y', 'x'))

def count_neighbors(y, x):  # 주변에 살아있는 셀들의 수를 계산한다.
    n_ = yield Query(y + 1, x + 0)
    ne = yield Query(y + 1, x + 1)
    e_ = yield Query(y + 0, x + 1)
    se = yield Query(y - 1, x + 1)
    s_ = yield Query(y - 1, x + 0)
    sw = yield Query(y - 1, x - 1)
    w_ = yield Query(y + 0, x - 1)
    nw = yield Query(y + 1, x - 1)
    neighbor_states = [n_, ne, e_, se, s_, sw, w_, nw]
    count = 0
    for state in neighbor_states:
        if state == ALIVE:
            count += 1
    return count

Transition = namedtuple('Transition', ('y', 'x', 'state'))

def step_cell(y, x):  # 셀의 다음 상태를 계산한다.
    state = yield Query(y, x)
    neighbors = yield from count_neighbors(y, x)  # yield from으로 coroutine들을 결합한다.
    next_state = game_logic(state, neighbors)  # yield from이 다 끝나면 수행된다.
    yield Transition(y, x, next_state)

def game_logic(state, neighbors):  # 주변에 살아있는 셀이 셋이면 살고, 아니면 죽는다.
    if state == ALIVE:
        return ALIVE if 2 <= neighbors <= 3 else EMPTY
    else:
        return ALIVE if neighbors == 3 else EMPTY

TICK = object()  # 게임 싸이클 한번이 끝났음을 뜻한다.

def simulate(height, width):  # 셀들의 상태를 계속해서 계산하고 반환한다.
    while True:
        for y in range(height):
            for x in range(width):
                yield from step_cell(y, x)
        yield TICK

이와같이 구현하면 simulate는 주변 환경들과 완전히 분리된다. 그리드가 어떻게 구현되는지, Query, Transition, TICK 값들이 어떻게 다루어지는지, 초기값이 어떻게 세팅되는지 정의하지 않고 로직에만 집중한다. 각 셀들은 step_cell에 의해 변화하고 모두 변화하고 나면 TICK으로 갱신의 끝을 알린다.

coroutine은 이처럼 로직에만 집중할 수 있게 한다. 환경에 대한 코드와 하고자 하는 구현을 분리시키고, coroutine들이 병렬적으로 동작하는 것처럼 보이게 한다.

이제 simulate이 동작할 주변 환경을 구현해보자.

class Grid(object):
    def __init__(self, height, width):
        self.height = height
        self.width = width
        self.rows = []
        for _ in range(self.height):
            self.rows.append([EMPTY] * self.width)

    def __str__(self):
        return '\n'.join([''.join(row) for row in self.rows])

    def query(self, y, x):
        return self.rows[y % self.height][x % self.width]

    def assign(self, y, x, state):
        self.rows[y % self.height][x % self.width] = state

마지막으로 simulate과 그 내부 다른 coroutine에서 yield되는 값을 사용하는 함수를 작성한다. 이 함수는 coroutine의 동작과 주변 환경을 연결한다. 그리드의 모든 셀에 대해 다음 상태를 계산하고 새로운 그리드를 반환한다.

def live_a_generation(grid, sim):
    progeny = Grid(grid.height, grid.width)
    item = next(sim)
    while item is not TICK:
        if isinstance(item, Query):
            state = grid.query(item.y, item.x)
            item = sim.send(state)
        else:
            progeny.assign(item.y, item.x, item.state)
            item = next(sim)
    return progeny