ORM(Object Relational Mapping)은 개발자가 SQL을 직접 작성하지 않고 객체지향적으로 DB를 다룰 수 있게 해줍니다. 하지만 ORM을 처음 쓰면 거의 모두가 마주치는 문제가 있습니다. 바로 N+1 문제입니다.
저 역시 프로젝트를 진행하면서 이 문제를 겪었고, 단순한 최적화를 넘어서 실제 데이터가 많아지면 어떻게 되는지 궁금했습니다. 그래서 이번 글에서는 N+1 문제 → 로딩 전략 선택 → 실제 AWS 환경에서 간단히 실험한 결과를 공유하려 합니다.
2. N+1 문제를 직접 겪다
예제로 User(작성자) – Post(게시글) 구조를 생각해 봅시다. User와 Post는 1:N 관계로 한 유저가 여러 포스트를 작성할수 있어요.
users 테이블
id
name
1
Choi
2
Kim
3
Yim
4
Lee
posts 테이블
id
title
user_id
content
1
Hello World
1
안녕??....................
2
ORM is awesome
2
ORM 은 너무 멋지다~..........
3
N+1 problem sucks
1
N+1 뭐냐 ..........
4
SQLAlchemy Tips
3
SQLalchmey는 이렇게 사용할 수 있어요! ...
5
Python is beautiful
4
Python은 너무 아름답다 그 이유는 ...
SQLalchemy를 활용해 ORM Model은 아래처럼 만들 수 있어요.
from sqlalchemy import Column, Integer, String, Text, ForeignKey
from sqlalchemy.orm import declarative_base, relationship
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(50), nullable=False)
# 역참조 (User.posts 로 접근 가능)
posts = relationship("Post", back_populates="user")
class Post(Base):
__tablename__ = "posts"
id = Column(Integer, primary_key=True, autoincrement=True)
title = Column(String(200), nullable=False)
content = Column(Text)
user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
# User와 관계 설정 (Post.user 로 접근 가능)
user = relationship("User", back_populates="posts")
N+1 발생 상황
모든 포스트에 대해서, 포스트 제목과 포스트를 작성한 사람의 이름을 같이 출력할 때 아래와 같이 작성할 수 있습니다. 포스트가 100개라고 가정할 때,
# N+1 발생 예시 (SQLAlchemy)
posts = session.query(Post).limit(100).all()
for post in posts:
print(post.title, post.user.name) # 여기서 user 접근 시마다 추가 쿼리
위 경우 실제 실행되는 SQL 쿼리문은 아래와 같아요.
SELECT * FROM posts LIMIT 100; -- 1번
SELECT * FROM users WHERE id=1; -- 2번
SELECT * FROM users WHERE id=2; -- 3번
SELECT * FROM users WHERE id=3; -- 4번
...
SELECT * FROM users WHERE id=N; -- N+1번
총 100+1번(N+1번) 쿼리 실행 → 데이터가 늘어날수록 기하급수적으로 느려집니다. 저도 실제로 이 문제 때문에 API 응답 속도가 수십 초까지 늘어나는 경험을 했습니다.
3. 해결 방법: 로딩 전략 선택
ORM에서는 relationship을 가져올 때 다양한 로딩 전략을 선택할 수 있습니다. 대표적인 것이 두 가지입니다.
3.1 Fetch Join (joinedload)
fetch join은 JPA/Hibernate(JAVA ORM)에서 온 개념인데, join을 통해 연관된 엔티티(객체)를 미리 가져온다는 개념입니다. SQLAlchemy도 같은 개념을 차용해서 joinedload 옵션을 제공하고, 이것을 흔히 "Fetch Join"이라고 부릅니다.
from sqlalchemy.orm import joinedload
posts = (
session.query(Post)
.options(joinedload(Post.user))
.limit(100)
.all()
)
for post in posts:
print(post.title, post.user.name) # 여기서 user 접근 시마다 추가 쿼리 없음
위 코드는 Postgres 기준으로 아래의 SQL query문이 실행됩니다.
SELECT posts.id AS posts_id,
posts.title AS posts_title,
posts.content AS posts_content,
posts.user_id AS posts_user_id,
users_1.id AS users_1_id,
users_1.name AS users_1_name
FROM posts
LEFT OUTER JOIN users AS users_1 ON users_1.id = posts.user_id
LIMIT 100;
즉 쿼리 1번으로 Post와 User를 JOIN 해서 모두 가져오는 의미로 Python에서 post.user.name 접근할 때 추가적으로 쿼리가 발생하지 않습니다.
Fetch Join은 쿼리 1번으로 해결되지만, 1:N 관계에서는 row 중복이 발생할 수 있습니다. 위와 달리, User 기준으로 posts를 가져올때, 아래와 같이 joinedload를 쓸수있는데요.
from sqlalchemy.orm import joinedload
users = (
session.query(User)
.options(joinedload(User.posts))
.limit(100)
.all()
)
for user in users:
for post in user.posts:
print(user.name, post.title)
실행되는 SQL
SELECT users.id AS users_id, users.name AS users_name,
posts_1.id AS posts_1_id, posts_1.title AS posts_1_title, posts_1.content AS posts_1_content, posts_1.user_id AS posts_1_user_id
FROM users
LEFT OUTER JOIN posts AS posts_1 ON users.id = posts_1.user_id
LIMIT 100;
결과테이블
users.id
users.name
posts_1_id
posts_1_title
posts_1_content
posts_1_user_id
1
Choi
1
Hello World
안녕??..............
1
1
Choi
3
N+1 problem sucks
N+1 뭐냐 ..........
1
2
Kim
2
ORM is awesome
ORM 은 너무 ........
2
3
Yim
4
SQLAlchemy Tips
SQLalchmey는 ...
3
4
Lee
5
Python is beautiful
Python은 .....
4
여기서 문제는 User 데이터가 Post 수만큼 중복 포함된다는 점입니다.
Choi가 2개의 글을 썼다면 row 2개에서 Choi가 중복됨.
만약 Choi가 1만 개의 글을 썼다면, User 데이터도 1만 번 반복됨.
LIMIT 기준도 row에 적용돼서 의도한 User 개수와 맞지 않을 수 있음.
3.2 Select IN (selectinload)
위와 같이 User기준으로 Posts를 가져올때는 selectedinload를 사용하는게 더 효과적입니다.
from sqlalchemy.orm import selectinload
users = (
session.query(User)
.options(selectinload(User.posts))
.limit(100)
.all()
)
for user in users:
for post in user.posts:
print(user.name, post.title)
실행되는 SQL
-- 1. users 조회
SELECT users.id AS users_id, users.name AS users_name
FROM users
LIMIT 100;
-- 2. posts 조회 (IN 조건)
SELECT posts.id AS posts_id, posts.title AS posts_title, posts.user_id AS posts_user_id
FROM posts
WHERE posts.user_id IN (1, 2, 3, 4, 5);
쿼리가 2번실행되지만 User와 Post를 분리해서 전송하기 때문에 User 관련 중복데이터 전송이 없습니다.
4. 실제 AWS 환경에서의 실험
이론적으로만 이해하는 것이 아니라, 실제로 데이터가 많아지면 얼마나 성능 차이가 나는지 확인해보고 싶었습니다. 그래서 간단한 실험 환경을 AWS에 구성했습니다.
실험 환경
EC2: Ubuntu 20.04, Python 3.10
RDS: PostgreSQL, db.t3.medium
라이브러리: SQLAlchemy, psycopg2
데이터 규모:
Users: 1,000명 (고정)
Posts: 1천 → 1만 → 5만 → 10만 개까지 점진적으로 증가
4.1 실험 방법
N+1: Post를 불러온 뒤 post.user에 접근 → User 쿼리가 매번 실행됨
Fetch Join (joinedload): Post와 User를 JOIN으로 한 번에 가져옴
Select IN (selectinload): Post를 먼저 불러온 뒤, 필요한 User만 IN 조건으로 추가 조회
각 방식으로 같은 수의 Post를 조회하고, 실행 시간을 비교했습니다.
4.2 실험 결과
아래 표는 같은 Post 수를 조회했을 때 걸린 시간입니다.
Posts 수
N+1 방식
joinedload
selectinload
1,000
1.55초
0.023초
0.038초
10,000
2.73초
0.17초
0.12초
50,000
3.72초
0.77초
0.66초
100,000
6.34초
1.47초
1.50초
4.3 결과 해석
N+1
Post가 만개만 되어도 이미 2초 가까이 소요.
5만 건부터는 3초 이상 걸림
→ 작은 데이터셋에서는 눈치 못 채다가, 데이터가 늘면 성능이 폭발적으로 나빠지는 전형적인 N+1 문제.
Fetch Join
User를 JOIN해서 한 번에 가져오기 때문에 N+1보다 압도적으로 빠름.
Post 수가 늘어도 안정적으로 1초~2초 수준에서 처리 가능.
Select IN
Post를 먼저 조회하고 필요한 User를 IN 조건으로 가져오기 때문에, Fetch Join보다 쿼리 한 번이 더 많음.
하지만 User 데이터가 중복 포함되지 않아서 대량 데이터에서 더 안정적.
이번 실험에서 10만 건 기준으로 Fetch Join과 비슷한 속도를 보임
5. 마무리
이번 글에서는 ORM을 사용할 때 흔히 맞닥뜨리는 N+1 문제를 직접 재현하고, SQLAlchemy의 Fetch Join(joinedload) 과 Select IN(selectinload) 을 통해 성능이 얼마나 개선되는지 확인했습니다.
중요한 것은 “쿼리 속도 몇 초 차이”가 아니라, N+1 구조는 데이터가 커지면 절대 확장될 수 없다는 점입니다. 따라서 ORM을 쓸 때는 반드시 로딩 전략을 의식적으로 선택해야 하며, 상황에 맞는 방법을 쓰는 습관이 필요합니다.
우리 회사 플랫폼의 제품군은 크게 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를 넣어줘야함