Python built-in library

newhyork·2022년 7월 17일

Argparse Dataclasses logging subprocess threading

logging

프로세스가 실행되면서 발생하는 다양한 이벤트를 추적하는 행위이다.
- 에러 발생 여부를 비롯해 다양한 로그를 기록하며,
  전반적으로 시스템을 유지 보수하는 데에 도움을 줄 목적으로 쓰인다.
level(심각도)
-- DEBUG: 상세한 정보. 문제의 원인을 파악하기 위한 로그.
-- INFO: 일반적인 정보. 의도한 대로 동작하는 지 추적하기 위한 로그.
-- WARNING: 예상치 못한 혹은 가까운 미래에, 문제가 될 수 있는 것에 대한 로그.
-- ERROR: 에러 로그. 심각한 문제로 인해, 의도한 대로 동작하지 않음을 보여주는 로그.
-- CRITICAL: 프로세스 자체가 중단될 수 있는 문제를 보여주는 로그.
- 기본 값인 WARNING으로 해 놓으면, 이보다 하위인 DEBUG와 INFO는 출력하지 않는다.
Python에서는, logging이라는 built-in library를 주로 사용한다.
- Flask에서는, Flask app객체에 logger라는 private variable이 있다.
  (Flask는 고성능을 위해, @locked_cached_property를 통해 logger값을 caching한다.)
  - 이는 결국 Python logging library의 logger객체를 return하는 것으로,
    이를 통해서도 level에 따른 logging을 할 수 있다.
  - Flask app객체의 errorhandler() 메서드를 통해,
    Flask app서버에서 발생하는 각종 HTTP error code 혹은 exception을
    쉽게 handling할 수도 있다.
일반적으로 다양한 formatter를 통해, 시각이나 에러 양식 등을 용도에 맞게 사용하곤 한다.
- 공통된 포맷을 사용하는 logging 모듈을 작성해 놓고,
  logging이 필요한 곳에서 import해서 logger 객체를 가져다 쓰는 방식으로 사용할 수 있다.
보통, logger 객체를 생성하는 클래스를 작성하고
이를 상속하여 각 용도 및 목적에 맞는 logger 객체를 생성하는 패턴으로 사용하곤 한다.

usage

# simple_logging.py
import logging


logging.basicConfig()
- 기본 handler를 지정해준다. 

logger = logging.getLogger()
- 아무런 logger를 지정해주지 않으면, rootlogger이다. 

logger.setLevel(logging.INFO)
- 기본값은 WARNING이다. 
  INFO으로 설정했으므로, DEBUG는 출력되지 않는다.

logger.debug('DEBUG MSG')
logger.info('INFO MSG')
logger.critical('CRITICAL MSG')


# output
INFO:root:INFO MSG
CRITICAL:root:CRITICAL MSG

# logging_with_handler.py
import logging


logger = logging.getLogger('simple_logger')
logger.setLevel(logging.DEBUG)
- 'simple_logger'에 대한 기본 세팅이다.

formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')

stream_handler = logging.StreamHandler()
stream_handler.setLevel(logging.INFO)
stream_handler.setFormatter(formatter)
logger.addHandler(stream_handler)
- 터미널에 로그를 출력한다.

file_handler = logging.FileHandler('log.log', mode='w')
file_handler.setLevel(logging.WARNING)
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
- 파일에 로그를 기록한다.
- 만약 여기서, setLevel()을 설정해주지 않으면
  log.log에서는 'simple_logger' 기본 세팅에 따라, DEBUG log까지 기록된다.

if __name__ == "__main__":
    logger.debug('DEBUG MSG')
    logger.info('INFO MSG')
    logger.error('ERROR MSG')
    logger.critical('CRITICAL MSG')


# output
2022-08-18 22:04:06,312 - simple_logger - INFO - INFO MSG
2022-08-18 22:04:06,313 - simple_logger - ERROR - ERROR MSG
2022-08-18 22:04:06,313 - simple_logger - CRITICAL - CRITICAL MSG


# log.log
2022-08-18 22:04:06,313 - simple_logger - ERROR - ERROR MSG
2022-08-18 22:04:06,313 - simple_logger - CRITICAL - CRITICAL MSG


-------------------------------------------------------------------------------


## logging_needed.py
from logging_with_handler import logger


logger.debug('debug msg')
logger.info('info msg')
logger.error('error msg')
logger.critical('critical msg')


## output
2022-08-18 22:05:19,743 - simple_logger - INFO - info msg
2022-08-18 22:05:19,744 - simple_logger - ERROR - error msg
2022-08-18 22:05:19,744 - simple_logger - CRITICAL - critical msg


## log.log
2022-08-18 22:05:19,744 - simple_logger - ERROR - error msg
2022-08-18 22:05:19,744 - simple_logger - CRITICAL - critical msg

# flask_logger.py
from flask import Flask


app = Flask(__name__)

@app.errorhandler(404)
def get_page_not_found(error):
    app.logger.error(error)
    return 'page_not_found.html'

app.run()


# enter URL: http://localhost:5000 (GET '/')
web page shows: page_not_found.html
terminal log:   ERROR in flask_logger: 404 Not Found: The requested URL was not found..

argparse

CLI에서 Python 모듈 실행 시, 옵션처럼 전달 받은 인자를 쉽게 이용할 수 있게 해준다.
옵션 문자열을 사용하기 때문에, sys.argv보다 좀 더 명시적으로 인자를 사용할 수 있다.
이 외에도, 인자와 관련된 다양한 기능을 지원한다.

usage

# prac_argparse.py 
import argparse

parser = argparse.ArgumentParser()

parser.add_argument("--pork", dest="food", action="store_true")
parser.add_argument("--beef", dest="food", action="store_true")
parser.add_argument("--number", dest='number', required=True)
- 이 모듈을 실행할 때 전달받은 인자를 사용하고자 할 때, add_argument()를 통해 이용할 수 있다.
  - 첫 번째 매개변수는 문자열로, 인자로써 전달받은 옵션의 이름이다.
  - dest는, 해당 인자의 값을 참조할 때 사용할 클래스 변수명이다.
  - action은, 해당 인자를 전달받았을 때 기본적으로 수행할 것을 나타낸다. 
    - flag로 쓸 인자는 value가 딱히 필요없기에, 주로 "store_true"로 두어 사용하곤 한다.
    - 기본값은 "store"이므로, 단순히 전달받은 인자의 값을 저장한다.
  - required는 필수 인자에 대해 사용한다.
    - 기본값은 False이다. 

args = parser.parse_args()
- 모듈을 실행할 때 전달받은 인자는, add_argument를 거치고
  parse_args()에 의해 Namespace 객체로 만들어져, 비로소 사용된다.
  - add_argument에 따라 각 인자마다 쌍을 이룬 형태인데,
    변수명은 dest값, value는 해당 인자의 값이다.

print(args.food)
print(args.number)
- args.{dest값}으로, 실행 시 전달받은 인자의 값을 참조할 수 있다. 


# CLI
$ python prac_argparse.py --pork --number 6


# output
True
6

threading

하나의 프로세스에서 2가지 이상의 일을 동시에 수행할 수 있도록,
여러 thread를 생성하여 실행할 수 있게 해준다.

usage

# task.py
import time

def task():
    for i in range(1, 6):
        time.sleep(1)
        print(f'{i}번째')

print('start')

for i in range(1, 6):
    task()

print('end')
'''
task.py의 task()는 5초 동안 실행되므로, 이를 5번 실행하여 총 25초가 걸린다. 
threading을 이용하여 이를 5초로 줄일 수 있다.
Thread 객체 5개를 만들어, 동시에 task()를 실행하는 것이다.
'''


# thread_task.py
import time
import threading

def thread_task():
    for i in range(1, 6):
        time.sleep(1)
        print(f'{i}번째')

print('create threads')

threads = []
for i in range(5):
    t = threading.Thread(target=thread_task)
    threads.append(t)
- thread_task 함수를 실행하는 Thread 객체를 만들어 리스트에 넣는다. 
  - 인자가 필요하면, args= 로 전달할 수 있다. 
- 이 때, Thread 객체를 print 해보면 initial 상태이다. 

print('run threads')

for t in threads:
    t.start()
- start로, 각각의 Thread를 실행한다. 
  즉, Thread에서는 target 함수를 실행한다. 
- 이 때, Thread 객체를 print 해보면 started 상태이다. 

for t in threads:
    t.join()
- join으로, 각 thread가 종료될 때까지 이후의 코드 블럭을 실행하지 않도록 한다.
  즉, 코드 블럭을 실행(제어)하는 부모 스레드가
  새로 생성한 Thread 객체인 자식 스레드의 종료를 기다린다.
  - 만약 이를 넣지 않으면, 프로그램이 정상 종료되지 않는다거나
    이후의 'end threads'가 의도한 대로 맨 마지막에 출력되지 않는다.
- 이 때, Thread 객체를 print 해보면 stopped 상태이다. 

print('end threads')

subprocess

현재 프로세스에서 자식 프로세스로써 shell 명령어를 실행할 수 있게 해준다.
- shell 명령어를 비롯해, AWS CLI 명령어나 파일 실행에 주로 쓰이곤 한다.
os.system보다, 관련하여 더 다양한 기능을 지원한다.

run

함수로써 자식 프로세스를 실행하는 것이다.
해당 프로세스의 stdout 이용하려면, capture_output=True를 인자로 전달한다.
- 해당 자식 프로세스가 종료된 시점에 모든 출력을 받아와, 사용할 수 있다.

usage

# prac_subprocess_run.py
import subprocess 

cmd = 'df -h'
sp = subprocess.run(cmd, capture_output=True, encoding='utf-8')
print(sp.stdout)


# output
Filesystem            Size  Used Avail Use% Mounted on
C:/Program Files/Git  459G  141G  319G  31% /

Popen

객체로써 자식 프로세스를 실행하는 것이다.
해당 프로세스의 stdout을 이용하려면, stdout=subprocess.PIPE를 인자로 전달한다.
- stderr 이용 시에는, stderr=subprocess.PIPE를 인자로 전달한다.
Popen 객체의 communicate() 메서드가 output과 error를 반환한다.
- 해당 자식 프로세스가 종료될 때까지 대기하므로, 메인 프로세스 코드 실행은 block된다.

usage

# prac_subprocess.py
import time
import sys

start = time.time()
print(f'{sys.argv[1]} process start')
time.sleep(3)
print(f'{sys.argv[1]} process end')
end = time.time()
print(end-start)


# prac_subprocess_Popen.py
import subprocess
import time

start = time.time()
sps = []
for i in range(1, 10+1):
    cmd = ['python', 'prac_subprocess.py', str(i)]
    sp = subprocess.Popen(cmd, encoding='utf-8',
                        stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    sps.append(sp)
    middle = time.time()
    print(f'middle {i} {middle-start}')
for _sp in sps:
    stdout, stderr = _sp.communicate()
    print(stdout, stderr)
end = time.time()
print(end-start)


# output 
middle 1 0.004940986633300781
middle 2 0.012955427169799805
... (점점 커짐)
middle 10 0.08871936798095703
1 process start
1 process end
3.008728265762329
 
2 process start
2 process end
3.0149028301239014

... (무작위)

10 process start
10 process end
3.0109472274780273

3.1385276317596436
'''
메인 프로세스가 실행되고 종료되기까지 약 3.14초가 걸렸다. 
각 자식 프로세스는 Popen()에 의해 실행되어 3초간 실행하도록 되어 있다. 
(따라서, 메인 프로세스를 실행한 후 output이 나오기 시작하기까지 약 3초가 걸린다)
각 자식 프로세스는 먼저 실행된 것의 종료를 기다리는 것이 아니고, 병렬로 실행된다. 
communicate()로 각 자식 프로세스의 출력값을 받아온다.
메인 프로세스가 약 0.14초 더 걸린 이유는 여러 가지가 있겠지만
그 중 하나는, 메인 프로세스에서 각 자식 프로세스를 실행하기까지의 시간이 걸린다는 것이다.
또한, 2번째 for문의 communicate()에 의해
먼저 종료되어 출력값을 반환해야할 자식 프로세스가 아직 끝나지 않았을 때,
이를 기다리기 때문이라고 생각할 수 있다. 
'''

dataclasses

데이터에 대한 클래스를 편리하게 정의하고 사용할 수 있게 해주는 모듈이다.
동적 언어인 Python 을 보완할 수 있는 점으로써, type 을 선언한다는 특징도 있다.

feature

일반 클래스로 정의했을 때와 다음과 같은 차이가 있다.
- 객체 속성이 자동으로 정의되어 있다.
  - __init__(self) 을 따로 써줄 필요가 없다.
- 객체를 print() 시, 해당 객체의 모든 속성을 보여준다.
  - 원래는 객체가 저장된 메모리 주소를 표시한다.
  - __repr__(self) (내지는 __str__(self)) 을 별도로 작성해줘야 하지만,
    @dataclass를 쓰면 이 점이 기본적으로 자동화되어 있어 편리하다.
  - 보여주고 싶지 않은 속성은, field(repr=False) 로 둔다.
- 객체 간에 equal(==) 비교 시, 속성 값이 모두 같다면 True 로 반환해준다.
  - 원래는 객체 메모리 주소를 비교하기 때문에 False 이다.
  - __eq__(self, other) 를 별도로 작성할 필요가 없다.
@dataclass 추가 옵션
- frozen=True: 객체 속성 값을 바꿀 수 없게 할 때
- order=True: 객체 간에 대소 비교 내지는 sort 를 할 때
  - 비교에 사용하지 않을 속성은 field(compare=False) 으로 둔다.
- unsafe_hash=True: 객체가 hashable 한 특성을 가지게 할 때
  - hashable: set 의 값 (주로, 중복 제거) 혹은 dictionary 의 key 로써 사용 가능하게 한다.
- kw_only=True: 객체 생성 시, 인수로 positional 이 아닌 keyword argument 만 허용할 때
객체 속성에 기본 값을 줄 수도 있다.
- 단, list 와 같은 mutable data type 에는 field(default_factory={type}) 를 사용해야 한다.
  - mutable data type 에 기본 값을 할당하면, 인스턴스 간에 공유가 되기 때문에 불허한다.
  - list 의 경우, 빈 리스트가 기본 값이다.
- 기본 값으로, 따로 정의되어 있는 함수 객체를 이용할 수도 있다.
  - field(default_factory={func}): 해당 함수의 return 값
  - field(default={func}): 해당 함수 객체 그 자체
    - 함수 따위가 아닌 그냥 임의의 값으로 두고자 한다면,
    {func} 대신 값을 넣으면 된다.
    (이럴 땐, 굳이 field()를 쓰지 않고 = 값으로 둬도 된다.)
객체 속성 값으로, 다른 속성 값을 이용할 수도 있다.
- 우선, 대상 속성을 field(init=False) 로 둔다.
  - 객체 생성 시, 해당 속성은 따로 인자를 전달하여 초기화하지 않도록 한다는 의미이다.
- __post_init__(self) 메서드에서 해당 속성을 정의한다.
  - __init__(self) 이후에 실행된다.
    (물론, 앞에서 언급했지만, 이 메서드 정의는 생략된다.)
객체를 tuple 혹은 dictionary 로 쉽게 변환할 수 있다.
- astuple()
- asdict()
fields(dataclass_object)
- 객체의 각 속성에 대한 명세(dataclasses.Field 의 __slots__)를
  원소로 하는 튜플을 반환한다.
- 보통, 객체 속성의 변수 명(name) 내지는 type 정보를 가져올 때 사용한다.

usage

from dataclasses import dataclass, field, asdict, astuple, fields

@dataclass(order=True)
class Person:
    first_name: str = field(compare=False)
    last_name: str = field(compare=False)
    full_name: str = field(init=False, compare=False)
    age: int = field(repr=False)
    hobby: list[str] = field(default_factory=list, compare=False)
    korean: bool = True
- 데이터 class 에 @dataclass 데코레이터를 적용한다.
- 기본적으로, 객체 속성은 type annotation 형식으로 작성한다.
  - 하지만 어디까지나 어노테이션일 뿐, 타입에 대한 강제성은 없다.
    즉, Python 의 동적 언어 특성을 해치지 않는다. 

    def __post_init__(self):
        self.full_name = self.last_name + self.first_name


hong = Person("길동", "홍", 18, ['stealing', 'flying'])
hong_clone = Person("길동", "홍", 18, ['stealing', 'flying'], True)
lee = Person("순신", "이", 30, "reading")
kim = Person("구", "김", 40)


print(hong)  # Person(first_name='길동', last_name='홍', full_name='홍길동', hobby=['stealing', 'flying'], korean=True)
print(hong == hong_clone)  # True
print(astuple(lee))  # ('순신', '이', '이순신', 30, 'reading', True)
print(asdict(kim))  # {'first_name': '구', 'last_name': '김', 'full_name': '김구', 'age': 40, 'hobby': [], 'korean': True}
print(lee > kim)  # False
print(sorted([kim, lee, hong]))  # [Person(..홍길동..), Person(..이순신..), Person(..김구..)]  // age: repr=False
print([field.name for field in fields(Person)])  # ['first_name', 'last_name', 'full_name', 'age', 'hobby', 'korean']

newhyork

이전 포스트

RegEx

다음 포스트

Python built-in library

logging

usage

argparse

usage

threading

usage

subprocess

run

usage

Popen

usage

dataclasses

feature

usage

RegEx

CI/CD, Github Actions

0개의 댓글