Python 기초를 다지기 위해 효과적인 파이썬이란 책을 공부 중입니다.
8장 강건성과 성능
- 기능을 개발한 후에는
- 오류가 발생해도 문제가 없도록 프로덕션화 해 코드를 방탄처리를 해야 한다
- 강건성(robust)에는 규모 확장성 과 성능 이라는 차원이 포함된다
65) try/except/else/finally
의 각 블록을 잘 활용하라
- finally 블록
예외가 발생하더라도 정리 코드를 실행해야 한다면, try/finally를 사용하라
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
|
def try_finally_example(filename):
print('* file open')
handle = open(filename, encoding='utf-8') # OSError
# 파일 open 오류는 finally 처리가 되어서는 안되므로
# try 블록 바깥에서 실행
try:
print('* read data')
return handle.read() # UnicodeDecodeError
# 예외 발생시 try_finally_example 호출자에게 예외 반환
finally:
print('* close file')
handle.close() # try 블록이 실행된 다음에는 항상 이 블록이 실행됨
# => 예외가 발생하더라도, 먼저 파일을 닫는다 (그 후에 예외 반환)
#############################
filename = 'random_data.txt'
with open(filename, 'wb') as f:
f.write(b'\xf1\xf2\xf3\xf4') # 잘못된 utf-8 이진 문자열
data = try_finally_example(filename)
|
- else 블록
코드에서 처리할 예외와 호출 스택을 거슬러 올라가며 전달할 예외를 명확히 구분하기 위해 try/except/else 를 사용하라.
try 블록이 예외를 발생시키지 않으면 else 블록이 실행된다.
else 블록을 사용하면 try 블록 안에 들어갈 코드를 최소화 할 수 있다. (가독성 좋아짐)
==> 원래 한묶음으로 작성될 코드가 try 블록으로 갈라지게 되는 경우
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
|
import json
def load_json_key(data, key):
try:
print('* read JSON data')
result_dict = json.loads(data) # ValueError
except ValueError as e:
print('* ValueError exception')
raise KeyError(key) from e
else: # try 블록 외부
print('* search key')
return result_dict[key] # KeyError
###############################
assert load_json_key('{"foo": "bar"}', 'foo') == 'bar'
>>> 정상
load_json_key('{"foo": "bar', 'foo') == 'bar'
>>>
Traceback ...
JSONDecodeError: ...
Traceback ...
KeyError: 'foo'
|
- 모든 요소를 한꺼번에 사용하기
예를 들어, 파일을 읽어 처리한 다음에 원본 파일 자체를 변경하고 싶다면
- try 블록: 파일을 읽고 처리
- except 블록: try 블록에서 예상되는 예외를 처리
- else 블록: 원본 파일의 내용을 변경하고, 오류 발생시 돌려준다
- finally 블록: 파일 핸들을 닫는다
66) 재사용 가능한 try/finally
동작을 원한다면 contextlib과 with 문을 사용하라
- with 문을 사용하면 try/finally 블록을 통해 사용해야 하는 로직을 재활용 하면서 시각적인 잡음도 줄 일 수 있다.
-
contextlib 내장모듈
이 제공하는 contextmanager 데코레이터
를 사용하면 사용자 함수를 with문
에 사용할 수 있다.
- 컨텍스트 매니저가 yield 하는 값은
with문
의 as 부분에 전달된다.
이를 활용하면 특별한 컨텍스트 내부에서 실행되는 코드 안에서 직접 그 컨텍스트에 접근할 수 있다.
- with 문과 try/finally 블록 (같다)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
|
from threading import Lock
lock = lock()
# with 블록에서 acquire(), release() 를 처리해줌
with lock:
# 어떤 불변 조건을 유지하면서 작업을 수행한다
...
###########################
# with 문을 사용하면 try/finally 와 구조가 같다
lock.acquire()
try:
# 어떤 불변 조건을 유지하면서 작업을 수행한다
...
finally:
lock.release()
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
|
import logging
# 기본 로그수준은 warning 이라서, 사전 설정을 해주지 않으면 error 만 출력
def my_function():
logging.debug('디버깅')
logging.error('이 부분은 오류')
logging.debug('추가 디버깅')
######################################
# 사용자 정의 with 구절
from contextlib import contextmanager
@contextmanager
def debug_logging(level):
logger = logging.getLogger()
old_level = logger.getEffectiveLevel()
logger.setLevel(level)
try:
# yield 식은 with 블록의 내용이 실행되는 부분을 의미함
yield
finally:
# with 블록이 끝나면 원래 log level 로 복구한다
logger.setLevel(old_level)
with debug_logging(logging.DEBUG):
print('* 내부 블록')
my_function() # 모두 출력
print('* 외부 블록')
my_function() # error 만 출력
|
- with 와 대상 변수 함께 사용하기
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
with open('my_output.txt', 'w') as handle:
handle.write('데이터입니다')
@contextmanager
def log_level(level, name):
logger = logging.getLogger(name)
old_level = logger.getEffectiveLevel()
logger.setLevel(level)
try:
yield logger # as 문으로 받아서 사용할 수 있음
finally:
logger.setLevel(old_level)
with log_level(logging.DEBUG, 'my-log') as logger:
logger.debug(f'대상: {logger.name}')
logging.debug('설정된 logger 를 사용 안하기 때문에 출력 안됨')
|
67) 지역 시간에는 time보다는 datetime
을 사용하라
- 여러 다른 시간대를 변환할 때는 time 모듈을 쓰지 말라
- 여러 다른 시간대를 신뢰할 수 있게 변환하고 싶으면 datetime 과 pytz 모듈을 함께 사용하라
- 항상 시간을 UTC 로 표시하고, 최종적으로 표현하기 직전에 지역시간으로 변환하라
- time 모듈
- datetime 모듈
68) copyreg를 사용해 pickle을 더 신뢰성 있게 만들라
- 신뢰할 수 있는 프로그램 사이에 객체 직렬화/역직렬화에 pickle 내장 모듈이 유용하다
- 이후 클래스가 바뀔 경우 (속성의 추가/삭제 등), 이전의 pickle 객체는 안전하지 않다
- 직렬화한 객체의 하위 호환성을 보장하려면 copyreg 내장모듈과 pickle 을 함께 사용하라
- 설계상 pickle 모듈의 직렬화 형식은 안전하지 않다
- 디폴트 애트리뷰트 값
- 클래스 버전 지정
- 안정적인 임포트 경로
69) 정확도가 매우 중요한 경우에는 decimal을 사용하라
- 돈과 관련된 계산 등 높은 정밀도가 필요하거나 근삿값 계산을 제어해야 할 때는
Decimal 클래스
가 좋다
- 부동 소수점으로 정확한 답을 계산해야 한다면 Decimal 생성자에
str 인스턴스
를 넘겨라 (float 가 아니라)
70) 최적화하기 전에 프로파일링을 하라
- 파이썬 성능을 느리게 하는 원인이 불분명한 경우가 많으므로 최적화하기 전에
프로파일링
하는 것이 중요하다
- profile 대신 cProfile 모듈을 사용하라 (더 정확한 정보를 제공한다)
- 함수 호출 트리를 독립적으로 프로파일링 하고 싶다면 Profile 객체의 runcall 메서드를 사용하면 된다
- Stats 객체를 사용하면 프로파일링 정보 중에서 살펴봐야할 부분만 선택해 출력할 수 있다
71) 생산자-소비자 큐로 deque를 사용하라
- 생산자는 append를 호출해 원소를 추가하고 소비자는 pop(0) 을 사용해 원소를 받게 만들면 리스트 타입을 FIFO 큐로 사용할 수 있다.
하지만 큐 길이가 늘어남에 따라 pop(0) 성능이 나빠져 문제가 될 수 있다.
- collections 내장 모듈의 deque 클래스는 큐 길이와 관계없이 상수 시간 안에 append 와 popleft 를 수행한다.
==> FIFO 큐 구현에 이상적
72) 정렬된 시퀀스를 검색할 때는 bisect를 사용하라
- 정렬된 리스트 데이터를 검색할 때 index 메서드 혹은 for 루프로 맹목적인 비교를 사용하면 선형 시간이 걸린다
- bisect 내장 모듈의 bisect_left 함수는 이진 검색 방식으로 O(logN) 시간이 걸린다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
# index 함수 수행시 데이터 길이만큼 선형 시간이 소요된다
data = list(range(10**5))
index = data.index(91234)
assert index == 91234
# for 루프에 의한 탐색도 마찬가지로 데이터 길이만큼 시간 소요
def find_closest(sequence, goal):
for index, value in enumerate(sequence):
if goal < value:
return index
raise ValueError(f'범위를 벗어남: {goal}')
index = find_closest(data, 91234.56)
assert index == 91234
|
1
2
3
4
5
6
7
8
|
# 이진검색 알고리즘 : 복잡도 O(logN)
from bisect import bisect_left
index = bisect_left(data, 91234) # exact match
assert index == 91234
index = bisect_left(data, 91234.56) # closest match
assert index == 91235
|
73) 우선순위 큐로 heapq를 사용하는 방법을 알아두라
- 우선순위 큐를 사용하면 선입선출이 아니라 원소의 중요도에 따라 원소를 처리할 수 있다
- 리스트로 우선순위를 구현하면 큐 크기가 커짐에 따라 프로그램 성능이 나빠진다
- heapq 내장모듈은 효율적으로 규모 확장이 가능한 우선순위 큐를 구현하는데 필요한 기능을 제공한다
- heapq 를 사용하려면 원소들이 자연스러운 순서를 가져야 한다. (
__lt__
같은 특별 메소드 필요)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
|
class Book:
def __init__(self, title, due_date):
self.title = title
self.due_date = due_date
def add_book(queue, book):
queue.append(book)
queue.sort(key=lambda x: x.due_date, reverse=True)
queue = []
add_book(queue, Book('돈키호테', '2020-06-07'))
add_book(queue, Book('프랑켄슈타인', '2020-06-05'))
add_book(queue, Book('레미제라블', '2020-06-08'))
add_book(queue, Book('전쟁과 평화', '2020-06-03'))
class NoOverdueBook(Exception):
pass
# 이 기능이 수행되기 위해서는 book list 가 정렬되어 있어야 한다
def next_overdue_book(queue, now):
if queue:
book = queue[-1]
if book.due_date < now:
queue.pop()
return book
raise NoOverdueBook
now = '2020-06-10'
found = next_overdue_book(queue, now)
print(found.title) # 전쟁과 평화
found = next_overdue_book(queue, now)
print(found.title) # 프랑켄슈타인
def return_book(queue, book):
queue.remove(book)
queue = []
book = Book('보물섬', '2020-06-04')
add_book(queue, book)
print('반납 전:', [x.title for x in queue]) # 보물섬
return_book(queue, book)
print('반납 후:', [x.title for x in queue])
try:
next_overdue_book(queue, now)
except NoOverdueBook:
pass # 이 문장이 실행될 것임
else:
assert False # 이 문장은 실행되지 않음
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
|
import random
import timeit
def print_result(count, tests):
# ...
pass
def print_delta(before, after):
# ...
pass
def list_overdue_benchmark(count):
def prepare():
to_add = list(range(count))
random.shuffle(to_add)
return [], to_add
def run(queue, to_add):
for i in to_add:
queue.append(i)
queue.sort(reverse=True)
while queue:
queue.pop()
test = timeit.repeat(
setup='queue, to_add = prepare()',
stmt=f'run(queue, to_add)'
globals=locals(),
repeat=100,
number=1
)
return print_results(count, tests)
baseline = list_overdue_benchmark(500)
for count in (1_000, 1_500, 2_000):
comparison = list_overdue_benchmark(count)
print_delta(baseline, comparison)
# >>> 원소수 500, 걸린 시간 0.000844초
# >>> 데이터 크기 2배, 걸린 시간 3.1배
# >>> 데이터 크기 3배, 걸린 시간 6.1배
# >>> 데이터 크기 4배, 걸린 시간 10.3배
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
|
from heapq import heappush
def add_book(queue, book):
heappush(queue, book) # 정렬 가능한 개체만 가능
# 참고1: chapter.51 합성 가능한 클래스 확장이 필요할 경우 클래스 데코레이터를 활용하라
# 참고2: chapter.41 커스텀 컨테이너 타입은 collections.abc 를 상속하라
import functools
@functools.total_ordering
class Book:
def __init__(self, title, due_date):
self.title = title
self.due_date = due_date
# for 정렬
def __lt__(self, other):
return self.due_date < other.due_date
queue = []
add_book(queue, Book('돈키호테', '2020-06-07'))
add_book(queue, Book('프랑켄슈타인', '2020-06-05'))
# 직접 정렬하거나
queue.sort()
# heapq.heapify 함수를 이용하면 선형시간에 힙을 만들 수 있다
from heapq import heapify
heapify(queue)
# 원소 추가시 heapq.heappush 함수를 사용 (list.push 대신에)
# 원소 삭제시 heapq.heappop 함수를 사용 (list.pop 대신에)
from heapq import heappush, heappop
def run(queue, to_add):
for i in to_add:
heappush(queue, i)
while queue:
heappop(queue)
def next_overdue_book(queue, now):
if queue:
book = queue[0]
if book.due_date < now:
heappop(queue)
return book
raise NoOverdueBooks
# >>> 원소수 500, 걸린 시간 0.000116초
# >>> 데이터 크기 2배, 걸린 시간 2.1배
# >>> 데이터 크기 3배, 걸린 시간 3.2배
# >>> 데이터 크기 4배, 걸린 시간 4.4배
|
제시간에 반납된 책은 어떻게 처리해야 할까?
==> 만기일까지 우선순위 큐에서 책을 절대 제거하지 않는 것이다. (상태변수 필요)
==> 이 접근 방법을 택하면 책을 반납할 때 우선순위 큐를 변경할 필요가 없어진다
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
|
import functools
@functools.total_ordering
class Book:
def __init__(self, title, due_date):
self.title = title
self.due_date = due_date
self.returned = False # 상태변수
def next_overdue_book(queue, now):
while queue:
book = queue[0]
if book.returned: # 기존 흐름에 자연스럽게 섞어 넣는다
heappop(queue) # 정상 반납된 것이면 다음 항목을 검사 (while)
continue
if book.due_date < now:
heappop(queue)
return book
break
raise NoOverdueBooks
# 반납처리가 단순해졌다
def return_book(queue, book):
book.returned = True
|
스레드 안전한 다른 선택이 필요하다면 queue.PriorityQueue 클래스를 보라.
74) bytes를 복사하지 않고 다루려면 memoryview와 bytearray를 사용하라
스루풋이 높은 병렬 I/O를 다양한 방식으로 지원할 수 있다.
-
memoryview
는 CPython의 고성능 버퍼 프로토콜을 프로그램에 노출시켜준다. (저수준 C API)
- 복사가 없는(zero-copy) 연산을 활성화 함으로써 Numpy 같은 수치계산 C 확장이나 I/O 위주 프로그램의 성능을 엄청나게 향상시킬 수 있다.
- 수 GB의 미디어 파일의 경우 2만배 이상의 속도 향상 (단순 메모리 공유로는 클라이언트 17대 붙이기도 어렵다)
1
2
3
4
5
6
|
data = '동해물과 백두산이 abc 마르고 닳도록'.encode('utf8')
view = memoryview(data)
chunk = view[12:19] # 슬라이싱
print('크기:', chunk.nbytes)
print('뷰의 데이터:', chunk.tobytes)
print('내부 데이터:', chunk.obj)
|
-
bytearray
타입은 bytes 에서 원하는 위치의 값을 바꿀 수 있는 가변 버전과 같다
-
bytearray
타입은 복사가 없는 읽기 함수(socket.recv_from 같은)에 사용할 수 있는 bytes 의 변경 가능 타입을 제공한다.
1
2
3
|
my_array = bytearray('hello 안녕'.encode('utf8'))
my_array[0] = 0x79
print('변경됨:', my_array) # yello ~
|
- memoryview 로 bytearray 를 감싸면 복사에 따른 비용을 추가 부담하지 않고도 수신받은 데이터를 버퍼에서 원하는 위치에 스플라이스 할 수 있다.