사용자 정의 예외 실습(비밀번호/은행/도서관), 상속과 다형성, 주차장 관리, 제너레이터 연습,
asyncio비동기,threading,sys.argv.
Day.6에서 예외 처리와 사용자 정의 예외를, Day.7에서 이터레이터와 제너레이터를 배웠다. 오늘은 앞부분에서 그 내용을 문제 풀이로 다시 써 보고, 뒷부분에서 비동기 프로그래밍(asyncio)과 스레드를 새로 배웠다.
조건은 두 가지다. 4자 이상이어야 하고, 숫자가 최소 1개는 있어야 한다. 조건마다 예외 클래스를 따로 만들었다.
class ShortPassword(Exception):
def __init__(self):
super().__init__("비밀번호는 4자 이상이어야 합니다")
class NoDigit(Exception):
def __init__(self):
super().__init__('비밀번호에 숫자가 최소 1개는 포함되어야 합니다.')
def check_pw(pw):
try:
if len(pw) < 4:
raise ShortPassword()
if not any(i.isdigit() for i in pw): # 숫자가 하나도 없으면
raise NoDigit()
except ShortPassword as e:
print(e)
except NoDigit as e:
print(e)
else:
print('사용 가능')
check_pw('ab1') # 비밀번호는 4자 이상이어야 합니다
check_pw('abcd') # 비밀번호에 숫자가 최소 1개는 포함되어야 합니다.
check_pw('abcd1') # 사용 가능
any(i.isdigit() for i in pw)는 Day.5의 any에 제너레이터 표현식(Day.7)을 넣은 것이다. 문자열 pw의 글자를 하나씩 꺼내 isdigit()(숫자 문자인지 True/False)로 검사하고, 하나라도 True면 any가 True를 반환한다.raise하면 그 아래 검사는 실행되지 않는다. 그래서 ab1은 숫자가 있어도 길이 에러만 나온다.input()으로 비밀번호를 받았는데, 테스트하기 쉽게 인자로 받도록 바꿨다.except를 예외마다 나눠 두었기 때문에 예외 종류별로 다른 처리를 할 수 있다.class BankError(Exception): # 은행 관련 예외의 공통 부모
pass
class InvalidAmountError(BankError): # 잘못된 금액
pass
class InsufficientFundsError(BankError): # 잔액 부족
pass
class Bank:
def __init__(self, balance=0):
self.balance = balance
def withdraw(self, amount):
if amount <= 0:
raise InvalidAmountError('출금액은 0보다 커야 한다')
if amount > self.balance:
raise InsufficientFundsError('잔액이 부족하다')
self.balance -= amount # balance: 잔고, amount: 출금액
return self.balance
acc = Bank(10000)
for amt in [0, 20000, 3000]:
try:
print(acc.withdraw(amt))
except InsufficientFundsError as e:
print('부족:', e)
except InvalidAmountError as e:
print('잘못:', e)
# 잘못: 출금액은 0보다 커야 한다
# 부족: 잔액이 부족하다
# 7000
class 예외이름(Exception): pass처럼 본문 없이 이름만 만들어도 사용자 정의 예외가 된다.raise InvalidAmountError('메시지')처럼 메시지를 인자로 넘기면 print(e)로 그 메시지가 나온다. Day.6에서 __str__을 오버라이딩하면 메시지가 무시되던 것과 달리, 여기서는 Exception의 기본 동작이 인자를 그대로 보관해서 출력한다.필기에서는 BankErro(오타)를 만들어 놓고 실제로는 쓰지 않았고, 두 예외가 모두 Exception을 직접 상속했다. 위처럼 공통 부모(BankError)를 만들어 상속하면 예외를 묶어서 처리할 수 있다.
try:
acc.withdraw(-1)
except BankError as e: # 자식 예외를 모두 잡는다
print('부모로 한 번에:', e) # 부모로 한 번에: 출금액은 0보다 커야 한다
필기의
acc.withdraw(000)은000이 그냥0으로 읽힌다. (파이썬 3에서000은 0이다.0으로 시작하는 다른 정수는 문법 에러라서010은 못 쓴다.)
책 객체와 도서관 객체를 만들고, 이미 대출 중인 책을 또 빌리면 예외를 낸다.
class AlreadyError(Exception):
pass
class Book:
def __init__(self, num, title, author):
self.num = num
self.title = title
self.author = author
self.borrowed = False # 처음에는 대출 중이 아님
class Library:
def __init__(self):
self.books = []
def add_book(self, book):
self.books.append(book)
def find_book(self, num):
for i in self.books:
if i.num == num:
return i
return None # 없으면 None
def borrow(self, num):
book = self.find_book(num)
if book is None:
print('해당 책은 도서관에 없습니다')
return
if book.borrowed:
raise AlreadyError('이 책은 이미 대출중입니다.')
book.borrowed = True
b1 = Book(1, 'Python 입문', 'Kim')
b2 = Book(2, '알고리즘', 'Lee')
lib = Library()
lib.add_book(b1)
lib.add_book(b2)
lib.borrow(1)
print(b1.borrowed) # True
try:
lib.borrow(1)
except AlreadyError as e:
print('대출 실패', e) # 대출 실패 이 책은 이미 대출중입니다.
lib.borrow(9) # 해당 책은 도서관에 없습니다
Day.7의 회원 관리(find_user)와 같은 구조다. 찾는 함수(find_book)가 None을 반환하고, 호출한 쪽(borrow)에서 is None으로 분기한다. 없는 책은 출력하고 끝내고, 이미 대출 중인 책은 예외를 던진다. 문제의 성격에 따라 둘을 다르게 처리했다.
class Employee:
def __init__(self, name, salary):
self.name = name
self.salary = salary
def pay(self):
return self.salary
class Manager(Employee):
def pay(self): # 오버라이딩: 기본급의 1.5배
return self.salary * 1.5
class Developer(Employee):
def __init__(self, name, salary, overtime=0):
super().__init__(name, salary) # 공통 속성은 부모에게
self.overtime = overtime # 자식만의 속성
def pay(self): # 오버라이딩: 기본급 + 야근 수당
return self.salary + self.overtime * 30000
emp = [Employee('Kim', 3000000), Manager('Lee', 4000000), Developer('Park', 5000000, overtime=10)]
for i in emp:
print(f'{i.name}: 기본급 {i.salary}원 → 지급액 {i.pay():,.0f}원')
# Kim: 기본급 3000000원 → 지급액 3,000,000원
# Lee: 기본급 4000000원 → 지급액 6,000,000원
# Park: 기본급 5000000원 → 지급액 5,300,000원
Day.5의 상속, 오버라이딩, super()를 한꺼번에 쓰는 문제다. 핵심은 마지막 반복문이다.
emp 리스트에는 Employee, Manager, Developer 세 종류의 객체가 섞여 있다.i.pay()라고 똑같이 호출하는데, 객체의 클래스에 따라 각자의 pay()가 실행된다. 이것을 다형성이라고 한다.if로 따질 필요가 없다. 새 직원 종류(Intern 등)를 추가해도 반복문은 그대로다.필기의 출력문은
i.salary(기본급)만 찍고 있었고 단위도 "만원"이었다. 그런데 값이3000000이라 단위는 "원"이 맞고, 클래스마다 다르게 정의한pay()를 호출해야 다형성이 드러난다. 그래서i.pay()를 출력하도록 고쳤다. 참고로Manager.pay()는* 1.5라서 실수(6000000.0)가 나온다.
차량 번호가 중복되면 예외를 내는 주차장 클래스다.
class AlreadyError(Exception):
pass
class Vehicle:
def __init__(self, number, kind):
self.number = number
self.kind = kind
class Parking:
def __init__(self):
self.cars = []
def add(self, vehicle):
# 이미 같은 번호가 있으면 예외
if any(i.number == vehicle.number for i in self.cars):
raise AlreadyError('이미 주차되어있는 차량입니다.')
self.cars.append(vehicle)
def list_car(self):
return [i.number for i in self.cars if i.kind == 'car'] # 승용차 번호만
lot = Parking()
ve1 = Vehicle('12가1234', 'car')
ve2 = Vehicle('23나1205', 'bike')
ve3 = Vehicle('45다9011', 'car')
lot.add(ve1)
lot.add(ve2)
lot.add(ve3)
try:
lot.add(ve1) # 같은 번호를 다시 추가
except AlreadyError as e:
print(e) # 이미 주차되어있는 차량입니다.
print(lot.list_car()) # ['12가1234', '45다9011']
처음에는 for와 if로 번호를 하나씩 비교했는데, any(... for ...)로 줄였다.
# 처음 방식
for i in self.cars:
if i.number == vehicle.number:
raise AlreadyError('이미 주차되어있는 차량입니다.')
# any 방식
if any(i.number == vehicle.number for i in self.cars):
raise AlreadyError('이미 주차되어있는 차량입니다.')
list_car도 문자열을 이어 붙이는 방식에서 리스트 컴프리헨션으로 바꿨다. 필기에 남아 있던 주석 코드는 return이 반복문 안에 있어서 첫 번째 승용차에서 바로 함수가 끝나는 문제가 있었다. return은 값을 반환하면서 함수를 끝내기 때문이다. 컴프리헨션은 이런 실수 없이 조건에 맞는 것만 모은다.
def total():
sum = 0
while True:
value = sum
yield value
sum += value
gen = total()
print(next(gen)) # 0
print(next(gen)) # 0
print(next(gen)) # 0
sum += value에서 value는 방금 sum을 복사한 값이다. 즉 자기 자신을 더하고 있어서 0 + 0 = 0이 되어 계속 0이다. 값을 더하려면 바깥에서 새 값을 받아야 한다. 제너레이터에는 이럴 때 쓰는 send()가 있다.
def running_total():
s = 0
while True:
n = yield s # yield가 s를 내보내고, send()로 받은 값이 n에 들어온다
if n is not None:
s += n
rt = running_total()
print(next(rt)) # 0 (첫 실행은 next로 시작한다)
print(rt.send(10)) # 10
print(rt.send(5)) # 15
print(rt.send(20)) # 35
send(값)은 멈춰 있던 yield 자리에 값을 넣고 다음 yield까지 실행한다. (next(g)는 send(None)과 같다.) 변수 이름을 sum으로 쓰면 내장 함수 sum을 가린다는 점도 다시 확인했다.
class Total:
def __init__(self):
self.num = 0
def __iter__(self):
return self
def __next__(self):
return self.num # 현재 값을 돌려준다 (값을 바꾸지 않는다)
def add(self, number):
self.num += number
return self.num
t = Total()
print(next(t)) # 0
print(t.add(10)) # 10
print(t.add(5)) # 15
print(t.add(20)) # 35
print(next(t)) # 35 (현재 값 그대로)
__next__가 값을 바꾸지 않고 현재 값만 돌려주기 때문에, 이 객체를 for에 넣으면 같은 값이 영원히 나온다. 값을 변경하는 일은 add가 맡는다. 제너레이터 버전과 같은 동작을 클래스로는 상태를 self.num에 두고 메소드로 바꾸는 식으로 구현한 것이다.
기다리는 시간이 많은 작업에서 쓴다.
이런 작업은 CPU가 일하는 시간보다 응답을 기다리는 시간이 대부분이다. 기다리는 동안 다른 일을 처리하면 여러 요청을 동시에 처리할 수 있다. (JavaScript의 async/await, Promise와 같은 배경이다. JS Day.9, Day.10 참고)
# 일반 함수: 호출하면 바로 실행되고, 끝날 때까지 멈추지 않는다
def func():
print('hi')
return 1
result = func()
print(result) # hi / 1
# 코루틴 함수: async def로 정의
import asyncio
async def main():
print('hello...')
await asyncio.sleep(1) # 1초 동안 "다른 일을 해도 된다"고 양보
print('....월드!')
asyncio.run(main())
# hello...
# (1초 뒤) ....월드!
| 문법 | 의미 |
|---|---|
async def | 코루틴 함수를 만든다 |
await | 다른 비동기 작업이 끝날 때까지 기다린다. async 함수 안에서만 쓸 수 있다 |
asyncio.run(코루틴) | 이벤트 루프를 만들어 코루틴을 실행한다. 프로그램의 시작점에서 한 번 호출한다 |
asyncio.sleep(초) | 비동기용 대기. time.sleep과 달리 기다리는 동안 다른 작업이 실행될 수 있다 |
코루틴 함수는 호출만 해서는 실행되지 않는다. 호출하면 코루틴 객체가 만들어질 뿐이고, await하거나 asyncio.run()에 넘겨야 실행된다.
c = show('x', 1)
print(type(c)) # <class 'coroutine'> (아직 아무것도 실행되지 않았다)
JS의 async 함수가 Promise를 반환하는 것과 비슷한 구조다.
await로 다른 코루틴을 호출하는 것도 된다.
async def say_hello():
print("안녕!")
await asyncio.sleep(2)
print("2초 뒤에 또 안녕!")
async def main():
await say_hello() # 비동기 함수 안에서 다른 비동기 함수를 호출
asyncio.run(main())
asyncio.gatherimport asyncio
async def show(name, sec):
print('어서오세요')
await asyncio.sleep(sec) # 기다리는 동안 다른 작업 허용
print(f'{name}님이 기다린 시간은 {sec}초입니다.')
asyncio.run(show('Tom', 3)) # 하나만 실행: 3초
async def main():
# 두 코루틴을 동시에 실행하고, 둘 다 끝날 때까지 기다린다
await asyncio.gather(
show('python', 2),
show('javascript', 3)
)
asyncio.run(main())
# 어서오세요
# 어서오세요 ← 둘이 거의 동시에 시작
# python님이 기다린 시간은 2초입니다. (2초 뒤)
# javascript님이 기다린 시간은 3초입니다. (3초 뒤)
흐름은 이렇다.
main()이 실행되고, gather가 show 두 개를 동시에 시작한다.어서오세요가 거의 동시에 출력된다.python이, 3초 뒤 javascript가 끝난다.gather가 끝나고 main()이 종료된다.전체 시간이 가장 오래 걸리는 작업 기준(3초)이다. 순서대로 await 했을 때와 비교해 직접 재 보았다.
async def seq():
await show('python', 2) # 2초 끝난 뒤에
await show('javascript', 3) # 3초 시작
# gather: 3.0초
# 순차 await: 5.0초
await를 나란히 두면 앞의 것이 끝나야 다음이 시작하기 때문에 합계(5초)가 걸린다. 동시에 돌리려면 gather로 묶어야 한다.
필기에는 음식점 알바에 빗대어 정리해 두었다. 주문, 요리, 서빙을 한 명이 순서대로 처리하면 줄 서서 기다려야 한다.
| 구분 | 스레드 | 비동기 |
|---|---|---|
| 비유 | 직원을 여러 명 고용 | 직원 1명이 요리하는 동안 다른 일도 처리 |
| 동작 | 여러 작업을 동시에 실행 | 기다리는 도중에 다른 작업 처리 |
| 실행 단위 | OS가 스레드를 스케줄링 | 이벤트 루프가 코루틴을 번갈아 실행 |
필기에는 스레드는 "CPU를 많이 쓰고 독립적인 작업", 비동기는 "기다리는 시간이 많은 작업"이라고 적혀 있다. 여기에는 보완이 필요하다. 표준 CPython에는 GIL이 있어서 한 시점에 한 스레드만 파이썬 코드를 실행한다. 그래서 threading은 CPU 연산을 빠르게 해 주지 못하고(그럴 때는 multiprocessing을 쓴다), 스레드도 기다림이 많은 I/O 작업에서 효과가 있다. 정리하면 이렇다.
multiprocessingthreading.Threadimport threading
import time
def game(name, sec):
print(f'{name}입장')
time.sleep(sec)
print(f'{name}퇴장')
# 스레드 객체 생성: Thread는 threading 모듈에 정의된 클래스
t1 = threading.Thread(target=game, args=('음악', 2))
t2 = threading.Thread(target=game, args=('채팅', 5))
t1.start() # target으로 지정한 함수를 새 스레드에서 실행
t2.start()
t1.join() # t1이 끝날 때까지 기다림
t2.join()
print('끝!')
target에 실행할 함수를, args에 인자를 튜플로 넘긴다. 인자가 하나일 때는 args=(i,)처럼 쉼표를 꼭 붙여야 튜플이 된다. (Day.2의 튜플)start()를 하면 두 스레드가 동시에 돌아가서 전체 시간은 5초(둘 중 긴 쪽)다.join()은 해당 스레드가 끝날 때까지 메인 프로그램이 기다리게 한다. 스레드는 독립적으로 실행되기 때문에 join()이 없으면 스레드가 끝나기 전에 메인이 먼저 끝!을 출력하고 지나갈 수 있다.start()를 먼저 호출했다고 먼저 실행된다는 보장은 없다. 어떤 스레드가 언제 실행될지는 OS 스케줄러가 결정한다.def game(name):
print(f'{name} 입장')
time.sleep(1)
print(f'{name} 퇴장')
threads = []
for n in ['A', 'B', 'C']:
t = threading.Thread(target=game, args=(n,))
t.start()
threads.append(t) # 나중에 join하려고 리스트에 보관
for i in threads:
i.join() # 모든 스레드가 끝날 때까지 대기
print('끝! (모든 스레드 종료 후)')
실행 결과(전체 약 1초)는 이렇게 섞여서 나온다.
A 입장
B 입장C 입장
A 퇴장
C 퇴장B 퇴장
끝! (모든 스레드 종료 후)
세 스레드가 동시에 print를 하다 보니 줄이 섞이고, 실행할 때마다 입장/퇴장 순서가 달라질 수 있다. 필기에 "스레드는 결과 값이 달라짐"이라고 적어 둔 부분이다.
필기에서는 마지막 반복문이 이렇게 되어 있었다.
for i in threads:
print(t)
t.join() # ← i가 아니라 t
반복 변수는 i인데 t.join()이라고 써서, 반복문 끝에 남아 있는 마지막 스레드 t만 세 번 join 했다. 그러면 나머지 스레드는 기다리지 않는다. 예를 들어 마지막 스레드가 첫 번째 스레드보다 먼저 끝나는 경우 끝!이 먼저 출력될 수 있다. 같은 시간(1초)씩 쉬는 예제에서는 결과가 비슷하게 나와서 눈치채기 어렵다. i.join()으로 쓰면 모든 스레드를 기다린다.
필기 예제의 이름은 인물 이름 세 개였는데
A,B,C로 바꿨다.
sys.argv: 실행할 때 값 넘기기import sys
print(sys.argv)
args = sys.argv[1:]
for i in args:
print(i)
터미널에서 파일 이름 뒤에 값을 붙여 실행하면 그 값들이 리스트로 들어온다.
$ python f.py apple 3 "hello world"
['f.py', 'apple', '3', 'hello world']
apple
3
hello world
sys.argv[0]은 실행한 파일 이름이고, sys.argv[1:]부터가 사용자가 넘긴 값이다. 그래서 [1:]로 슬라이싱했다. (Day.1의 슬라이싱)3도 '3'으로 들어오므로 숫자로 쓰려면 int()가 필요하다.sys.exit(), sys.path와 같은 sys 모듈의 기능이다.class 이름(Exception): pass만으로도 만들 수 있다. 공통 부모 예외를 두고 상속하면 except 부모:로 한꺼번에 처리할 수 있다.pay())를 호출해도 객체의 클래스에 따라 각자의 동작이 실행된다.any(조건 for ... in ...)와 리스트 컴프리헨션으로 for/if 반복을 짧게 줄일 수 있다.yield로 값을 내보내고, send()로 값을 받을 수 있다.async def로 코루틴을 만들고, await로 기다리고, asyncio.run으로 시작한다. asyncio.gather로 묶으면 동시에 실행되어 전체 시간이 가장 긴 작업 기준이 된다.threading.Thread(target, args) → start() → join(). 실행 순서는 OS가 정하고, join은 모든 스레드에 해 줘야 한다. 파이썬에서 CPU 연산을 병렬화하려면 스레드가 아니라 프로세스를 쓴다.sys.argv로 실행 시점에 값을 받을 수 있고, 항상 문자열이다.Tags: Python 예외처리 다형성 상속 제너레이터 asyncio 비동기 코루틴 threading sys.argv 개발자 학습기록