우리 회사 플랫폼의 제품군은 크게 Label, Curate, Model, 그리고 Apps로 구성되어 있다.
고객이 업로드한 이미지나 비디오에서 Curate를 통해 데이터를 선별하고, Label에서 라벨을 붙이고, Model에서 AI 모델을 학습시켜 플랫폼 사용자는 나만의 Vision AI 모델을 손쉽게 API 형태로 사용할 수 있다.
그중 Apps는 고객이 플랫폼을 더 쉽고 유연하게 활용할 수 있도록 돕는 자동화 도구들을 제공한다.
주요 목적은 다음과 같다:
기존에 다른 라벨링 툴에서 작업한 데이터를 Superb Platform 에 업로드
(ex. YOLO, COCO, Labelme 등 다양한 어노테이션 포맷 → Superb 형식으로 자동 변환 후 플랫폼에 업로드)
이미지에서 사람 얼굴을 감지하고 자동으로 비식별화(blur)
OCR로 텍스트 영역을 자동 감지하고 바운딩 박스를 생성
즉, 고객이 “이사”오는 과정에서 겪는 여러 번거로운 과정을 자동화함으로써 onboarding을 부드럽게 만든다.
Apps는 어떻게 실행되는가?
Apps는 여러 기능들을 개별 app 형태로 제공하는 Superb Platform의 구성 요소이며, 각 app은 독립적인 컨테이너로 실행되어 Kubernetes 상에서 운영된다. 실행 흐름은 다음과 같다:
사용자가 웹 프론트를 통해 데이터를 업로드하면, 해당 파일은 S3에 저장된다.
지정된 app이 실행되어 S3의 입력 데이터를 내려받아 처리한다.
처리 결과는 다시 S3에 업로드되거나, 다운로드 링크 또는 플랫폼 내 리소스로 제공된다.
이 모든 과정은 Apps의 백엔드 서버와의 통신을 통해 상태 및 진행 상황이 관리된다.
개별 app은 다양한 개발자가 만들지만, 공통된 실행 흐름과 시스템 환경을 모두 이해해야만 동작하도록 만들고 싶지는 않았다.
AppWrapper의 목적은 앱 실행에 필요한 공통 작업들을 추상화하여, 개발자가 비즈니스 로직에만 집중할 수 있게 하는 것이었다.
하지만 이 실행 흐름은 단순해 보이는 것과 달리, 환경 설정, presigned URL 요청, 오류 처리, 결과 업로드 등을 앱마다 수작업으로 구현하면 유지보수와 품질 관리가 매우 어려워진다.
AppWrapper의 역할
Apps의 실행 흐름에는 파일 다운로드, 결과 업로드, 상태 보고 등 반복되는 작업이 필수적으로 포함된다. 이런 공통 작업을 자동으로 처리해주는 유틸리티가 바로 AppWrapper다.
AppWrapper는 클래스 데코레이터 방식으로 동작하며, 다음과 같은 실행 흐름을 관리한다:
앱 실행 전 (Pre-Processing)
앱 실행에 필요한 입력 파일 목록과 파일 저장 경로, 작업 ID 등의 실행 정보를 시스템으로부터 전달받는다.
전달받은 정보를 기반으로, 사용자가 웹에서 업로드한 파일들을 S3에서 가져와 Pod의 로컬 디렉터리에 저장한다. (정확히는 emptyDir 형태)
만약 실행 준비 중 오류가 발생하면, 해당 작업은 즉시 실패 처리되고 상태가 백엔드로 보고된다.
앱 실행 중
개발자가 작성한 process() 함수를 호출하여 실제 비즈니스 로직을 수행한다.
예를 들어 다음과 같이 간단히 작성할 수 있다:
@AppWrapper()
def process():
# 예: YOLO 라벨 포맷을 Superb 포맷으로 변환
return {
"type": "download",
"file_path": "/tmp/converted.zip"
}
process()
실행 후 (Post-Processing)
결과의 타입이 download인 경우 → presigned PUT URL을 통해 파일 업로드 후 웹 프론트에서 사용자가 결과 파일 download 가능
결과의 타입이 link인 경우 → URL 그대로 반환
상태 업데이트 및 로그 저장 (S3에 로그 업로드 포함)
결과 포맷은 단순하지만 명확하다
이 통일된 구조 덕분에, 어떤 앱이든 동일한 방식으로 결과를 처리할 수 있다.
Case 1: type이 link인 경우
{
"type": "link",
"url": "특정url",
"label": "Go to Project"
}
app의 결과물이 hyperlink인 경우
url key가 필요 (ex) platform의 특정 url)
label key는 화면에서 보여줄 내용 (ex) 아래 사진에 가장 오른쪽 “Go to Project “버튼 )
Case 2: type이 download인 경우
app의 결과물이 파일인경우
file_path key가 필요 (Pod 내부의 파일 경로)
AppWrapper의 도입 효과
항목
개선 전
개선 후
코드 중복
presigned 요청, S3 처리 반복
데코레이터로 추상화
예외 처리
앱마다 오류 처리 제각각
실패 시 상태 일괄 업데이트
로그 수집
개발자가 직접 구현
자동 S3 업로드
개발 속도
새로운 앱 추가 시 부가작업 다수
비지니스 로직만 구현하면 끝
이 구조 덕분에 앱 개발자는 “파일을 어떻게 받을지, 어디에 저장할지, 상태를 어떻게 보고할지” 같은 운영상의 디테일을 몰라도 앱 로직만 작성하면 된다.
실제 AppWrapper 데코레이터 내부는 어떻게 구현되어 있을까?
AppWrapper는 클래스지만, __call__ 메서드를 구현함으로써 함수 데코레이터처럼 동작한다. 로컬 모드와 운영 모드가 명확히 분기되어 있어 개발/운영 환경에서 모두 활용 가능하고, 예외 발생 시 즉시 로그 저장 & 상태 보고를 통해 시스템 안정성 확보한다.
class AppWrapper:
....
def __call__(self, func):
def inner(*args, **kwargs):
if self.LOCAL_MODE:
try:
result = func(*args, **kwargs)
except Exception as e:
logger.error(e)
sys.exit(1)
else:
try:
result = func(*args, **kwargs) # 비지니스 로직 함수
except Exception as e:
logger.error(traceback.format_exc())
.....
if task_data["Task"]["status"] != Status.Canceled.value:
self.send_pod_log_to_s3() # pod 로그 s3로 전송
self.update_status(Status.Failed.value, str(e))
try:
# 반환된 결과 포맷 유효성 검사
self.validate_result_format(result)
# 파일이 포함된 경우 업로드
if result["type"] == "download":
self.upload_file_to_s3(result)
# 결과를 backend에 전송
data = {"task_id": self.task_id, "result": result}
self.send_pod_log_to_s3()
requests.post(
f"{self.TOAD_HOST}/output/",
data=json.dumps(data),
)
# 성공 상태 업데이트
self.update_status(Status.Complete.value, "App completed")
except Exception as e:
error_log = f"Post app failed: {e}"
self.update_status(Status.Failed.value, error_log)
return inner
AppWrapper는 어떻게 활용되고 있을까?
AppWrapper는 내부 앱 개발자들이 공통으로 사용할 수 있도록 구조화되어 있으며,
별도의 설치 없이 앱 코드에서 쉽게 가져다 사용할 수 있도록 PyPI에 패키징되어 배포되어 있다.
$ pip install AppWrapper
PyPI에 공개(링크)되어 있어 누구나 설치는 가능하지만, 소스 코드는 private GitHub repository에서 관리되고 있어 내부 사용자만 직접 수정하거나 검토할 수 있다.
앱 개발자는 별도의 설정 없이 @AppWrapper 데코레이터만 붙이면, 파일 다운로드, 결과 업로드, 상태 보고, 로그 저장 등 반복되는 실행 흐름을 자동으로 처리할 수 있다.
개선 포인트와 앞으로의 방향
AppWrapper는 현재까지 수십 개의 앱에서 안정적으로 사용되며, 개별 app들의 실행 흐름에 많은 역할을 하고 있다. 하지만 실제 운영하면서 다음과 같은 개선 가능성도 보였다:
1. 단일 클래스가 너무 많은 책임을 가짐
환경 초기화, 파일 다운로드/업로드, 상태 전송, 로그 처리, 예외 핸들링 등 너무 많은 역할이 한 클래스에 몰려 있다.
SRP(Single Responsibility Principle)를 따르는 구조로 리팩토링할 필요가 있다. (예: S3Client, StatusManager, ResultValidator 등으로 분리)
2. 결과 포맷의 유효성 검사는 코드로만 정의됨
result는 dict 형식에 "type", "url" 혹은 "file_path"가 필요하지만, 이를 Pydantic 등으로 명시하지 않았다. Pydantic 기반의 result schema 도입을 통해 유효성 검사를 개선할 수 있겠다.
/* Tokenizer state */
struct tok_state {
/* Input state; buf <= cur <= inp <= end */
/* NB an entire line is held in the buffer */
char *buf; /* Input buffer, or NULL; malloc'ed if fp != NULL */
char *cur; /* Next character in buffer */
char *inp; /* End of data in buffer */
const char *end; /* End of input buffer if buf != NULL */
const char *start; /* Start of current token if not NULL */
int done; /* E_OK normally, E_EOF at EOF, otherwise error code */
/* NB If done != E_OK, cur must be == inp!!! */
FILE *fp; /* Rest of input; NULL if tokenizing a string */
int tabsize; /* Tab spacing */
int indent; /* Current indentation index */
int indstack[MAXINDENT]; /* Stack of indents */
int atbol; /* Nonzero if at begin of new line */
int pendin; /* Pending indents (if > 0) or dedents (if < 0) */
const char *prompt, *nextprompt; /* For interactive prompting */
int lineno; /* Current line number */
int first_lineno; /* First line of a single line or multi line string
expression (cf. issue 16806) */
int level; /* () [] {} Parentheses nesting level */
/* Used to allow free continuations inside them */
char parenstack[MAXLEVEL];
int parenlinenostack[MAXLEVEL];
PyObject *filename;
/* Stuff for checking on different tab sizes */
int altindstack[MAXINDENT]; /* Stack of alternate indents */
/* Stuff for PEP 0263 */
enum decoding_state decoding_state;
int decoding_erred; /* whether erred in decoding */
int read_coding_spec; /* whether 'coding:...' has been read */
char *encoding; /* Source encoding. */
int cont_line; /* whether we are in a continuation line. */
const char* line_start; /* pointer to start of current line */
const char* multi_line_start; /* pointer to start of first line of
a single line or multi line string
expression (cf. issue 16806) */
PyObject *decoding_readline; /* open(...).readline */
PyObject *decoding_buffer;
const char* enc; /* Encoding for the current str. */
char* str;
char* input; /* Tokenizer's newline translated copy of the string. */
int type_comments; /* Whether to look for type comments */
/* async/await related fields (still needed depending on feature_version) */
int async_hacks; /* =1 if async/await aren't always keywords */
int async_def; /* =1 if tokens are inside an 'async def' body. */
int async_def_indent; /* Indentation level of the outermost 'async def'. */
int async_def_nl; /* =1 if the outermost 'async def' had at least one
NEWLINE token after it. */
};
(2) 토크나이저 상태는 커서의 현재 위치 같은 정보를 저장 (Parser/tokenizer.h)
/* Tokenizer state */
struct tok_state {
/* Input state; buf <= cur <= inp <= end */
/* NB an entire line is held in the buffer */
char *buf; /* Input buffer, or NULL; malloc'ed if fp != NULL */
char *cur; /* Next character in buffer */
char *inp; /* End of data in buffer */
const char *end; /* End of input buffer if buf != NULL */
const char *start; /* Start of current token if not NULL */
int done; /* E_OK normally, E_EOF at EOF, otherwise error code */
/* NB If done != E_OK, cur must be == inp!!! */
FILE *fp; /* Rest of input; NULL if tokenizing a string */
int tabsize; /* Tab spacing */
int indent; /* Current indentation index */
int indstack[MAXINDENT]; /* Stack of indents */
int atbol; /* Nonzero if at begin of new line */
int pendin; /* Pending indents (if > 0) or dedents (if < 0) */
const char *prompt, *nextprompt; /* For interactive prompting */
int lineno; /* Current line number */
int first_lineno; /* First line of a single line or multi line string
expression (cf. issue 16806) */
int level; /* () [] {} Parentheses nesting level */
/* Used to allow free continuations inside them */
char parenstack[MAXLEVEL];
int parenlinenostack[MAXLEVEL];
PyObject *filename;
/* Stuff for checking on different tab sizes */
int altindstack[MAXINDENT]; /* Stack of alternate indents */
/* Stuff for PEP 0263 */
enum decoding_state decoding_state;
int decoding_erred; /* whether erred in decoding */
int read_coding_spec; /* whether 'coding:...' has been read */
char *encoding; /* Source encoding. */
int cont_line; /* whether we are in a continuation line. */
const char* line_start; /* pointer to start of current line */
const char* multi_line_start; /* pointer to start of first line of
a single line or multi line string
expression (cf. issue 16806) */
PyObject *decoding_readline; /* open(...).readline */
PyObject *decoding_buffer;
const char* enc; /* Encoding for the current str. */
char* str;
char* input; /* Tokenizer's newline translated copy of the string. */
int type_comments; /* Whether to look for type comments */
/* async/await related fields (still needed depending on feature_version) */
int async_hacks; /* =1 if async/await aren't always keywords */
int async_def; /* =1 if tokens are inside an 'async def' body. */
int async_def_indent; /* Indentation level of the outermost 'async def'. */
int async_def_nl; /* =1 if the outermost 'async def' had at least one
NEWLINE token after it. */
};
파서-토크나이저는 tok_get()으로 다음 토큰을 얻고 그 아이디를 파서로 전달
// Parser/tokenizer.c L1174
/* Get next token, after space stripping etc. */
static int
tok_get(struct tok_state *tok, const char **p_start, const char **p_end)
{
int c;
int blankline, nonascii;
*p_start = *p_end = NULL;
nextline:
tok->start = NULL;
blankline = 0;
/* Get indentation level */
if (tok->atbol) {
...
return PyToken_OneChar(c);
}
파서는 파서 생성기 오토마타(DFA)로 CST에 노드를 추가
640줄이 넘는 tok_get()은 CPython 코드 중에서도 손꼽히게 복잡한 부분중 하나. 루프에서 토크나이저와 파서를 호출하는 과정은 아래와 같음
CST→AST로 변환하려면 PyParser_ParseFileObject()에서 반환된 CST의 루트인 node가 필요
typedef struct _node {
short n_type;
char *n_str;
int n_lineno;
int n_col_offset;
int n_nchildren;
struct _node *n_child;
int n_end_lineno;
int n_end_col_offset;
} node;
CST는 구문, 토큰 아이디, 심벌을 모두 포함하기 때문에 컴파일러가 사용하기에는 적합하지 않음
AST를 살펴보기에 앞서 파서 단계의 결과를 확인하는 방법이 있는데, parser 모듈은 C함수의 파이썬 API를 제공
>>> import parser
st<stdin>:1: DeprecationWarning: The parser module is deprecated and will be removed in future versions of Python
>>> st = parser.expr('a+1')
>>>
>>>
>>> pprint(parser.st2list(st))
[258,
[332,
[306,
[310,
[311,
[312,
[313,
[316,
[317,
[318,
[319,
[320,
[321, [322, [323, [324, [325, [1, 'a']]]]]],
[14, '+'],
[321, [322, [323, [324, [325, [2, '1']]]]]]]]]]]]]]]]],
[4, ''],
[0, '']]
Parser 모듈의 출력은 숫자형식으로 make regen-grammar 단계에서 Include/token.h파일에 저장된 코튼과 심벌의 번호와 같음
좀더 보기 쉽게 symbol과 token 모듈의 모든 번호로 딕셔너리를 만든 후 parser.st2list()의 출력을 토큰과 심벌의 이름으로 재귀적으로 바꾸면 아래와 같음
import symbol
import token
import parser
from pprint import pprint
def lex(expression):
symbols = {v: k for k, v in symbol.__dict__.items() if isinstance(v, int)}
tokens = {v: k for k, v in token.__dict__.items() if isinstance(v, int)}
lexicon = {**symbols, **tokens}
st = parser.expr(expression)
st_list = parser.st2list(st)
def replace(l: list):
r = []
for i in l:
if isinstance(i, list):
r.append(replace(i))
else:
if i in lexicon:
r.append(lexicon[i])
else:
r.append(i)
return r
return replace(st_list)
pprint(lex("a + 1"))
CST는 코드 파일의 텍스트를 있는 그대로 표현하는 구조로, 텍스트로부터 토큰을 추출하여 토큰 종류만 구분해 둔 상태에 불과
CST로 기본적인 문법 구조는 알 수 있지만 함수, 스코프, 루프 같은 파이썬 언어 사양에 대한 의미를 결정할 수 없음
코드를 컴파일 하기 전 CST를 실제 파이썬 언어구조와 의미 요소를 표현하는 고수준 구조인 AST로 변환해야 함
예를 들어 AST에서 이항 연산은 표현식의 한 종류인 BinOp로 표현. 해당 표현식은 세가지 요소로 이루어짐
left: 왼쪽 항
op: 연산자(+, -, * 등)
right: 오른쪽 항
다음은 a + 1 에 대한 AST
AST는 CPython 파싱 과정 중에 생성하지만 표준 라이브러리 ast모듈을 사용해서 파이썬 코드에서 AST를 생성할 수도 있음
6.3.1 AST 연관된 소스 파일 목록
Include/python-ast.h: Parser/asdl_c.py 로 생성한 AST 노드 타입 선언
parser/Python.asdl: 도메인 특화 언어인 ASDL(abstract syntax description language) 5로 작성된 ast 노드 타입들과 프로퍼티 목록
Python/ast.c: AST 구현
6.3.2 인스타비즈로 AST 시각화하기
$ pip install instaviz
AST와 컴파일된 코드를 웹 인터페이스로 시각화 하는 파이썬 패키지
❯ python
Python 3.9.20 (main, Sep 6 2024, 19:03:56)
[Clang 15.0.0 (clang-1500.3.9.4)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>> import instaviz
>>> def example():
... a = 1
... b = a + 1
... return b
...
>>>
>>> instaviz.show(example)
트리의 각 노드의 타입은 AST 노드 클래스
ast 모듈에서 찾을 수 있는 노드 클래스들은 모두 _ast.AST를 상속
CST와 달리, AST의 노드들은 특정 프로퍼티들을 통해 자식 노드와 연결됨
b = a + 1 이 선언된 줄과 연결된 assign 노드를 클릭하면 아래와 같음
Assign 노드는 2개의 프로퍼티를 가짐
targets는 값이 할당될 이름의 목록. 언패킹을 통해 한 번에 여러 이름에 값을 할당할 수 있기 때문에 목록이 필요
value는 이름에 할당할 값. 이 경우에 BinOp 표현식 a + 1 이 할당됨
BinOp노드는 세 개의 프로퍼티를 가짐
left: 왼쪽 항
op: 연산자, 이 경우에는 더 하기를 뜻하는 Add 노드(+)
right: 오른쪽 항
6.3.3 AST 컴파일
C에서 AST를 컴파일하는 것을 매우 복잡한 작업으로 Python/ast.c 모듈은 5000줄이 넘는 코드로 이루어져 있음
AST의 공개 API는 CST와 파일 이름, 컴파일러 플래그 ,메모리 저장 영역을 인자로 받음
반환 타입은 파이썬 모듈을 표현하는 mod_ty 타입으로 Include/Python-ast.h 에서 정의
mod_ty는 아래 4가지 모듈 타입 중 하나를 담는 컨테이너 구조체
Module
Interactive
Expression
FunctionType
모듈 타입은 Parser/Python.asdl에서 정의하는데 문장, 표현식, 연산자, 컴프리헨션 타입들도 찾을 수 있음
AST가 생성하는 클래스들과 표준 라이브러리 ast 모듈의 클래스들은 Parser/Python.asdl에서 정의하는 타입들
AST는 ALMOSTEQUAL 토큰이 비교 연산자인 AlE라는 것을 아직 알 수 없어서, 토큰을 연산자로 인식할 수 있게 AST관련 C 코드를 수정
Python/ast.c의 ast_for_comp_op()로 이동해서 switch문을 찾아보면, _cmpop 열거형 값 중 하나를 반환 함
static cmpop_ty
ast_for_comp_op(struct compiling *c, const node *n)
{
/* comp_op: '<'|'>'|'=='|'>='|'<='|'!='|'in'|'not' 'in'|'is'
|'is' 'not'
*/
REQ(n, comp_op);
if (NCH(n) == 1)
{
n = CHILD(n, 0);
switch (TYPE(n))
{
case LESS:
return Lt;
case GREATER:
return Gt;
case ALMOSTEQUAL: // 추가된 내용
return AlE; // 추가된 내용
case EQEQUAL: /* == */
return Eq;
case LESSEQUAL:
return LtE;
case GREATEREQUAL:
return GtE;
case NOTEQUAL:
return NotEq;
case NAME:
if (strcmp(STR(n), "in") == 0)
return In;
if (strcmp(STR(n), "is") == 0)
return Is;
/* fall through */
이제 토크나이저와 AST모두 코드를 파싱할 수 있지만 컴파일러는 아직 이 연산자를 실행하는 방법을 모름
AST로 거의 같음 연산자를 아래와 같이 확인해볼 수 있음
import ast
m = ast.parse('1 ~= 2')
m.body[0].value.ops[0]
<_ast.AlE object at 0x111111>
일반적으로 누락된 파라미터를 기본 값으로 바꾸어도 큰 문제가 없지만 오류가 있는 데이터를 유사한 값으로 대체하는 것을 더 위험하여 일부 오류를 숨겨버릴 수 있음
예외처리
어떤 경우에는 잘못된 데이터를 사용하여 계속 실행하는 것보다는 차라리 실행을 멈추는 것이 더 좋을 수 있음
입력이 잘못되었을 때만 함수에 문제가 생기는 것이 아님 (외부 컴포넌트에 연결되어 있는 경우)
이런 경우에는 함수 자체의 문제가 아니기 때문에 적절하게 인터페이스를 설계하면 쉽게 디버깅 할 수 있음
⇒ 예외적인 상황을 명확하게 알려주고 원래의 비즈니스 로직에 따라 흐름을 유지하는 것이 중요
정상적인 시나리오나 비즈니스 로직을 예외처리하려고 하면 프로그램의 흐름을 읽기가 어려워짐
→ 예외를 go-to문처럼 사용하는 것과 같다. 올바른 위치에서 추상화를 하지 못하게 되고 로직을 캡슐화하지도 못하게 됨.
마지막으로 예외를 대게 호출자에게 잘못을 알려주는 것으로 캡슐화를 약화시키기 때문에 신중하게 사용해야 함→이는 함수가 너무 많은 책임을 가지고 있다는 것을 의미할 수도 있음. 함수에서 너무 많은 예외를 발생시켜야 한다면 여러개의 작은 기능으로 나눌 수 있는지 검토해야 함
올바른 수준의 추상화 단계에서 예외 처리
예외는 오직 한가지 일을 하는 함수의 한 부분이어야 함
서로 다른 수준의 추상화를 혼합하는 예제. deliver_event 메소드를 중점적으로 살펴보면
import logging
import time
logger = logging.getLogger(__name__)
class DataTransport:
"""다른 레벨에서 예외를 처리하는 객체의 예"""
_RETRY_BACKOFF: int = 5
_RETRY_TIMES: int = 3
def __init__(self, connector):
self._connector = connector
self.connection = None
def deliver_event(self, event):
try:
self.connect()
data = event.decode()
self.send(data)
except ConnectionError as e:
logger.info("커넥션 오류 발견: %s", e)
raise
except ValueError as e:
logger.error("%r 이벤트에 잘못된 데이터 포함: %s", event, e)
raise
def connect(self):
for _ in range(self._RETRY_TIMES):
try:
self.connection = self._connector.connect()
except ConnectionError as e:
logger.info("%s: 새로운 커넥션 시도 %is", e, self._RETRY_BACKOFF)
time.sleep(self._RETRY_BACKOFF)
else:
return self.connection
raise ConnectionError(f"연결실패 재시도 횟수 {self._RETRY_TIMES} times")
def send(self, data):
return self.connection.send(data)
def deliver_event(self, event):
try:
self.connect()
data = event.decode()
self.send(data)
except ConnectionError as e:
logger.info("커넥션 오류 발견: %s", e)
raise
except ValueError as e:
logger.error("%r 이벤트에 잘못된 데이터 포함: %s", event, e)
raise
ConnectionError와 ValueError는 별로 관계가 없음
매우 다른 유형의 오류를 살펴봄으로써 책임을 어떻게 분산해야 하는지에 대한 아이디어를 얻을 수 있음
ConnectionError는 connect 메소드 내에서 처리되어야 함. 이렇게 하면 행동을 명확하게 분리할 수 있다. 메소드가 재시도를 지원하는 경우 메소드 내에서 예외처리를 할 수 있음
ValueError는 event의 decode 메소드에 속한 에러로 event를 send 메소드에 파라미터로 전달 후 send 메소드 내에서 예외처리를 할 수 있음
위 내용처럼 구현을 수정하면 deliver_event 메소드에서 예외를 catch할 필요가 없음
def connect_with_retry(connector, retry_n_times: int, retry_backoff: int = 5):
"""<connector>를 사용해 연결을 시도함.
연결에 실패할 경우 <retry_n_times>회 만큼 재시도
재시도 사이에는 <retry_backoff>초 만큼 대기
연결에 성공하면 connection 객체를 반환
재시도 횟수를 초과하여 연결에 실패하면 ConnectionError 오류 발생
:param connector: connect() 메소드를 가진 객체
:param retry_n_times: 연결 재시도 횟수
:param retry_backoff: 재시도 사이의 대기 시간(초)
"""
for _ in range(retry_n_times):
try:
return connector.connect()
except ConnectionError as e:
logger.info("%s: 새로운 커넥션 시도 %is", e, retry_backoff)
time.sleep(retry_backoff)
exc = ConnectionError(f"연결 실패 ({retry_n_times}회 재시도)")
logger.exception(exc)
raise exc
class DataTransport:
"""추상화 수준에 따른 예외 분리를 한 객체"""
_RETRY_BACKOFF: int = 5
_RETRY_TIMES: int = 3
def __init__(self, connector: Connector) -> None:
self._connector = connector
self.connection = None
def deliver_event(self, event: Event):
self.connection = connect_with_retry(
self._connector, self._RETRY_TIMES, self._RETRY_BACKOFF
)
self.send(event)
def send(self, event: Event):
try:
return self.connection.send(event.decode())
except ValueError as e:
logger.error("%r contains incorrect data: %s", event, e)
raise
deliver_event 메소드 내에서 예외 catch 하는 부분 없어짐
엔드 유저에게 Traceback 노출 금지
보안을 위한 고려사항으로 예외가 전파되도록하는 경우는 중요한 정보를 공개하지 않고 “알 수 없는 문제가 발생했습니다” 또는 “페이지를 찾을 수 없습니다”와 같은 일반적인 메세지를 사용해야 함
비어있는 except 블록 지양
파이썬의 안티패턴 중 가장 악마같은 패턴(REAL 01)으로 어떠한 예외도 발견할 수 업슨 문제점이 있음
명시적으로 해당 오류를 무시하려면 contextlib.suppress 함수를 사용하는 것이 올바른 방법
import contextlib
with contextlib.suppress(KeyError):
process_data()
원본 예외 포함
raise <e> from <original_exception> 구문을 사용하면 여러 예외를 연결할 수 있음
원본 오류의 traceback 정보가 새로운 exception에 포함되고 원본 오류는 새로운 오류의 원인으로 분류되어 cause 속성에 할당 됨
class InternalDataError(Exception):
"""업무 도메인 데이터의 예외"""
def process(data_dictionary, record_id):
try:
return data_dictionary[record_id]
except KeyError as e:
raise InternalDataError("데이터가 존재하지 않음") from e
test_dict = {"a": 1}
process(test_dict, "b")
Traceback (most recent call last): File "/Users/woo-seongchoi/Desktop/CleanCode/ch3/main.py", line 7, in process return data_dictionary[record_id] ~~~~~~~~~~~~~~~^^^^^^^^^^^ KeyError: 'b'*
The above exception was the direct cause of the following exception: Traceback (most recent call last): File "/Users/woo-seongchoi/Desktop/CleanCode/ch3/main.py", line 14, in <module> process(test_dict, "b") File "/Users/woo-seongchoi/Desktop/CleanCode/ch3/main.py", line 9, in process raise InternalDataError("데이터가 존재하지 않음") from e InternalDataError: 데이터가 존재하지 않음*
파이썬에서 assertion 사용하기
절대로 일어나지 않아야 하는 상황에 사용되므로 assert 문에 사용된 표현식을 불가능한 조건을 의미로 프로그램을 중단시키는 것이 좋다
try:
assert condition.holds(), "조건에 맞지 않음"
except AssertionError:
alternative_procedure() # catch 후에도 계속 프로그램을 실행하면 안됨
위 코드가 나쁜 또 다른 이유는 AssertionError를 처리하는 것 이외에 assertion 문장이 함수라는 것
assert condition.holds(), "조건에 맞지 않음"
함수 호출은 부작용을 가질 수 있으며 항상 반복가능하지 않음. 또한 디버거를 사용해 해당 라인에서 중지하여 오류 결과를 편리하게 볼 수 없으며 다시 함수를 호출한다 하더라도 잘못된 값이었는지 알 수 없음
result = condition.holds()
assert result > 0, f"Error with {result}"
/* --- PyPreConfig ----------------------------------------------- */
typedef struct {
int _config_init; /* _PyConfigInitEnum value */
/* Parse Py_PreInitializeFromBytesArgs() arguments?
See PyConfig.parse_argv */
int parse_argv;
/* If greater than 0, enable isolated mode: sys.path contains
neither the script's directory nor the user's site-packages directory.
Set to 1 by the -I command line option. If set to -1 (default), inherit
Py_IsolatedFlag value. */
int isolated;
/* If greater than 0: use environment variables.
Set to 0 by -E command line option. If set to -1 (default), it is
set to !Py_IgnoreEnvironmentFlag. */
int use_environment;
/* Set the LC_CTYPE locale to the user preferred locale? If equals to 0,
set coerce_c_locale and coerce_c_locale_warn to 0. */
int configure_locale;
/* Coerce the LC_CTYPE locale if it's equal to "C"? (PEP 538)
Set to 0 by PYTHONCOERCECLOCALE=0. Set to 1 by PYTHONCOERCECLOCALE=1.
Set to 2 if the user preferred LC_CTYPE locale is "C".
If it is equal to 1, LC_CTYPE locale is read to decide if it should be
coerced or not (ex: PYTHONCOERCECLOCALE=1). Internally, it is set to 2
if the LC_CTYPE locale must be coerced.
Disable by default (set to 0). Set it to -1 to let Python decide if it
should be enabled or not. */
int coerce_c_locale;
/* Emit a warning if the LC_CTYPE locale is coerced?
Set to 1 by PYTHONCOERCECLOCALE=warn.
Disable by default (set to 0). Set it to -1 to let Python decide if it
should be enabled or not. */
int coerce_c_locale_warn;
#ifdef MS_WINDOWS
/* If greater than 1, use the "mbcs" encoding instead of the UTF-8
encoding for the filesystem encoding.
Set to 1 if the PYTHONLEGACYWINDOWSFSENCODING environment variable is
set to a non-empty string. If set to -1 (default), inherit
Py_LegacyWindowsFSEncodingFlag value.
See PEP 529 for more details. */
int legacy_windows_fs_encoding;
#endif
/* Enable UTF-8 mode? (PEP 540)
Disabled by default (equals to 0).
Set to 1 by "-X utf8" and "-X utf8=1" command line options.
Set to 1 by PYTHONUTF8=1 environment variable.
Set to 0 by "-X utf8=0" and PYTHONUTF8=0.
If equals to -1, it is set to 1 if the LC_CTYPE locale is "C" or
"POSIX", otherwise it is set to 0. Inherit Py_UTF8Mode value value. */
int utf8_mode;
/* If non-zero, enable the Python Development Mode.
Set to 1 by the -X dev command line option. Set by the PYTHONDEVMODE
environment variable. */
int dev_mode;
/* Memory allocator: PYTHONMALLOC env var.
See PyMemAllocatorName for valid values. */
int allocator;
} PyPreConfig;
5.1.1 딕셔너리 초기화 구성
사용자 환경 또는 운영 체제와 관련된 구성이기 때문에 런타임 구성과 구분됨
PyPreConfig 의 세가지 주요 기능
파이썬 메모리 할당자 설정
LC_CTYPE 로캘(locale)을 시스템 또는 사용자 선호 로캘로 구성하기
UTF-8 모드 설정하기(PEP540)
아래와 같은 int 타입 필드들을 포함
allocator: PYMEM_ALLOCATOR_MALLOC 같은 값을 사용해 메모리 할당자를 선택
configure_locale: LC_CTYPE 로캘을 사용자 선호 로캘로 설정. 0으로 설정하면 coerce_c_locale과 coerce_c_locale_warn을 0으로 설정
coerce_c_locale: 2로 설정하면 C 로캘을 강제로 적용. 1로 설정하면 LC_CTYPE을 읽은 후 강제로 적용할지 결정
coerce_c_locale_warn: 0이 아니면 C로캘이 강제로 적용될 때 경고가 발생
dev_mode: 개발 모드를 활성화
isolated: 격리 모드를 활성화. sys.path에 스크립트 디렉토리와 사용자의 사이트 패키지 디렉토리가 포함되지 않음
legacy_windows_fs_encoding: 0이 아니면 UTF-8 모드를 비활성화하고 파이썬 파일 시스템 인코딩을 mbcs로 설정(윈도우 전용)
parse_argv: 0이 아니면 명령줄 인자를 사용
use_environment: 0보다 큰 값이면 환경 변수를 사용
utf8_mode_: 0이 아니면 UTF-8모드를 활성화
5.1.2 연관된 소스 파일 목록
PyPreConfig와 연관된 소스 파일 목록
Python/initconfig.c : 시스템 환경에서 불러온 구성을 명령줄 플래그와 결합
Include/cpython/initconfig.h : 초기화 구성 구조체를 정의
5.1.3 런타임 구성 구조체
PyConfig 런타임 구성 구조체는 아래 값들을 포함
‘디버그’나 ‘최적화’같은 실행 모드 플래그
스크립트 파일이나 stdin, 모듈 등 실행 모드
-X <option>으로 설정 가능한 확장 옵션
런타임 설정을 위한 환경 변수
런타임 구성 데이터는 CPython 런타임 기능의 활성화 여부를 결정
5.1.4 명령줄로 런타임 구성 설정하기
파이썬은 다양한 명령줄 인터페이스 옵션을 제공하는데 예로 상세 모드(verbose) 가 있고 주로 개발자 대상으로 활용됨
wchar_t* 타입은 UTF-8 문자를 저장할 수 있기 때문에 CPython에서 저수준 유니코드 데이터를 저장하는 타입으로 사용됨
Objects/unicodeobject.c의 헬퍼 함수 PyUnicode_FromWideChar()를 이용해 wchar_t*를 파이썬 유니코드 문자열로 변환할 수 있음. 유니코드 문자열→ UTF-8로 인코딩하려면 PyUnicode_AsUTF8String() 을 사용
pymain_run_command() 함수 코드
static int
pymain_run_command(wchar_t *command, PyCompilerFlags *cf)
{
PyObject *unicode, *bytes;
int ret;
unicode = PyUnicode_FromWideChar(command, -1);
if (unicode == NULL) {
goto error;
}
if (PySys_Audit("cpython.run_command", "O", unicode) < 0) {
return pymain_exit_err_print();
}
bytes = PyUnicode_AsUTF8String(unicode);
Py_DECREF(unicode);
if (bytes == NULL) {
goto error;
}
ret = PyRun_SimpleStringFlags(PyBytes_AsString(bytes), cf);
Py_DECREF(bytes);
return (ret != 0);
error:
PySys_WriteStderr("Unable to decode the command from the command line:\n");
return pymain_exit_err_print();
}
pymain_run_command()는 파이썬 바이트열 객체를 PyRun_SimpleStringFlags()로 넘겨서 실행
static int
pymain_run_command(wchar_t *command, PyCompilerFlags *cf)
{
PyObject *unicode, *bytes;
int ret;
unicode = PyUnicode_FromWideChar(command, -1);
bytes = PyUnicode_AsUTF8String(unicode);
ret = PyRun_SimpleStringFlags(PyBytes_AsString(bytes), cf);
}
Python/pythonrun.c의 PyRun_SimpleStringFlags()는 문자열을 파이썬 모듈로 변환하고 실행
int
PyRun_SimpleStringFlags(const char *command, PyCompilerFlags *flags)
{
PyObject *m, *d, *v;
m = PyImport_AddModule("__main__"); // (1)
if (m == NULL)
return -1;
d = PyModule_GetDict(m);
v = PyRun_StringFlags(command, Py_file_input, d, d, flags); // (2)
if (v == NULL) {
PyErr_Print();
return -1;
}
Py_DECREF(v);
return 0;
}
(1) 파이썬 모듈을 독립된 모듈로 실행하려면 __main__ 진입점이 필요하기 때문에 PyRun_simpleStringFlags()가 진입점을 자동으로 추가
(2) PyRun_simpleStringFlags() 는 딕셔너리와 모듈을 만든 후 PyRun_StringFlags() 를 호출해, 가짜 파일 이름을 만들고 파이썬 파서를 실행해 문자열에서 추상 구문 트리(abstract syntax tree, AST)를 생성해 모듈로 반환
5.3.4 로컬 모듈 입력
파이썬의 -m 옵션과 모듈 이름으로 파이썬 명령을 실행할 수도 있음
예시
$ ./python.exe -m unittest
위 명령으로 표준 라이브러리의 unittest 모듈을 실행할 수 있음
-m 옵션은 모듈 패키지의 진입점(__main__)을 실행함. 이 때 해당 모듈은 sys.path에서 검색
임포트 라이브러리(importlib)의 검색 메커니즘 덕분에 unittest 모듈의 파일 시스템 위치를 기억할 필요 없음
CPython은 표준 라이브러리 모듈 runpy를 임포트하고 PyObject_Call()로 해당 모듈을 실행. runpy 모듈은 Lib/runpy.py에 위치한 순수한 파이썬 모듈임.
임포트는 Python/import.c의 C API 함수 PyImport_ImportModule()이 담당
python -m <module> 을 실행하는 것은 python -m runpy <module> 을 실행하는 것과 같음. runpy 모듈은 운영체제에서 모듈을 찾아 실행하는 프로세스를 추상화
runpy는 세단계의 모듈을 실행 …?
1. 제공된 모듈 이름을 __import__()로 임포트 2. __name__(모듈 이름)을 __main__ 이름 공간에서 설정 3. __main__ 이름 공간에서 모듈을 실행
5.3.5 표준 입력 또는 스크립트 파일 입력
python test.py처럼 python을 실행할 때 첫 번째 인자가 파일명이라면 Cpython은 파일 핸들을 열어 Python/pythonrun.c의 PyRun_SimpleFileExFlags()로 핸들을 넘김
이 함수는 세 종류의 파일 경로를 처리할 수 있음
.pyc 파일 경로면 run_pyc_file()을 호출
스크립트 파일(.py) 경로면 PyRun_FileExFlags()를 호출
<command> | python 처럼 파일 경로가 stdin이면 stdin을 파일 핸들로 취급하고 PyRun_FileExFlags()를 호출
stdin이나 스크립트 파일의 경우 CPython은 파일 핸들을 Python/pythonrun.c의 PyRunFileExFlags()로 넘기는데 이는 PyRun_SimpleStringFlags()와 비슷. CPython은 파일 핸들을 PyParser_ASTFromFileObject()로 전달
PyRun_SimpleStringFlags() 처럼 PyRunFileExFlags()는 파일에서 파이썬 모듈을 생성하고 run_mode()로 보내 실행
5.3.6 컴파일된 바이트 코드 입력
python을 .pyc 파일 경로와 함께 실행하면 CPython은 파일을 텍스트 파일로 불러와 파싱하는 대신 .pyc 파일에서 디스크에 기록된 코드 객체를 찾음
PyRun_SimpleFileExFlags()에는 .pyc파일 경로를 처리하는 부분이 있음
Python/pythonrun.c의 run_pyc_file()은 파일 핸들을 사용해 .pyc 파일에서 코드 객체를 마셜링함
마셜링은 파일 내용을 메모리를 복사하여 특정 데이터 구조로 변환하는 것을 의미
CPython 컴파일러는 스크립트가 호출될 때 마다 파싱하는 대신 디스크의 코드 객체 구조체에 컴파일한 코드를 캐싱
메모리에 마셜링된 코드 객체는 Python/ceval.c 를 호출하는 run_eval_code_obj()로 전달되어 실행됨
통역을 하려면 출발어(source language)와 도착어(target language)의 문법 구조를 알아야 함
컴파일러의 선택기준: 이식성
저수준 기계어: C/C++, Go, 파스칼은 바이너리 실행파일로 컴파일하는데 이는 컴파일한 플랫폼과 동일한 플랫폼에서만 사용할 수 있음
중간 언어: java, 닷넷 CLR은 여러 시스템 아키텍처에서 사용할 수 있는 중간언어로 컴파일 해서 가상머신에서 실행될 수 있음
파이썬 애플리케이션은 보통 소스 코드 형태로 배포됨
파이썬 인터프리터는 소스 코드를 변환한 후 한 줄씩 실행
CPython 런타임이 첫 번째 실행될 때 코드를 컴파일하지만 이 단계는 일반 사용자에게 노출되지 않음
파이썬 코드는 기계어 대신 바이트코드라는 저수준 중간 언어로 컴파일되고 바이트코드는 .pyc 파일에 저장됨(캐싱)
코드를 변경하지 않고 같은 파이썬 애플리케이션을 다시 실행하면 매번 다시 컴파일하지 않고 컴파일된 바이트 코드를 불러오기 때문에 더 빠르게 실행
4.1 CPython이 파이썬이 아니라 C로 작성된 이유
컴파일러가 작동하는 방식 때문
컴파일러가 작동하는 방식 유형
셀프 호스팅 컴파일러: 자기 자신으로 작성한 컴파일러로 부트스트래핑이라는 단계를 통해서 만들어짐
Go: C로 작성된 첫 번째 Go 컴파일러가 Go를 컴파일할 수 있게 되자 컴파일러를 Go 언어로 재 작성
PyPy: 파이썬으로 작성된 파이썬 컴파일러
source to source 컴파일러: 컴파일러를 이미 가지고 있는 다른 언어로 작성한 컴파일러
Cpython : C를 사용하여 Python 컴파일
ssl 이나 sockets 같은 표준 라이브러리 모듈이 저수준 운영체제 API에 접근하기 위해 C로 작성됨
4.2 파이썬 언어 사양
컴파일러가 언어를 실행하려면 문법 구조에 대한 엄격한 규칙이 필요
CPython 소스 코드에 포함된 언어 사양은 모든 파이썬 인터프리터 구현이 사용하는 레퍼런스 사양
사람이 읽을 수 있는 형식 + 기계가 읽을 수 있는 형식으로 제공
문법 형식과 각 문법 요소가 실행되는 방식을 자세히 설명
4.2.1 파이썬 언어 레퍼런스
파이썬 언어의 기능을 설명하는 reStructuredText(.rst) 파일을 담고 있음 (사람이 읽기 위한 언어 사양)
cpython/Doc/reference
├── compound_stmts.rst # 복합문 (if, while, for, 함수 정의 등)
├── introduction.rst # 레퍼런스 문서 개요
├── index.rst # 언어 레퍼런스 목차
├── datamodel.rst # 객체, 값, 타입
├── executionmodel.rst # 프로그램 구조
├── expressions.rst # 표현식 구성 요소
├── grammar.rst # 문법 규격(Grammar/Grammar 참조)
├── import.rst # import 시스템
├── lexical_analysis.rst # 어휘 구조 (줄, 들여쓰기, 토큰, 키워드 등)
├── simple_stmts.rst # 단순문 (assert, import, return, yield 등)
└── toplevel_components.rst # 스크립트 및 모듈 실행 방법 설명
예시
Doc→reference→compound_stmts.rst 에서 간단한 예시로 with 문의 정의를 찾을 수 있음
기계가 읽을 수 있는 사양은 Grammar→python.gram이라는 단일 파일 안에 들어 있음
4.2.2 문법 파일
파서 표현식 문법(Parsing Expression Grammar, PEG) 사양을 사용하고 아래 표기법을 사용
파이썬 컴파일러는 문법 파일을 직접 사용하지 않고 파서 생성기가 문법 파일에서 생성한 파서를 사용
문법 파일을 수정하면 파서를 재생성한 후 CPython을 다시 컴파일해야 함
파서란?
파이썬 3.9부터 CPython은 파서 테이블 생성기(pgen 모듈) 대신 문맥 의존 문법 파서를 사용
기존 파서는 파이썬 3.9까지는 -X oldparser 플래그를 활성화해 사용할 수 있으며 파이썬 3.10에서 완전히 제거됨
4.4 문법 다시 생성하기
새로운 PEG 생성기인 pegen을 테스트해보기 위해 문법 일부를 변경해봄
# Grammar/python.gram L66 을 아래처럼 변경 (|'proceed' 추가)
| ('pass'|'proceed') { _Py_Pass(EXTRA) }
변경 후 아래 명령어로 문법 파일을 다시 빌드
$ make regen-pegen
PYTHONPATH=./Tools/peg_generator python3.9 -m pegen -q c \\
./Grammar/python.gram \\
./Grammar/Tokens \\
-o ./Parser/pegen/parse.new.c
python3.9 ./Tools/scripts/update_file.py ./Parser/pegen/parse.c ./Parser/pegen/parse.new.c
makefile이 있는 폴더에서 아래 명령어를 실행하면 proceed 키워드를 사용할 수 있음을 확인
$ ./python.exe
Python 3.9.20+ (heads/3.9-dirty:011fb84db5f, Nov 8 2024, 21:32:35)
[Clang 15.0.0 (clang-1500.3.9.4)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>> def example():
... proceed
...
>>>
>>> example()
위 과정을 통해 CPython 문법을 수정하고 컴파일해서 새로운 CPython을 만든 시도를 한 것
4.4.1 토큰
Grammar 폴더 내에 Tokens 파일에서 파스 트리의 leaf node에서 사용되는 고유한 토큰들을 정의함
각 토큰은 이름과 자동으로 생성된 고유 아이디(ID)를 가지고, 이름을 사용하면 토크타이저에서 토큰을 더 쉽게 참조할 수 있음
LPAR '('
RPAR ')'
SEMI ':'
Tokens 파일을 수정하면 pegen을 다시 실행해야 하고 tokenize 모듈을 이용하면 토큰이 사용되는 걸 호가인할 수 있음
현재의 문자를 나타내는 value, 다음에 나올 문자를 나타내는 next_ 배열을 가지고 있음
linked list나 tree 형태와 비슷
from typing import List
from dataclasses import dataclass, field
R = 26
@dataclass
class RTrieNode:
size = R
value: int
next_: List["RTrieNode"] = field(default_factory=lambda: [None] * R)
def __post_init__(self):
if len(self.next_) != self.size:
raise ValueError(f"리스트(next_)의 길이가 유효하지 않음")
size는 class variable로 모든 객체가 값을 공유
value는 정수형이지만 기본값이 없으므로 객체 생성시 반드시 값을 정해줘야 함
next_는 R크기 만큼의 길이를 가진 list로 초기화
__post_init__은 next_가 원하는 형태로 잘 생성되었는지 확인하는 검증
from typing import List
from dataclasses import dataclass, field
R = 26 # 영어 알파벳
@dataclass
class RTrieNode:
size = R
value: int
next_: List["RTrieNode"] = field(default_factory=list)
def __post_init__(self):
if len(self.next_) != self.size:
raise ValueError(f"리스트(next_)의 길이가 유효하지 않음")
rt_node = RTrieNode(value=0) # ValueError: 리스트(next_)의 길이가 유효하지 않음
이터러블 객체
__iter__ 매직 메소드를 구현한 객체
파이썬의 반복은 이터러블 프로토콜이라는 자체 프로토콜을 사용해 동작
for e in my_object
위 형태로 객체를 반복할 수 있는지 확인하기 위해 파이썬은 고수준에서 아래 두가지 차례로 검사
객체가 __next__나 __iter__ 메서드 중 하나를 포함하는지 여부
객체가 시퀀스이고 __len__과 __getitem__을 모두 가졌는지 여부
For-loop에 대한 구체적인 과정
my_list = ["사과", "딸기", "바나나"]
for i in my_list:
print(i)
for 문이 시작할 때 my_list의 __iter__()로 iterator를 생성
내부적으로 i = __next__() 호출
StopIteration 예외가 발생하면 반복문 종료
Iterable과 Iterator의 차이
Iterable: loop에서 반복될 수 있는 python 객체, __iter__() 가 구현되어있어야 함
Iterator: iterable 객체에서 __iter__() 호출로 생성된 객체로 __iter__()와 __next__()가 있어야하고, iteration 시 현재의 순서를 가지고 있어야 함
이터러블 객체 만들기
객체 반복 시 iter() 함수를 호출하고 이 함수는 해당 객체에 __iter__ 메소드가 있는지 확인
from datetime import timedelta
from datetime import date
class DateRangeIterable:
"""자체 이터레이터 메서드를 가지고 있는 iterable"""
def __init__(self, start_date, end_date):
self.start_date = start_date
self.end_date = end_date
self._present_day = start_date
def __iter__(self):
return self # 객체 자신이 iterable 임을 나타냄
def __next__(self):
if self._present_day >= self.end_date:
raise StopIteration()
today = self._present_day
self._present_day += timedelta(days=1)
return today
for day in DateRangeIterable(date(2024, 6, 1), date(2024, 6, 4)):
print(day)
2024-06-01 2024-06-02 2024-06-03
for 루프에서 python은 객체의 iter() 함수를 호출하고 이 함수는 __iter__ 매직 메소드를 호출
self를 반환하면서 객체 자신이 iterable임을 나타냄
루프의 각 단계에서마다 자신의 next() 함수를 호출
next 함수는 다시 __next__ 메소드에게 위임하여 요소를 어떻게 생산하고 하나씩 반환할 것인지 결정
더 이상 생산할 것이 없는 경우 파이썬에게 StopIteration 예외를 발생시켜 알려줘야함
⇒ for 루프가 작동하는 원리는 StopIteration 예외가 발생할 때까지 next()를 호출하는 것과 같다
from datetime import timedelta
from datetime import date
class DateRangeIterable:
"""자체 이터레이터 메서드를 가지고 있는 이터러블"""
def __init__(self, start_date, end_date):
self.start_date = start_date
self.end_date = end_date
self._present_day = start_date
def __iter__(self):
return self
def __next__(self):
if self._present_day >= self.end_date:
raise StopIteration()
today = self._present_day
self._present_day += timedelta(days=1)
return today
r = DateRangeIterable(date(2024, 6, 1), date(2024, 6, 4))
print(next(r)) # 2024-06-01
print(next(r)) # 2024-06-02
print(next(r)) # 2024-06-03
print(next(r)) # raise StopIteration()
위 예제는 잘 동작하지만 하나의 작은 문제가 있음
max 함수 설명
iterable한 object를 받아서 그 중 최댓값을 반환하는 내장함수이다
숫자형뿐만 아니라 문자열 또한 비교 가능
str1 = 'asdzCda'
print(max(str1)) # z
str2 = ['abc', 'abd']
print(max(str2)) # abd 유니코드가 큰 값
str3 = ['2022-01-01', '2022-01-02']
print(max(str3)) # 2022-01-02
# 숫자로 이루어진 문자열을 비교할 때 각 문자열의 앞 부분을 비교해서 숫자가 큰 것을 출력
with 문이 없고 함수를 호출하면 offline_backup 함수가 context manager 안에서 자동으로 실행됨
원본 함수를 래핑하는 데코레이터 형태로 사용
단점은 완전히 독립적이라 데코레이터는 함수에 대해 아무것도 모름 (사실 좋은 특성)
contextlib 의 추가적인 기능
import contextlib
with contextlib.suppress(DataConversionException):
parse_data(nput_json_or_dict)
안전하다고 확신되는 경우 해당 예외를 무시하는 기능
DataConversionException이라고 표현된 예외가 발생하는 경우 parse_data 함수를 실행
컴프리헨션과 할당 표현식
코드를 간결하게 작성할 수 있고 가독성이 높아짐
def run_calculation(i):
return i
numbers = []
for i in range(10):
numbers.append(run_calculation(i))
print(numbers) # [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
위의 코드를 아래와 같이 바로 리스트 컴프리헨션으로 만들 수 있음
numbers = [run_calculation(i) for i in range(10)]
list.append를 반복적으로 호출하는 대신 단일 파이썬 명령어를 호출하므로 일반적으로 더 나은 성능을 보임
dis 패키지를 이용한 어셈블리코드 비교각 assembly 코드 (list comprehension)
import dis
def run_calculation(i):
return i
def list_comprehension():
numbers = [run_calculation(i) for i in range(10)]
return numbers
# Disassemble the list comprehension function
dis.dis(list_comprehension)
def for_loop():
numbers = []
for i in range(10):
numbers.append(run_calculation(i))
return numbers
# Disassemble the for loop function
dis.dis(for_loop)
import re
from typing import Iterable, Set
# Define the regex pattern for matching the ARN format
ARN_REGEX = r"arn:(?P<partition>[^:]+):(?P<service>[^:]+):(?P<region>[^:]*):(?P<account_id>[^:]+):(?P<resource_id>[^:]+)"
def collect_account_ids_from_arns(arns: Iterable[str]) -> Set[str]:
"""
arn:partition:service:region:account-id:resource-id 형태의 ARN들이 주어진 경우 account-id를 찾아서 반환
"""
collected_account_ids = set()
for arn in arns:
matched = re.match(ARN_REGEX, arn)
if matched is not None:
account_id = matched.groupdict()["account_id"]
collected_account_ids.add(account_id)
return collected_account_ids
# Example usage
arns = [
"arn:aws:iam::123456789012:user/David",
"arn:aws:iam::987654321098:role/Admin",
"arn:aws:iam::123456789012:group/Developers",
]
unique_account_ids = collect_account_ids_from_arns(arns)
print(unique_account_ids)
# {'123456789012', '987654321098'}
위 코드 중 collect_account_ids_from_arns 함수를 집중해서 보면,
def collect_account_ids_from_arns(arns: Iterable[str]) -> Set[str]:
"""
arn:partition:service:region:account-id:resource-id 형태의 ARN들이 주어진 경우 account-id를 찾아서 반환
"""
collected_account_ids = set()
for arn in arns:
matched = re.match(ARN_REGEX, arn)
if matched is not None:
account_id = matched.groupdict()["account_id"]
collected_account_ids.add(account_id)
return collected_account_ids
위 코드를 컴프리헨션을 이용해 간단히 작성 가능
def collect_account_ids_from_arns(arns: Iterable[str]) -> Set[str]:
"""
arn:partition:service:region:account-id:resource-id 형태의 ARN들이 주어진 경우 account-id를 찾아서 반환
"""
matched_arns = filter(None, (re.match(ARN_REGEX, arn) for arn in arns))
return {m.groupdict()["account_id"] for m in matched_arns}
python 3.8이후에는 할당표현식을 이용해 한문장으로 다시 작성 가능
def collect_account_ids_from_arns(arns: Iterable[str]) -> Set[str]:
"""
arn:partition:service:region:account-id:resource-id 형태의 ARN들이 주어진 경우 account-id를 찾아서 반환
"""
return {
matched.groupdict()["account_id"]
for arn in arns
if (matched := re.match(ARN_REGEX, arn)) is not None
}
정규식 이용한 match 결과들 중 None이 아닌 것들만 matched 변수에 저장되고 이를 다시 사용
더 간결한 코드가 항상 더 나은 코드를 의미하는 것은 아니지만 분명 두번째나 세번째 코드가 첫번째 코드보다는 낫다는 점에서는 의심의 여지가 없음
이는 여러번 확장되는 클래스의 메소드 이름을 충돌없이 오버라이드 하기 위해 만들어진거로 pythonic code의 예가 아님
결론
⇒ 속성을 private으로 정의하는 경우 하나의 밑줄 사용
프로퍼티(Property)
class Coordinate:
def __init__(self, lat: float, long: float) -> None:
self._latitude = self._longitude = None
self.latitude = lat
self.longitude = long
@property
def latitude(self) -> float:
return self._latitude
@latitude.setter
def latitude(self, lat_value: float) -> None:
print("here")
if lat_value not in range(-90, 90+1):
raise ValueError(f"유호하지 않은 위도 값: {lat_value}")
self._latitude = lat_value
@property
def longitude(self) -> float:
return self._longitude
@longitude.setter
def longitude(self, long_value: float) -> None:
if long_value not in range(-180, 180+1):
raise ValueError(f"유효하지 않은 경도 값: {long_value}")
self._longitude = long_value
coord = Coordinate(10, 10)
print(coord.latitude)
coord.latitude = 190 # ValueError: 유호하지 않은 위도 값: 190
property 데코레이터는 무언가에 응답하기 위한 쿼리
setter는 무언가를 하기 위한 커맨드
둘을 분리하는 것이 명령-쿼리 분리 원칙을 따르는 좋은 방법
보다 간결한 구문으로 클래스 만들기
객체의 값을 초기화하는 일반적인 보일러플레이트
보일러 플레이트: 모든 프로젝트에서 반복해서 사용하는 코드
def __init__(self, x, y, ...):
self.x = x
self.y = y
파이썬 3.7부터는 dataclasses 모듈을 사용하여 위 코드를 훨씬 단순화할 수 있다 (PEP-557)
@dataclass 데코레이터를 제공
클래스에 적용하면 모든 클래스의 속성에 대해서 마치 __init__ 메소드에서 정의한 것처럼 인스턴스 속성으로 처리
@dataclass 데코레이터가 __init__ 메소드를 자동 생성
field라는 객체 제공해서 해당 속성에 특별한 특징이 있음을 표시
속성 중 하나가 list처럼 변경가능한 mutable 데이터 타입인 경우 __init__에서 비어 있는 리스트를 할당할 수 없고 대신에 None으로 초기화한 다음에 인스턴스마다 적절한 값으로 다시 초기화 해야함
from dataclasses import dataclass
@dataclass
class Foo:
bar: list = []
# ValueError: mutable default <class 'list'> for field a is not allowed: use default_factory
안되는 이유는 위의 bar 변수가 class variable이라 모든 Foo 객체들 사이에서 공유되기 때문
class C:
x = [] # class variable
def add(self, element):
self.x.append(element)
c1 = C()
c2 = C()
c1.add(1)
c2.add(2)
print(c1.x) # [1, 2]
print(c2.x) # [1, 2]
아래처럼 default_factory 파라미터에 list 를 전달하여 초기값을 지정할 수 있도록 하면 됨
from dataclasses import dataclass, field
@dataclass
class Foo:
bar = field(default_factory=list)
Return a proxy object that delegates method calls to a parent or sibling class of type. This is useful for accessing inherited methods that have been overridden in a class.
공식문서 설명은 늘 어려움.
쉽게 말해, 부모나 형제 클래스의 임시 객체를 반환하고, 반환된 객체를 이용해 슈퍼 클래스의 메소드를 사용할 수 있음.
Cube가 아닌 Square기 때문에 super(Square, self)의 반환은 Square 클래스의 부모 클래스인 Rectangle 클래스의 임시 객체
결과적으로 Rectangle 인스턴스에서 area() 메소드를 찾음
Q. Square 클래스에 area 메소드를 구현하면??
그래도 super(Square, self) 가 Rectangle 클래스를 반환하기 때문에 Rectangle 인스턴스에서 area() 메소드를 호출
## super 클래스의 정의
class super(object):
def __init__(self, type1=None, type2=None): # known special case of super.__init__
"""
super() -> same as super(__class__, <first argument>)
super(type) -> unbound super object
**super(type, obj) -> bound super object; requires isinstance(obj, type)
super(type, type2) -> bound super object; requires issubclass(type2, type)**
Typical use to call a cooperative superclass method:
class C(B):
def meth(self, arg):
super().meth(arg)
This works for class methods too:
class C(B):
@classmethod
def cmeth(cls, arg):
super().cmeth(arg)
"""
# (copied from class doc)
두번째 argument : 첫번째 argument의 클래스 인스턴스를 넣어주거나 subclass를 넣어줘야함
(walrus := True) 는 assignment expression으로 walrus에 값 True를 할당한다.
둘의 미묘한 차이중 하나는 walrus = False는 값을 반환하지 않지만 (walrus := True)는 값을 반환한다는 것이다!
>>> walrus = False
>>> (walrus := True)
True
등장한 이유
PEP 572에 Abstract에 아래와 expression 내에서 변수에 할당하는 방법을 제안하고 있다.
creating a way to assign to variables within an expression using the notation NAME := expr.
C언어에서는 변수에 값을 할당하는 statement도 expression인데 강력하지만 찾기 힘든 버그를 생산하기도 한다.
int main(){
int x = 3, y = 8;
if (x = y) {
printf("x and y are equal (x = %d, y = %d)", x, y);
}
return 0;
}
x와 y값을 비교후 값이 같으면 두 값을 출력하는 코드지만 x와 y값이 다르기 때문에 아무것도 출력 안되길 기대되지만 실제 코드 실행 결과는 아래와 같이 print 문이 출력된다. 왜일까?
x and y are equal (x = 8, y = 8)
문제는 위 코드 세번째 줄 if (x = y) 에서 equality comparison operator(==) 대신 assignment operator(=) 를 사용하고 있기 때문이다. if 문의 조건에는 expression이 와야하는데 C언어에서는 x = y를 expression으로 x값이 8로 할당되고 1이상의 값으로 True로 판단되서 print문이 출력된다.
그럼 Python에서는?
x, y = 3, 8
if x = y:
print(f"x and y are equal ({x = }, {y = })")
SyntaxError: invalid syntax. Maybe you meant '==' or ':=' instead of '='?
Syntax Error를 내뱉는데 expression이 아닌 statement이기 때문이다. 파이썬은 이를 분명히 구분하고 walrus operator에도 이러한 설계 원칙이 반영되었다. 그래서 walrus operator를 이용해서 일반적인 assignment를 할 수 없다.
위 수식을 이용해 오슬로(59.9°N 10.8°E) 와 밴쿠버(49.3°N 123.1°W) 사이의 거리를 구하면,
from math import asin, cos, radians, sin, sqrt
# Approximate radius of Earth in kilometers
rad = 6371
# Locations of Oslo and Vancouver
ϕ1, λ1 = radians(59.9), radians(10.8)
ϕ2, λ2 = radians(49.3), radians(-123.1)
# Distance between Oslo and Vancouver
print(2 * rad * asin(
sqrt(
sin((ϕ2 - ϕ1) / 2) ** 2
+ cos(ϕ1) * cos(ϕ2) * sin((λ2 - λ1) / 2) ** 2
)
))
# 7181.7841229421165 (km)
위 수식을 검증하기 위해서 수식의 일부 값을 확인해야할 수 있는데 수식의 일부를 복&붙으로 확인할 수 있다.
이 경우 코드를 읽는 사람들에게 len_numbers와 sum_numbers 변수는 계산을 최적화하기 위해 dictionary 내부에서만 사용했고 다시 사용되지 않음을 명확히 전달 할 수 있다
(3) Text 파일에서 lines, words, character 수 세는 예시
# wc.py
import pathlib
import sys
for filename in sys.argv[1:]:
path = pathlib.Path(filename)
counts = (
path.read_text().count("\\n"), # Number of lines
len(path.read_text().split()), # Number of words
len(path.read_text()), # Number of characters
)
print(*counts, path) # 11 32 307 wc.py
wc.py 파일은 11줄, 32단어, 307 character로 구성되어있다
위 코드를 보면 path.read_text() 가 반복적으로 호출되는걸 알 수 있다 ⇒ walrus operator를 이용해 개선해보면,
import pathlib
import sys
for filename in sys.argv[1:]:
path = pathlib.Path(filename)
counts = (
**(text := path.read_text()).count("\\n"), # Number of lines**
len(text.split()), # Number of words
len(text), # Number of characters
)
print(*counts, path)
물론 아래처럼 text 변수를 이용하면 코드는 한줄 늘어나지만 readability를 훨신 높일 수 있다.
import pathlib
import sys
for filename in sys.argv[1:]:
path = pathlib.Path(filename)
text = path.read_text()
counts = (
text.count("\\n"), # Number of lines
len(text.split()), # Number of words
len(text), # Number of characters
)
print(*counts, path)
그러므로 walrus operator가 코드를 간결하게 해주더라도 readability를 고려해야 한다.
(4) List Comprehensions
List comprehension과 함께 연산이 많은 함수를 사용하게 될 때, walrus operator의 사용은 효과적일 수 있다.
import time
t_start = time.time()
def slow(num):
time.sleep(5)
return num
numbers = [4, 3, 1, 2, 5]
results = [slow(num) for num in numbers if slow(num) > 4]
t_end = time.time()
print("elapsed time: ", t_end - t_start)
elapsed time: 30.01522707939148
numbers 리스트의 각 element에 slow 함수를 적용 후 3보다 큰 경우에만 results에 slow 호출 결과를 저장하는 코드
문제는 slow 함수가 2번 호출됨
slow 호출 후 반환 결과가 3보다 큰지 확인할 때
results 리스트에 저장하기 위해 slow 호출할 때
가장 일반적인 해결책은 list comprehension 대신 for loop을 사용하는 것이다.
import time
t_start = time.time()
def slow(num):
time.sleep(5)
return num
numbers = [4, 3, 1, 2, 5]
results = []
for num in numbers:
slow_num = slow(num)
if slow_num > 4:
results.append(slow_num)
t_end = time.time()
print("elapsed time: ", t_end - t_start)
elapsed time: 25.021725063323975
slow 함수가 모든 경우에 한번씩만 호출됨
하지만 코드 양이 늘어나고 가독성이 떨어짐
walrus operator를 사용하면 list comprehension을 유지하면서 가독성을 높일 수 있음
import time
t_start = time.time()
def slow(num):
time.sleep(5)
return num
numbers = [4, 3, 1, 2, 5]
results = [slow_num for num in numbers if (slow_num := slow(num)) > 4]
print(results)
t_end = time.time()
print("elapsed time: ", t_end - t_start)
elapsed time: 25.018176908493042
(5) While Loop
question = "Do you use the walrus operator?"
valid_answers = {"yes", "Yes", "y", "Y", "no", "No", "n", "N"}
user_answer = input(f"\n{question} ")
while user_answer not in valid_answers:
print(f"Please answer one of {', '.join(valid_answers)}")
user_answer = input(f"\n{question} ")
위 코드는 사용자의 입력을 받는 input 함수가 두번 반복됨
이를 개선하기 위해 While True 와 break를 사용하여 코드를 다시 작성하는 것이 일반적임
question = "Do you use the walrus operator?"
valid_answers = {"yes", "Yes", "y", "Y", "no", "No", "n", "N"}
while True:
user_answer = input(f"\n{question} ")
if user_answer in valid_answers:
break
print(f"Please answer one of {', '.join(valid_answers)}")
walrus operator를 이용해서 while loop을 간결하게 할 수 있음
question = "Do you use the walrus operator?"
valid_answers = {"yes", "Yes", "y", "Y", "no", "No", "n", "N"}
while (user_answer := input(f"\n{question} ")) not in valid_answers:
print(f"Please answer one of {', '.join(valid_answers)}")
사용자로부터 받은 input 입력을 user_answer 변수에 저장하고 동시에 valid_answers 내에 포함되어있는지를 체크하여 가독성을 높일 수 있음
