데이터 사이의 복잡한 관계를 효율적으로 필터링하는 방법을 알아보자. 음양오행의 조화와 상호작용을 담는 데이터베이스를 만들어 보겠다.

60갑자는 10가지 천간과 12가지 지지가 조합되어 구성된다. 60개의 글자는 서로 상호작용하여 어떤 관계를 이룬다. 깊게 들어가면 너무 방대해지므로 여기선 간단한 관계만 살펴보겠다.

각 천간과 지지가 가진 음양오행 속성과 글자들간의 합/충/형/파 관계, 그리고 프로필 정보는 데이터베이스를 통해 관리하고 만세력 변환 및 분석은 Rust 로직으로 구현하겠다. 분석된 데이터는 다시 데이터베이스에 JSONB로 저장한다.

전반적인 흐름은 다음과 같다.

1. 입력 (Input): 사용자가 생년월일시를 입력하면, Rust 엔진의 만세력 로직이 이를 8개의 간지 코드로 변환합니다.
2. 저장 (Store): 변환된 간지 코드와 기본 정보를 saju_profiles에 저장합니다.
3. 분석 (Analyze): Rust 엔진이 DB의 ganji_interactions(정적 관계)를 참고하여 이 사주의 특징(오행 균형, 합충 등)을 계산합니다.
4. 캐싱 (Cache): 계산된 복잡한 결과물을 JSON 객체로 만들어 analysis_result 필드에 UPDATE 합니다.
5. 조회 (Query): 이후 사용자가 프로필을 볼 때는 다시 계산할 필요 없이 JSONB 필드만 읽어서 보여줍니다. 만약 "목(Wood) 기운이 30점 이상인 사람"을 찾는다면 GIN 인덱스를 통해 빠르게 필터링합니다.

작업공간 생성 및 구조 확인

기존에 만든 데이터베이스 템플릿을 복제해서 사용해 보겠다.

~/workspace$ cp -r db-template saju-analysis
~/workspace$ cd saju-analysis
~/workspace/saju-analysis$ tree -a -I .venv
.
├── .env                # 우리 프로젝트의 DB로 연결되게 수정 필요
├── .github
│   └── workflows
│       └── CI.yml
├── .gitignore
├── app
│   └── main.py            # 우리 프로젝트의 라우트 함수 추가 필요
├── Cargo.toml            # 우리 프로젝트의 정보로 수정 필요
├── docker-compose.yml    # 우리 프로젝트의 DB로 연결되게 수정 필요
├── pyproject.toml        # 우리 프로젝트의 정보로 수정 필요
└── src
    ├── db.rs            # 우리 프로젝트에 필요한 함수 추가 필요
    ├── lib.rs            # 우리 프로젝트에 필요한 함수 추가 필요
    └── logger.rs        # 그대로 유지해도 무방

pyproject.toml 파일을 열어 프로젝트 이름을 수정해 주겠다. 나머지 의존성 부분은 이번 프로젝트에서도 사용하는 것들이니 가만히 둔다.

~/workspace/saju-analysis$ vi pyproject.toml

pyproject.toml

[build-system]
requires = ["maturin>=1.12,<2.0"]
build-backend = "maturin"
>
[project]
name = "saju-analysis"
requires-python = ">=3.12"
classifiers = [
    "Programming Language :: Rust",
    "Programming Language :: Python :: Implementation :: CPython",
    "Programming Language :: Python :: Implementation :: PyPy",
]
dynamic = ["version"]
dependencies = [
    "fastapi>=0.124.4",
    "orjson>=3.10.15",
    "uvicorn>=0.33.0",
]
>
[dependency-groups]
dev = [
    "maturin>=1.12.6",
]

다음과 같이 uv 를 통해 pyproject.toml 파일의 모든 의존성을 설치할 수 있다.

~/workspace/saju-analysis$ uv sync

마찬가지로 Cargo.toml 파일도 수정해 준다. 이 때, 필요한 크레이트도 추가해 주도록 하자.

Cargo.toml

[package]
name = "saju-analysis"
version = "0.1.0"
edition = "2024"
>
[lib]
name = "rust_engine"
crate-type = ["cdylib"]
>
[dependencies]
# Python 연결
pyo3 = "0.28.0"
pythonize = "0.28"
# 비동기 엔진
tokio = { version = "1.43", features = ["full"] }
# 데이터베이스 연결
dotenvy = "0.15"
sqlx = { version = "0.8", features = ["runtime-tokio", "tls-rustls", "postgres", "macros", "uuid", "chrono", "json"] }
# 데이터베이스 호환
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
uuid = { version = "1.23", features = ["v7", "serde"] }
# 로그 수집
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }
tracing-appender = "0.2"

템플릿 프로젝트의 DB 정보를 그대로 사용하면 DB가 꼬이게 될 테니 DB 정보도 수정해 준다.

docker-compose.yml 의 변경 필요한 부분은 전부 환경변수로 되어 있으니 .env 파일만 편집하면 된다.

계정 정보는 그대로 사용해도 무방하지만 COMPOSE_PROJECT_NAME 과 POSTGRES_DB 는 수정해 주어야 한다.

~/workspace/saju-analysis$ vi .env

.env

# Docker Compose Project
COMPOSE_PROJECT_NAME=saju-analysis
>
# PostgreSQL
POSTGRES_USER=peter
POSTGRES_PASSWORD=ku201711424
POSTGRES_DB=saju_db
POSTGRES_HOST=127.0.0.1
POSTGRES_PORT=5432
>
# pgAdmin
PGADMIN_EMAIL=admin@pjos.dev
PGADMIN_PASSWORD=admin201711424
>
# SQLX
DATABASE_URL=postgres://peter:ku201711424@127.0.0.1:5432/saju_db?sslmode=disable

재차 말하지만 학습 기록용이니까 업로드하지 .env 파일의 내용은 유출하는 거 아니다. 보안 이슈!

DB 설정

스키마

이번 실습에서는 여러 개의 데이터베이스 테이블을 사용할 것이다.

기본적인 간지 정보를 담은 테이블의 스키마는 다음과 같다.

각 글자의 고정적인 상호작용을 담은 테이블의 스키마는 다음과 같다.

(부모)

(자식1: 글자를 기준으로 그것이 가질 수 있는 상호작용 및 설명을 담고 있다.)

(자식2: 상호작용의 해석을 담고 있다.)

관계 식별 예시 사주 팔자의 여덟 글자를 중복 없이 나열하여 관계를 확인한다. 관계의 유형별로 천간합은 100번대, 천간충은 200번대 등의 그룹 ID를 지정한다. '을'이 2개, '경'이 1개 있었어도 여기서는 '을'과 '경'을 하나씩 확인하고 (1, '합', 102, '을경합금'), (6, '합', 102, '을경합금') 과 같이 "2개가 만났을 때 합을 이루는 관계"에 대해 2개의 값이 있을 경우 이 관계가 있다고 본다. 1대1이 아닌 '을' 2개와 '경' 1개가 만나는 상황을 쟁합(爭合)이라고 하는데 이런 부분에 대한 처리는 데이터베이스가 아닌 Rust 엔진에서 하도록 하겠다. 데이터베이스에서는 단지 관계의 존재 여부만 확인한다.

개인의 사주 팔자가 담긴 사주 프로필 테이블의 스키마는 다음과 같다.

이 데이터베이스 테이블들이 존재하지 않는 경우에만 새로 생성하는 코드를 작성해 보자.

~/workspace/saju-analysis$ mkdir scripts
~/workspace/saju-analysis$ vi scripts/init.sql

scripts/init.sql

-- 간지 정보
>
DO $$
BEGIN
    IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'element_type') THEN
        CREATE TYPE element_type AS ENUM ('목', '화', '토', '금', '수');
    END IF;
    IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'yin_yang_type') THEN
        CREATE TYPE yin_yang_type AS ENUM ('음', '양');
    END IF;
END $$;
>
CREATE TABLE IF NOT EXISTS ganji_metadata (
    code SMALLINT PRIMARY KEY, -- 천간: 0~9, 지지: 10-21
    name_ko CHAR(1) NOT NULL,
    name_hani CHAR(1) NOT NULL,
    type TEXT CHECK (type IN ('천간', '지지')),
    element element_type NOT NULL,
    yin_yang yin_yang_type NOT NULL
);
>
-- 정적 관계
>
CREATE TABLE IF NOT EXISTS interaction_groups (
    id INTEGER PRIMARY KEY,
    category VARCHAR(20) NOT NULL
);
>
CREATE TABLE IF NOT EXISTS ganji_interaction (
    id SERIAL PRIMARY KEY,
    code SMALLINT REFERENCES ganji_metadata(code) ON DELETE CASCADE,
    group_id INTEGER REFERENCES interaction_groups(id) ON DELETE CASCADE,
    description TEXT
);
>
CREATE TABLE IF NOT EXISTS ganji_interaction (
    id SERIAL PRIMARY KEY,
    group_id INTEGER UNIQUE REFERENCES interaction_groups(id) ON DELETE CASCADE,
    title VARCHAR(100) NOT NULL,
    keyword TEXT[] DEFAULT '{}',
    interpretation TEXT NOT NULL
);
>
-- 사주 팔자
>
CREATE TABLE IF NOT EXISTS saju_profiles (
    id UUID PRIMARY KEY DEFAULT uuidv7(),
    name TEXT NOT NULL,
    gender CHAR(1) CHECK (gender IN ('M', 'F')),
    birth_date TIMESTAMPTZ NOT NULL,
    is_solar BOOLEAN DEFAULT TRUE,
>
    year_gan SMALLINT REFERENCES ganji_metadata(code),
    year_ji SMALLINT REFERENCES ganji_metadata(code),
    month_gan SMALLINT REFERENCES ganji_metadata(code),
    month_ji SMALLINT REFERENCES ganji_metadata(code),
    day_gan SMALLINT REFERENCES ganji_metadata(code),
    dayr_ji SMALLINT REFERENCES ganji_metadata(code),
    hour_gan SMALLINT REFERENCES ganji_metadata(code),
    hour_ji SMALLINT REFERENCES ganji_metadata(code),
>
    analysis_result JSONB DEFAULT '{}',
>
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW()
);
>
CREATE INDEX IF NOT EXISTS idx_saju_birth_date ON saju_profiles(birth_date);
CREATE INDEX IF NOT EXISTS idx_saju_analysis_result ON saju_profiles USING GIN (analysis_result);

파일을 생성하지 않고 터미널에서 직접 쿼리를 날려도 되지만 따로 파일을 작성하는 편이 유지보수 측면에서 효과적이다.

docker

PostgreSQL을 docker 서비스로 실행한다.

~/workspace/saju-analysis$ docker compose up -d
~/workspace/saju-analysis$ docker ps
CONTAINER ID   IMAGE                COMMAND                   CREATED         STATUS         PORTS                                         NAMES
0e23bd848f7a   dpage/pgadmin4       "/entrypoint.sh"          8 seconds ago   Up 8 seconds   0.0.0.0:8080->80/tcp, [::]:8080->80/tcp       pgadmin_ui
42f80ebad4a1   postgres:18-alpine   "docker-entrypoint.s…"   8 seconds ago   Up 8 seconds   0.0.0.0:5432->5432/tcp, [::]:5432->5432/tcp   saju-analysis_postgres

작성해 놓은 데이터베이스 생성 쿼리 파일을 전달한다.

~/workspace/saju-analysis$ docker exec -i saju-analysis_postgres psql -U peter -d saju_db < scripts/init.sql

다음과 같이 데이터베이스 테이블이 생성된 것을 확인할 수 있다.

~/workspace/image-assets-management$ docker exec -it saju-analysis_postgres psql -U peter -d saju_db -c "\dt"
>
                List of tables
 Schema |         Name         | Type  | Owner 
--------+----------------------+-------+-------
 public | ganji_interaction    | table | peter
 public | ganji_metadata       | table | peter
 public | interaction_groups   | table | peter
 public | interaction_metadata | table | peter
 public | saju_profiles        | table | peter
(5 rows)

데이터 삽입

간지 정보는 변하지 않는 값이니 스크립트에 적어 미리 넣어둔다.

~/workspace/saju-analysis$ vi scripts/insert_saju_metadata.sql

scripts/insert_saju_metadata.sql

INSERT INTO ganji_metadata (code, name_ko, name_hani, type, element, yin_yang) VALUES
-- 천간 (0~9)
(0, '갑', '甲', '천간', '목', '양'),
(1, '을', '乙', '천간', '목', '음'),
(2, '병', '丙', '천간', '화', '양'),
(3, '정', '丁', '천간', '화', '음'),
(4, '무', '戊', '천간', '토', '양'),
(5, '기', '己', '천간', '토', '음'),
(6, '경', '庚', '천간', '금', '양'),
(7, '신', '辛', '천간', '금', '음'),
(8, '임', '壬', '천간', '수', '양'),
(9, '계', '癸', '천간', '수', '음'),
-- 지지 (10~21)
(10, '자', '子', '지지', '수', '양'),
(11, '축', '丑', '지지', '토', '음'),
(12, '인', '寅', '지지', '목', '양'),
(13, '묘', '卯', '지지', '목', '음'),
(14, '진', '辰', '지지', '토', '양'),
(15, '사', '巳', '지지', '화', '음'),
(16, '오', '午', '지지', '화', '양'),
(17, '미', '未', '지지', '토', '음'),
(18, '신', '申', '지지', '금', '양'),
(19, '유', '酉', '지지', '금', '음'),
(20, '술', '戌', '지지', '토', '양'),
(21, '해', '亥', '지지', '수', '음');

~/workspace/saju-analysis$ docker exec -i saju-analysis_postgres psql -U peter -d saju_db < scripts/insert_ganji_metadata.sql

관계에 대해서도 마찬가지로 스크립트로 넣어둔다.

~/workspace/saju-analysis$ vi scripts/insert_saju_interaction.sql

scripts/insert_saju_interaction.sql

INSERT INTO interaction_groups (id, category) VALUES
(100, '천간합'), (200, '천간충'),
(300, '지지육합'), (400, '지지삼합'), (500, '지지방합'),
(600, '지지육충'), (700, '지지형'), (800, '지지파'), (900, '지지해');
>
-- 세부 그룹 ID (상속 관계 시각화용)
INSERT INTO interaction_groups (id, category) VALUES
(101, '천간합'), (102, '천간합'), (103, '천간합'), (104, '천간합'), (105, '천간합'),
(201, '천간충'), (202, '천간충'), (203, '천간충'), (204, '천간충'),
(301, '지지육합'), (302, '지지육합'), (303, '지지육합'), (304, '지지육합'), (305, '지지육합'), (306, '지지육합'),
(401, '지지삼합'), (402, '지지삼합'), (403, '지지삼합'), (404, '지지삼합'),
(501, '지지방합'), (502, '지지방합'), (503, '지지방합'), (504, '지지방합'),
(601, '지지육충'), (602, '지지육충'), (603, '지지육충'), (604, '지지육충'), (605, '지지육충'), (606, '지지육충'),
(701, '지지형'), (702, '지지형'), (703, '지지형'),
(801, '지지파'), (802, '지지파'), (803, '지지파'), (804, '지지파'), (805, '지지파'), (806, '지지파'),
(901, '지지해'), (902, '지지해'), (903, '지지해'), (904, '지지해'), (905, '지지해'), (906, '지지해');
>
INSERT INTO ganji_interaction (code, group_id, description) VALUES
-- 천간합
(0, 101, '갑기합토'), (5, 101, '갑기합토'),
(1, 102, '을경합금'), (6, 102, '을경합금'),
(2, 103, '병신합수'), (7, 103, '병신합수'),
(3, 104, '정임합목'), (8, 104, '정임합목'),
(4, 105, '무계합화'), (9, 105, '무계합화'),
-- 천간충
(0, 201, '갑경충'), (6, 201, '갑경충'),
(1, 202, '을신충'), (7, 202, '을신충'),
(2, 203, '병임충'), (8, 203, '병임충'),
(3, 204, '정계충'), (9, 204, '정계충'),
-- 지지육합
(10, 301, '자축합토'), (11, 301, '자축합토'),
(12, 302, '인해합목'), (21, 302, '인해합목'),
(13, 303, '묘술합화'), (20, 303, '묘술합화'),
(14, 304, '진유합금'), (19, 304, '진유합금'),
(15, 305, '사신합수'), (18, 305, '사신합수'),
(16, 306, '오미합화'), (17, 306, '오미합화'),
-- 지지삼합
(21, 401, '해묘미'), (13, 401, '해묘미'), (17, 401, '해묘미'),
(12, 402, '인오술'), (16, 402, '인오술'), (20, 402, '인오술'),
(15, 403, '사유축'), (19, 403, '사유축'), (11, 403, '사유축'),
(18, 404, '신자진'), (10, 404, '신자진'), (14, 404, '신자진'),
-- 지지방합
(12, 501, '인묘진'), (13, 501, '인묘진'), (14, 501, '인묘진'),
(15, 502, '사오미'), (16, 502, '사오미'), (17, 502, '사오미'),
(18, 503, '신유술'), (19, 503, '신유술'), (20, 503, '신유술'),
(21, 504, '해자축'), (10, 504, '해자축'), (11, 504, '해자축'),
-- 지지육충
(10, 601, '자오충'), (16, 601, '자오충'),
(11, 602, '축미충'), (17, 602, '축미충'),
(12, 603, '인신충'), (18, 603, '인신충'),
(13, 604, '묘유충'), (19, 604, '묘유충'),
(14, 605, '진술충'), (20, 605, '진술충'),
(15, 606, '사해충'), (21, 606, '사해충'),
-- 지지형
(12, 701, '인사신 삼형'), (15, 701, '인사신 삼형'), (18, 701, '인사신 삼형'),
(11, 702, '축술미 삼형'), (20, 702, '축술미 삼형'), (17, 702, '축술미 삼형'),
(10, 703, '자묘형'), (13, 703, '자묘형'),
-- 지지파
(10, 801, '자유파'), (19, 801, '자유파'),
(11, 802, '축진파'), (14, 802, '축진파'),
(12, 803, '인해파'), (21, 803, '인해파'),
(13, 804, '묘오파'), (16, 804, '묘오파'),
(15, 805, '사신파'), (18, 805, '사신파'),
(20, 806, '술미파'), (17, 806, '술미파'),
-- 지지해
(10, 901, '자미해'), (17, 901, '자미해'),
(11, 902, '축오해'), (16, 902, '축오해'),
(12, 903, '인사해'), (15, 903, '인사해'),
(13, 904, '묘진해'), (14, 904, '묘진해'),
(18, 905, '신해해'), (21, 905, '신해해'),
(19, 906, '유술해'), (20, 906, '유술해');
>
INSERT INTO interaction_metadata (group_id, title, keyword, interpretation) VALUES
-- 천간합
(101, '갑기합(甲己合): 중정의 합', ARRAY['신뢰', '중심', '안정'], '강직한 갑목과 포용력 있는 기토가 만나 흔들림 없는 중심을 잡고 신뢰를 형성합니다.'),
(102, '을경합(乙庚合): 인의의 합', ARRAY['정의', '결속', '의리'], '유연한 을목과 강직한 경금이 만나 서로를 보완하며 강한 결속력과 의리를 발휘합니다.'),
(103, '병신합(丙辛합): 위제의 합', ARRAY['변화', '냉정', '권위'], '뜨거운 병화와 차가운 신금이 만나 매서운 위엄을 갖추며 새로운 기운으로 변화합니다.'),
(104, '정임합(丁壬合): 인수(淫)의 합', ARRAY['다정', '생산', '예술'], '따뜻한 정화와 넓은 임수가 만나 다정다감한 기운을 만들며 창의적인 생산력을 발휘합니다.'),
(105, '무계합(戊癸合): 무정의 합', ARRAY['냉정', '지혜', '현실'], '무게감 있는 무토와 흐르는 계수가 만나 감정보다는 현실적인 지혜와 효율을 추구합니다.'),
-- 천간충
(201, '갑경충(甲庚沖): 정면충돌', ARRAY['결단', '충격', '변화'], '시작의 기운과 결실의 기운이 부딪혀 삶의 큰 방향 전환과 결단이 필요한 상황을 만듭니다.'),
(202, '을신충(乙辛沖): 날카로운 대립', ARRAY['예민', '스트레스', '섬세'], '부드러운 초목이 날카로운 칼날을 만난 격으로, 신경이 예민해지기 쉬우나 섬세한 감각을 발휘합니다.'),
(203, '병임충(丙壬沖): 수화기제', ARRAY['폭발', '충돌', '창의'], '거대한 빛과 넓은 물이 부딪혀 역동적인 변화를 일으키며 강렬한 에너지를 분출합니다.'),
(204, '정계충(丁癸沖): 감정의 격변', ARRAY['민감', '변화', '갈등'], '작은 불씨와 빗물이 만나듯이 작은 자극에도 감정의 동요가 크고 세심한 조율이 필요합니다.'),
-- 지지육합
(301, '자축합(子丑合): 비밀스러운 결속', ARRAY['협력', '신중', '안정'], '추운 겨울의 기운들이 만나 겉으로 드러나지 않는 끈끈한 유대감과 안정감을 형성합니다.'),
(302, '인해합(寅亥合): 성장의 동력', ARRAY['생동', '시작', '활력'], '큰 물이 나무를 생하며 합하는 구조로, 끊임없는 생명력과 추진력을 만들어냅니다.'),
(303, '묘술합(卯戌合): 실리적인 결합', ARRAY['현실', '안정', '타협'], '부드러운 기운과 마른 땅이 만나 현실적인 이익을 추구하며 안정적인 관계를 맺습니다.'),
(304, '진유합(辰酉合): 견고한 결실', ARRAY['완성', '명예', '단단함'], '습한 기운이 금기를 생하여 더욱 단단하게 만드니 결과물이 명확하고 자부심이 강해집니다.'),
(305, '사신합(巳申合): 변화의 합', ARRAY['복합', '재주', '변동'], '불과 금이 만나 합을 이루나 속에는 형(刑)의 기운이 있어 복잡한 변화와 재주를 발휘합니다.'),
(306, '오미합(午未合): 열정의 결집', ARRAY['뜨거움', '집중', '팽창'], '여름의 절정 기운들이 만나 에너지의 밀도가 매우 높으며 열정적인 결과를 지향합니다.'),
-- 지지삼합
(401, '해묘미(亥卯未) 삼합: 목(木)의 국', ARRAY['성장', '창작', '협동'], '다른 성질의 기운들이 모여 거대한 성장의 기류를 형성하고 공동의 목적을 달성합니다.'),
(402, '인오술(寅午戌) 삼합: 화(火)의 국', ARRAY['발산', '열정', '화려'], '에너지가 외부로 강력하게 분출되며 화려한 성과와 강한 행동력을 보여줍니다.'),
(403, '사유축(巳酉丑) 삼합: 금(金)의 국', ARRAY['결실', '수렴', '엄격'], '단단하게 굳어지는 기운으로 목표 중심적이며 결과에 대한 집요함이 강합니다.'),
(404, '신자진(申子辰) 삼합: 수(水)의 국', ARRAY['지혜', '유연', '깊이'], '유연하게 흐르는 거대한 물줄기를 형성하여 지적 깊이와 탁월한 적응력을 발휘합니다.'),
-- 지지방합
(501, '인묘진(寅卯辰) 방합: 봄의 세력', ARRAY['생동', '추진', '동료'], '계절적 동질성을 가진 강력한 목기가 형성되어 거침없는 시작과 추진력을 보여줍니다.'),
(502, '사오미(巳午未) 방합: 여름의 세력', ARRAY['팽창', '강렬', '확산'], '극강의 화기가 형성되어 자신의 영역을 넓히고 중심에 서려는 욕구가 강해집니다.'),
(503, '신유술(申酉戌) 방합: 가을의 세력', ARRAY['견고', '냉철', '정리'], '날카로운 금기가 모여 불필요한 것을 정리하고 결과물의 밀도를 높이는 힘이 강력합니다.'),
(504, '해자축(亥子丑) 방합: 겨울의 세력', ARRAY['저장', '침잠', '지력'], '차가운 수기가 결집하여 내면의 세계를 탐구하고 지식을 저장하는 힘이 탁월합니다.'),
-- 지지육충
(601, '자오충(子午沖): 자존심의 충돌', ARRAY['폭발', '극단', '예민'], '정반대의 강한 에너지가 부딪혀 감정의 변화가 크고 환경의 변동이 잦을 수 있습니다.'),
(602, '축미충(丑未沖): 터전의 변동', ARRAY['조정', '불화', '이동'], '단단한 흙들이 부딪혀 기반이 흔들리는 형상으로, 환경적인 변화나 조정 과정을 겪습니다.'),
(603, '인신충(寅申沖): 역마의 정면충돌', ARRAY['활동', '변화', '속도'], '시작과 결실의 기운이 부딪혀 매우 활동적이며 거주지나 직업의 변동이 잦을 수 있습니다.'),
(604, '묘유충(卯酉沖): 상처와 대립', ARRAY['민감', '분리', '충격'], '부드러운 나무와 날카로운 칼날이 부딪혀 대인관계의 소외감이나 건강상 주의가 필요합니다.'),
(605, '진술충(辰戌沖): 영적인 충돌', ARRAY['복잡', '변화', '충돌'], '거대한 기운들이 부딪혀 정신적인 혼란이나 예기치 못한 환경 변화를 가져옵니다.'),
(606, '사해충(巳亥沖): 정보와 이동', ARRAY['변동', '마찰', '속도'], '지식과 활동의 기운이 부딪혀 소통 과정의 마찰이나 잦은 이동수가 발생합니다.'),
-- 지지형
(701, '인사신(寅巳申) 삼형: 권력의 제어', ARRAY['조절', '수술', '기술'], '강력한 세 기운이 얽혀 시시비비를 가려야 하니, 전문적인 제어 기술이나 리더십이 요구됩니다.'),
(702, '축술미(丑戌未) 삼형: 무은의 형', ARRAY['배신', '시비', '조율'], '가까운 관계에서의 시비나 배신수가 있을 수 있으니 철저한 문서 관리와 조율이 필요합니다.'),
(703, '자묘형(子卯刑): 무례의 형', ARRAY['구설', '불화', '무례'], '생(生)하는 관계임에도 예의에 어긋나는 상황이 생길 수 있어 대인관계의 주의가 필요합니다.'),
-- 지지파
(801, '자유파(子酉破): 조율의 마찰', ARRAY['분열', '파괴', '수정'], '서로 다른 목적이 부딪혀 진행 중인 일이 일시적으로 중단되거나 수정이 필요할 수 있습니다.'),
(802, '축진파(丑辰破): 내부적 균열', ARRAY['미묘', '불편', '조정'], '비슷한 흙들이 서로를 밀어내며 내부적인 갈등이나 건강상의 미묘한 문제를 일으킵니다.'),
(803, '인해파(寅亥破): 합 뒤의 균열', ARRAY['배신', '마찰', '변심'], '합을 이루었으나 내부에 균열이 있으니 다 된 밥에 코 빠뜨리는 격으로 마무리에 주의해야 합니다.'),
(804, '묘오파(卯午破): 열정의 과부하', ARRAY['손상', '충돌', '과열'], '목기가 화기를 너무 급하게 생하여 에너지가 소진되거나 마찰이 생길 수 있습니다.'),
(805, '사신파(巳申破): 기술적 수정', ARRAY['복잡', '수리', '변동'], '변화가 많은 합 속에서 마찰이 생겨 지속적인 수리와 조정이 필요한 구조입니다.'),
(806, '술미파(戌未破): 기반의 손상', ARRAY['조정', '불편', '변화'], '건조한 흙들이 부딪혀 실질적인 기반이 약해지거나 관계의 소외감이 발생합니다.'),
-- 지지해
(901, '자미해(子未害): 감정의 앙금', ARRAY['질투', '원망', '방해'], '서로의 합을 방해하는 관계라 예기치 못한 원망이나 질투로 인한 마찰이 생기기 쉽습니다.'),
(902, '축오해(丑午害): 열정의 동결', ARRAY['냉담', '방해', '불신'], '뜨거운 열정을 차가운 기운이 방해하니 의욕이 꺾이거나 주변의 시기 질투를 받습니다.'),
(903, '인사해(寅巳害): 소통의 단절', ARRAY['오해', '방해', '충격'], '강한 추진력들이 서로를 방해하여 소통 과정에서 오해가 생기고 진행이 더뎌집니다.'),
(904, '묘진해(卯辰害): 성장판의 위축', ARRAY['침해', '방해', '시기'], '부드러운 목기와 습한 토가 만나 자존심 대결을 벌이며 서로의 성장을 미묘하게 방해합니다.'),
(905, '신해해(申亥害): 결실의 훼손', ARRAY['방해', '변동', '마찰'], '결과를 맺으려는 금기와 시작하려는 수기가 부딪혀 마무리 단계에서 장애가 생길 수 있습니다.'),
(906, '유술해(酉戌해): 날카로운 대립', ARRAY['질투', '마찰', '분열'], '날카로운 결실들이 만나 서로를 시기하니 대인관계의 소외감이나 스트레스에 주의해야 합니다.');

~/workspace/saju-analysis$ docker exec -i saju-analysis_postgres psql -U peter -d saju_db < scripts/insert_ganji_interaction.sql

데이터 확인

데이터가 제대로 들어갔는지 확인한다.

~/workspace/saju-analysis$ docker exec -it saju-analysis_postgres psql -U peter -d saju_db -c "SELECT * FROM ganji_metadata;"             
 code | name_ko | name_hani | type | element | yin_yang 
------+---------+-----------+------+---------+----------
    0 | 갑      | 甲        | 천간 | 목      | 양
    1 | 을      | 乙        | 천간 | 목      | 음
    2 | 병      | 丙        | 천간 | 화      | 양
    3 | 정      | 丁        | 천간 | 화      | 음
    4 | 무      | 戊        | 천간 | 토      | 양
    5 | 기      | 己        | 천간 | 토      | 음
    6 | 경      | 庚        | 천간 | 금      | 양
    7 | 신      | 辛        | 천간 | 금      | 음
    8 | 임      | 壬        | 천간 | 수      | 양
    9 | 계      | 癸        | 천간 | 수      | 음
   10 | 자      | 子        | 지지 | 수      | 양
   11 | 축      | 丑        | 지지 | 토      | 음
   12 | 인      | 寅        | 지지 | 목      | 양
   13 | 묘      | 卯        | 지지 | 목      | 음
   14 | 진      | 辰        | 지지 | 토      | 양
   15 | 사      | 巳        | 지지 | 화      | 음
   16 | 오      | 午        | 지지 | 화      | 양
   17 | 미      | 未        | 지지 | 토      | 음
   18 | 신      | 申        | 지지 | 금      | 양
   19 | 유      | 酉        | 지지 | 금      | 음
   20 | 술      | 戌        | 지지 | 토      | 양
   21 | 해      | 亥        | 지지 | 수      | 음
(22 rows)

~/workspace/saju-analysis$ docker exec -it saju-analysis_postgres psql -U peter -d saju_db -c "SELECT * FROM interaction_groups;"
 id  | category 
-----+----------
 100 | 천간합
 200 | 천간충
 300 | 지지육합
 400 | 지지삼합
 500 | 지지방합
 600 | 지지육충
 700 | 지지형
 800 | 지지파
 900 | 지지해
 101 | 천간합
 102 | 천간합
 103 | 천간합
 104 | 천간합                                           
 105 | 천간합
 201 | 천간충
 202 | 천간충
 203 | 천간충
 204 | 천간충
 301 | 지지육합
 302 | 지지육합
 303 | 지지육합
 304 | 지지육합
 305 | 지지육합
 306 | 지지육합
 401 | 지지삼합
 402 | 지지삼합
 403 | 지지삼합
 404 | 지지삼합
 501 | 지지방합
 502 | 지지방합
 503 | 지지방합
 504 | 지지방합
 601 | 지지육충
 602 | 지지육충
 603 | 지지육충
 604 | 지지육충
 605 | 지지육충
 606 | 지지육충
 701 | 지지형
 702 | 지지형
 703 | 지지형
 801 | 지지파
 802 | 지지파
 803 | 지지파
 804 | 지지파
 805 | 지지파
 806 | 지지파
 901 | 지지해
 902 | 지지해
 903 | 지지해
 904 | 지지해
 905 | 지지해
 906 | 지지해
(53 rows)

~/workspace/saju-analysis$ docker exec -it saju-analysis_postgres psql -U peter -d saju_db -c "SELECT * FROM ganji_interaction;"
 id | code | group_id | description 
----+------+----------+-------------
  1 |    0 |      101 | 갑기합토
  2 |    5 |      101 | 갑기합토
  3 |    1 |      102 | 을경합금
  4 |    6 |      102 | 을경합금
  5 |    2 |      103 | 병신합수
  6 |    7 |      103 | 병신합수
  7 |    3 |      104 | 정임합목
  8 |    8 |      104 | 정임합목
  9 |    4 |      105 | 무계합화
 10 |    9 |      105 | 무계합화
 11 |    0 |      201 | 갑경충
 12 |    6 |      201 | 갑경충
 13 |    1 |      202 | 을신충
 14 |    7 |      202 | 을신충
 15 |    2 |      203 | 병임충
 16 |    8 |      203 | 병임충
 17 |    3 |      204 | 정계충
 18 |    9 |      204 | 정계충
 19 |   10 |      301 | 자축합토
 20 |   11 |      301 | 자축합토
 21 |   12 |      302 | 인해합목
 22 |   21 |      302 | 인해합목
 23 |   13 |      303 | 묘술합화
 24 |   20 |      303 | 묘술합화
 25 |   14 |      304 | 진유합금
 26 |   19 |      304 | 진유합금
 27 |   15 |      305 | 사신합수
 28 |   18 |      305 | 사신합수
 29 |   16 |      306 | 오미합화
 30 |   17 |      306 | 오미합화
 31 |   21 |      401 | 해묘미
 32 |   13 |      401 | 해묘미
 33 |   17 |      401 | 해묘미
 34 |   12 |      402 | 인오술
 35 |   16 |      402 | 인오술
 36 |   20 |      402 | 인오술
 37 |   15 |      403 | 사유축
 38 |   19 |      403 | 사유축
 39 |   11 |      403 | 사유축
 40 |   18 |      404 | 신자진
 41 |   10 |      404 | 신자진
 42 |   14 |      404 | 신자진
 43 |   12 |      501 | 인묘진
 44 |   13 |      501 | 인묘진
 45 |   14 |      501 | 인묘진
 46 |   15 |      502 | 사오미
 47 |   16 |      502 | 사오미
 48 |   17 |      502 | 사오미
 49 |   18 |      503 | 신유술
 50 |   19 |      503 | 신유술
 51 |   20 |      503 | 신유술
 52 |   21 |      504 | 해자축
 53 |   10 |      504 | 해자축
 54 |   11 |      504 | 해자축
 55 |   10 |      601 | 자오충
 56 |   16 |      601 | 자오충
 57 |   11 |      602 | 축미충
 58 |   17 |      602 | 축미충
 59 |   12 |      603 | 인신충
 60 |   18 |      603 | 인신충
 61 |   13 |      604 | 묘유충
 62 |   19 |      604 | 묘유충
 63 |   14 |      605 | 진술충
 64 |   20 |      605 | 진술충
 65 |   15 |      606 | 사해충
 66 |   21 |      606 | 사해충
 67 |   12 |      701 | 인사신 삼형
 68 |   15 |      701 | 인사신 삼형
 69 |   18 |      701 | 인사신 삼형
 70 |   11 |      702 | 축술미 삼형
 71 |   20 |      702 | 축술미 삼형
 72 |   17 |      702 | 축술미 삼형
 73 |   10 |      703 | 자묘형
 74 |   13 |      703 | 자묘형
 75 |   10 |      801 | 자유파
 76 |   19 |      801 | 자유파
 77 |   11 |      802 | 축진파
 78 |   14 |      802 | 축진파
 79 |   12 |      803 | 인해파
 80 |   21 |      803 | 인해파
 81 |   13 |      804 | 묘오파
 82 |   16 |      804 | 묘오파
 83 |   15 |      805 | 사신파
 84 |   18 |      805 | 사신파
 85 |   20 |      806 | 술미파
 86 |   17 |      806 | 술미파
 87 |   10 |      901 | 자미해
 88 |   17 |      901 | 자미해
 89 |   11 |      902 | 축오해
 90 |   16 |      902 | 축오해
 91 |   12 |      903 | 인사해
 92 |   15 |      903 | 인사해
 93 |   13 |      904 | 묘진해
 94 |   14 |      904 | 묘진해
 95 |   18 |      905 | 신해해
 96 |   21 |      905 | 신해해
 97 |   19 |      906 | 유술해
 98 |   20 |      906 | 유술해
(98 rows)

~/workspace/saju-analysis$ docker exec -it saju-analysis_postgres psql -U peter -d saju_db -c "SELECT * FROM interaction_metadata;"
 id | group_id |              title               |       keyword        |                                         interpretation                                          
----+----------+----------------------------------+----------------------+-------------------------------------------------------------------------------------------------
  1 |      101 | 갑기합(甲己合): 중정의 합        | {신뢰,중심,안정}     | 강직한 갑목과 포용력 있는 기토가 만나 흔들림 없는 중심을 잡고 신뢰를 형성합니다.
  2 |      102 | 을경합(乙庚合): 인의의 합        | {정의,결속,의리}     | 유연한 을목과 강직한 경금이 만나 서로를 보완하며 강한 결속력과 의리를 발휘합니다.
  3 |      103 | 병신합(丙辛합): 위제의 합        | {변화,냉정,권위}     | 뜨거운 병화와 차가운 신금이 만나 매서운 위엄을 갖추며 새로운 기운으로 변화합니다.
  4 |      104 | 정임합(丁壬合): 인수(淫)의 합    | {다정,생산,예술}     | 따뜻한 정화와 넓은 임수가 만나 다정다감한 기운을 만들며 창의적인 생산력을 발휘합니다.
  5 |      105 | 무계합(戊癸合): 무정의 합        | {냉정,지혜,현실}     | 무게감 있는 무토와 흐르는 계수가 만나 감정보다는 현실적인 지혜와 효율을 추구합니다.
  6 |      201 | 갑경충(甲庚沖): 정면충돌         | {결단,충격,변화}     | 시작의 기운과 결실의 기운이 부딪혀 삶의 큰 방향 전환과 결단이 필요한 상황을 만듭니다.
  7 |      202 | 을신충(乙辛沖): 날카로운 대립    | {예민,스트레스,섬세} | 부드러운 초목이 날카로운 칼날을 만난 격으로, 신경이 예민해지기 쉬우나 섬세한 감각을 발휘합니다.
  8 |      203 | 병임충(丙壬沖): 수화기제         | {폭발,충돌,창의}     | 거대한 빛과 넓은 물이 부딪혀 역동적인 변화를 일으키며 강렬한 에너지를 분출합니다.
  9 |      204 | 정계충(丁癸沖): 감정의 격변      | {민감,변화,갈등}     | 작은 불씨와 빗물이 만나듯이 작은 자극에도 감정의 동요가 크고 세심한 조율이 필요합니다.
 10 |      301 | 자축합(子丑合): 비밀스러운 결속  | {협력,신중,안정}     | 추운 겨울의 기운들이 만나 겉으로 드러나지 않는 끈끈한 유대감과 안정감을 형성합니다.
 11 |      302 | 인해합(寅亥合): 성장의 동력      | {생동,시작,활력}     | 큰 물이 나무를 생하며 합하는 구조로, 끊임없는 생명력과 추진력을 만들어냅니다.
 12 |      303 | 묘술합(卯戌合): 실리적인 결합    | {현실,안정,타협}     | 부드러운 기운과 마른 땅이 만나 현실적인 이익을 추구하며 안정적인 관계를 맺습니다.
 13 |      304 | 진유합(辰酉合): 견고한 결실      | {완성,명예,단단함}   | 습한 기운이 금기를 생하여 더욱 단단하게 만드니 결과물이 명확하고 자부심이 강해집니다.
 14 |      305 | 사신합(巳申合): 변화의 합        | {복합,재주,변동}     | 불과 금이 만나 합을 이루나 속에는 형(刑)의 기운이 있어 복잡한 변화와 재주를 발휘합니다.
 15 |      306 | 오미합(午未合): 열정의 결집      | {뜨거움,집중,팽창}   | 여름의 절정 기운들이 만나 에너지의 밀도가 매우 높으며 열정적인 결과를 지향합니다.
 16 |      401 | 해묘미(亥卯未) 삼합: 목(木)의 국 | {성장,창작,협동}     | 다른 성질의 기운들이 모여 거대한 성장의 기류를 형성하고 공동의 목적을 달성합니다.
 17 |      402 | 인오술(寅午戌) 삼합: 화(火)의 국 | {발산,열정,화려}     | 에너지가 외부로 강력하게 분출되며 화려한 성과와 강한 행동력을 보여줍니다.
 18 |      403 | 사유축(巳酉丑) 삼합: 금(金)의 국 | {결실,수렴,엄격}     | 단단하게 굳어지는 기운으로 목표 중심적이며 결과에 대한 집요함이 강합니다.
 19 |      404 | 신자진(申子辰) 삼합: 수(水)의 국 | {지혜,유연,깊이}     | 유연하게 흐르는 거대한 물줄기를 형성하여 지적 깊이와 탁월한 적응력을 발휘합니다.
 20 |      501 | 인묘진(寅卯辰) 방합: 봄의 세력   | {생동,추진,동료}     | 계절적 동질성을 가진 강력한 목기가 형성되어 거침없는 시작과 추진력을 보여줍니다.
 21 |      502 | 사오미(巳午未) 방합: 여름의 세력 | {팽창,강렬,확산}     | 극강의 화기가 형성되어 자신의 영역을 넓히고 중심에 서려는 욕구가 강해집니다.
 22 |      503 | 신유술(申酉戌) 방합: 가을의 세력 | {견고,냉철,정리}     | 날카로운 금기가 모여 불필요한 것을 정리하고 결과물의 밀도를 높이는 힘이 강력합니다.
 23 |      504 | 해자축(亥子丑) 방합: 겨울의 세력 | {저장,침잠,지력}     | 차가운 수기가 결집하여 내면의 세계를 탐구하고 지식을 저장하는 힘이 탁월합니다.
 24 |      601 | 자오충(子午沖): 자존심의 충돌    | {폭발,극단,예민}     | 정반대의 강한 에너지가 부딪혀 감정의 변화가 크고 환경의 변동이 잦을 수 있습니다.
 25 |      602 | 축미충(丑未沖): 터전의 변동      | {조정,불화,이동}     | 단단한 흙들이 부딪혀 기반이 흔들리는 형상으로, 환경적인 변화나 조정 과정을 겪습니다.
 26 |      603 | 인신충(寅申沖): 역마의 정면충돌  | {활동,변화,속도}     | 시작과 결실의 기운이 부딪혀 매우 활동적이며 거주지나 직업의 변동이 잦을 수 있습니다.
 27 |      604 | 묘유충(卯酉沖): 상처와 대립      | {민감,분리,충격}     | 부드러운 나무와 날카로운 칼날이 부딪혀 대인관계의 소외감이나 건강상 주의가 필요합니다.
 28 |      605 | 진술충(辰戌沖): 영적인 충돌      | {복잡,변화,충돌}     | 거대한 기운들이 부딪혀 정신적인 혼란이나 예기치 못한 환경 변화를 가져옵니다.
 29 |      606 | 사해충(巳亥沖): 정보와 이동      | {변동,마찰,속도}     | 지식과 활동의 기운이 부딪혀 소통 과정의 마찰이나 잦은 이동수가 발생합니다.
 30 |      701 | 인사신(寅巳申) 삼형: 권력의 제어 | {조절,수술,기술}     | 강력한 세 기운이 얽혀 시시비비를 가려야 하니, 전문적인 제어 기술이나 리더십이 요구됩니다.
 31 |      702 | 축술미(丑戌未) 삼형: 무은의 형   | {배신,시비,조율}     | 가까운 관계에서의 시비나 배신수가 있을 수 있으니 철저한 문서 관리와 조율이 필요합니다.
 32 |      703 | 자묘형(子卯刑): 무례의 형        | {구설,불화,무례}     | 생(生)하는 관계임에도 예의에 어긋나는 상황이 생길 수 있어 대인관계의 주의가 필요합니다.
 33 |      801 | 자유파(子酉破): 조율의 마찰      | {분열,파괴,수정}     | 서로 다른 목적이 부딪혀 진행 중인 일이 일시적으로 중단되거나 수정이 필요할 수 있습니다.
 34 |      802 | 축진파(丑辰破): 내부적 균열      | {미묘,불편,조정}     | 비슷한 흙들이 서로를 밀어내며 내부적인 갈등이나 건강상의 미묘한 문제를 일으킵니다.
 35 |      803 | 인해파(寅亥破): 합 뒤의 균열     | {배신,마찰,변심}     | 합을 이루었으나 내부에 균열이 있으니 다 된 밥에 코 빠뜨리는 격으로 마무리에 주의해야 합니다.
 36 |      804 | 묘오파(卯午破): 열정의 과부하    | {손상,충돌,과열}     | 목기가 화기를 너무 급하게 생하여 에너지가 소진되거나 마찰이 생길 수 있습니다.
 37 |      805 | 사신파(巳申破): 기술적 수정      | {복잡,수리,변동}     | 변화가 많은 합 속에서 마찰이 생겨 지속적인 수리와 조정이 필요한 구조입니다.
 38 |      806 | 술미파(戌未破): 기반의 손상      | {조정,불편,변화}     | 건조한 흙들이 부딪혀 실질적인 기반이 약해지거나 관계의 소외감이 발생합니다.
 39 |      901 | 자미해(子未害): 감정의 앙금      | {질투,원망,방해}     | 서로의 합을 방해하는 관계라 예기치 못한 원망이나 질투로 인한 마찰이 생기기 쉽습니다.
 40 |      902 | 축오해(丑午害): 열정의 동결      | {냉담,방해,불신}     | 뜨거운 열정을 차가운 기운이 방해하니 의욕이 꺾이거나 주변의 시기 질투를 받습니다.
 41 |      903 | 인사해(寅巳害): 소통의 단절      | {오해,방해,충격}     | 강한 추진력들이 서로를 방해하여 소통 과정에서 오해가 생기고 진행이 더뎌집니다.
 42 |      904 | 묘진해(卯辰害): 성장판의 위축    | {침해,방해,시기}     | 부드러운 목기와 습한 토가 만나 자존심 대결을 벌이며 서로의 성장을 미묘하게 방해합니다.
 43 |      905 | 신해해(申亥害): 결실의 훼손      | {방해,변동,마찰}     | 결과를 맺으려는 금기와 시작하려는 수기가 부딪혀 마무리 단계에서 장애가 생길 수 있습니다.
 44 |      906 | 유술해(酉戌해): 날카로운 대립    | {질투,마찰,분열}     | 날카로운 결실들이 만나 서로를 시기하니 대인관계의 소외감이나 스트레스에 주의해야 합니다.
(44 rows)

이후, 이 데이터를 조합하여 분석하는 Rust 로직을 구현해 보자.

데이터베이스에 담긴 자료를 삭제하는 방식은 두 가지가 있다. DELETE 를 사용하여 완전히 삭제해서 복구할 수 없게 되는 Hard Delete, 그리고 "삭제됨"을 표시하고 조회 대상에서 제외하여 복구 가능한 Soft Delete.

여기서는 후자를 알아보도록 하겠다.

DB

삭제 여부를 나타내는 칼럼을 추가해야 한다. 단순히 기다 아니다를 넘어 언제 삭제했는지 알 수 있도록 삭제 일자 칼럼을 추가하도록 하겠다. 이 값이 NULL 이면 아직 삭제되지 않은 것이다.

스키마 변경

칼럼 추가를 위해 다음과 같이 DB 스키마를 변경한다.

~/workspace/image-assets-management$ docker exec -it image_assets_management_postgres psql -U peter -d image_db -c "ALTER TABLE image_assets ADD COLUMN deleted_at TIMESTAMPTZ DEFAULT NULL;"

변경 후 확인해 보면,

~/workspace/image-assets-management$ docker exec -it image_assets_management_postgres psql -U peter -d image_db -c "SELECT         
    column_name, 
    data_type, 
    is_nullable, 
    column_default
FROM 
    information_schema.columns
WHERE 
    table_name = 'image_assets';"
 column_name |        data_type         | is_nullable |  column_default   
-------------+--------------------------+-------------+-------------------
 deleted_at  | timestamp with time zone | YES         | 
 created_at  | timestamp with time zone | YES         | CURRENT_TIMESTAMP
 id          | uuid                     | NO          | uuidv7()
 width       | integer                  | NO          | 
 height      | integer                  | NO          | 
 metadata    | jsonb                    | YES         | 
 file_name   | text                     | NO          | 
 file_path   | text                     | NO          | 
(8 rows)

변경된 스키마 반영

이에 따라 Rust 코드에 작성된 구조체도 수정한다. Python에서 이 값을 직접 사용하지는 않을 것이므로 getter 를 추가하지는 않아도 되지만 sqlx를 사용하는 코드는 구조체에 맞게 수정해야 한다.

deleted_at 값을 변경하지 않고 쿼리로 직접 넣어줄 경우 이에 대한 구조체 필드가 당장은 필요하지 않을 수 있지만 "삭제한 지 한 달이 넘으면 영구 삭제" 같은 로직 구현 시 필요하므로 확장성을 고려해 추가해 두는 게 좋다.

조회 함수는 deleted_at 이 NULL 인 값만 조회하도록 수정한다.

query_at! 매크로를 사용할 때는 RETURNING 하는 값의 구성이 데이터를 저장하는 구조체와 일치해야 한다는 것을 유의하자.

src/image.rs

use pyo3::prelude::*;
use pyo3::exceptions::PyRuntimeError;
use sqlx::{Pool, Postgres};
use uuid::Uuid;
use serde_json::Value as JsonValue;
use tracing::{info, instrument, error};
>
#[pyclass(dict, from_py_object)]
#[derive(Clone, serde::Serialize, serde::Deserialize, sqlx::FromRow)]
pub struct ImageAsset {
    pub id: Uuid,
    #[pyo3(get)]
    pub file_name: String,
    #[pyo3(get)]
    pub file_path: String,
    #[pyo3(get)]
    pub width: i32,
    #[pyo3(get)]
    pub height: i32,
    pub metadata: JsonValue,
    pub deleted_at: Option<chrono::DateTime<chrono::Utc>>, // NEW!
}
>
# 중략
>
#[instrument(skip(pool))]
pub async fn search_assets(
    pool: &Pool<Postgres>,
    name_query: Option<String>,
    format_query: Option<String>,
    limit: i64,
    offset: i64
) -> Result<AssetListResponse, sqlx::Error> {
    let rows = sqlx::query!(
        r#"
        SELECT id, file_name, file_path, width, height, metadata, deleted_at, count(*) OVER() as "total_count!" -- MODIFIED!
        FROM image_assets
        WHERE ($1::TEXT IS NULL OR file_name ILIKE $1)
            AND ($2::JSONB IS NULL OR metadata @> $2)
            AND deleted_at IS NULL
        ORDER BY id DESC
        LIMIT $3 OFFSET $4
        "#,
        name_query.map(|n| format!("%{}%", n)),
        format_query.map(|f| serde_json::json!({"format": f})),
        limit,
        offset
    )
    .fetch_all(pool)
    .await?;
>
    let total_count = rows.first().map(|r| r.total_count).unwrap_or(0);
    let items = rows.into_iter().map(|r| ImageAsset {
        id: r.id,
        file_name: r.file_name,
        file_path: r.file_path,
        width: r.width,
        height: r.height,
        metadata: r.metadata.unwrap_or(serde_json::Value::Null),
        deleted_at: r.deleted_at, // NEW!
    }).collect();
>
    let size = limit as i32;
    let page = (offset as i32 / size) + 1;
>
    Ok(AssetListResponse{ total_count, items, page, size })
}
>
#[instrument(skip(pool))]
pub async fn update_metadata(
    pool: &Pool<Postgres>,
    id: uuid::Uuid,
    new_meta: serde_json::Value,
) -> Result<ImageAsset, sqlx::Error> {
    let updated = sqlx::query_as!(
        ImageAsset,
        r#"
        UPDATE image_assets
        SET metadata = metadata || $1
        WHERE id = $2 AND deleted_at IS NULL
        RETURNING id, file_name, file_path, width, height, metadata, deleted_at -- MODIFIED!
        "#,
        new_meta,
        id
    )
    .fetch_one(pool)
    .await?;
>
    Ok(updated)
}

어차피 기본값으로 생성할 거라 필요하지는 않지만 이미지 에셋 데이터를 처음 생성하는 부분에도 새로 추가된 필드의 값을 명확히 설정해 주어야 오류가 나지 않는다.

src/lib.rs

// 전략
>
#[pyfunction]
#[instrument(skip(metadata))]
fn add_image_asset(
    file_name: String,
    file_path: String,
    width: i32,
    height: i32,
    metadata: String
) -> PyResult<String> {
    let rt = get_runtime();
>
    let meta_json: serde_json::Value = serde_json::from_str(&metadata)
        .map_err(|e| pyo3::exceptions::PyValueError::new_err(e.to_string()))?;
>
    let asset = image::ImageAsset {
        id: uuid::Uuid::nil(),
        file_name,
        file_path,
        width,
        height,
        metadata: meta_json,
        deleted_at: None, // NEW!
    };
>
    let pool = db::DB_POOL.get()
        .ok_or(pyo3::exceptions::PyRuntimeError::new_err("DB 연결 풀이 없습니다."))?;
>
    let id = rt.block_on(async {
        image::insert_asset(pool, asset).await
    }).map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))?;
>
    Ok(id.to_string())
}
> 
// 후략

코드

Rust 코드

어떤 이미지 에셋의 deleted_at 값을 설정하여 Soft Delete를 수행하는 코드를 작성한다.

DB에서 완전히 제거된 게 아니므로 다시 복구할 수 있다. 따라서 복구하는 코드도 함께 작성하겠다.

src/image.rs

use pyo3::prelude::*;
use pyo3::exceptions::PyRuntimeError;
use sqlx::{Pool, Postgres};
use uuid::Uuid;
use serde_json::Value as JsonValue;
use tracing::{info, instrument, error};
>
// 기존 함수 생략
>
#[instrument(skip(pool))]
pub async fn soft_delete_asset(
    pool: &Pool<Postgres>,
    id: uuid::Uuid,
) -> Result<bool, sqlx::Error> {
    let result = sqlx::query!(
        r#"
        UPDATE image_assets
        SET deleted_at = NOW()
        WHERE id = $1 AND deleted_at IS NULL
        "#,
        id
    )
    .execute(pool)
    .await?;
>
    Ok(result.rows_affected() > 0)
}
>
#[instrument(skip(pool))]
pub async fn restore_asset(
    pool: &Pool<Postgres>,
    id: uuid::Uuid,
) -> Result<bool, sqlx::Error> {
    let result = sqlx::query!(
        r#"
        UPDATE image_assets
        SET deleted_at = NULL
        WHERE id = $1 AND deleted_at IS NOT NULL
        "#,
        id
    )
    .execute(pool)
    .await?;
>
    Ok(result.rows_affected() > 0)
}

src/lib.rs

mod db;
mod logger;
mod image;
>
use pyo3::prelude::*;
use tokio::runtime::Runtime;
use dotenvy::dotenv;
use std::env;
use std::sync::OnceLock;
use tracing::{debug, error, info, instrument};
use tracing_appender::non_blocking::WorkerGuard;
>
static LOG_GUARD: OnceLock<WorkerGuard> = OnceLock::new();
static TOKIO_RUNTIME: OnceLock<Runtime> = OnceLock::new();
>
fn get_runtime() -> &'static Runtime {
    TOKIO_RUNTIME.get_or_init(|| {
        Runtime::new().expect("Tokio 런타임 생성 실패")
    })
}
>
// 기존 함수 생략
>
#[pyfunction]
fn delete_asset(id_str: String) -> PyResult<bool> {
    let rt = get_runtime();
    let pool = db::DB_POOL.get()
        .ok_or(pyo3::exceptions::PyRuntimeError::new_err("DB 연결 풀이 없습니다."))?;
>
    let id = uuid::Uuid::parse_str(&id_str)
        .map_err(|e| pyo3::exceptions::PyValueError::new_err(e.to_string()))?;
>
    let success = rt.block_on(async {
        image::soft_delete_asset(pool, id).await
    }).map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))?;
>
    Ok(success)
}
>
#[pyfunction]
fn restore_asset(id_str: String) -> PyResult<bool> {
    let rt = get_runtime();
    let pool = db::DB_POOL.get()
        .ok_or(pyo3::exceptions::PyRuntimeError::new_err("DB 연결 풀이 없습니다."))?;
>
    let id = uuid::Uuid::parse_str(&id_str)
        .map_err(|e| pyo3::exceptions::PyValueError::new_err(e.to_string()))?;
>
    let success = rt.block_on(async {
        image::restore_asset(pool, id).await
    }).map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))?;
>
    Ok(success)
}
>
#[pymodule]
fn rust_engine(m: &Bound<'_, PyModule>) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(init_engine, m)?)?;
    m.add_function(wrap_pyfunction!(shutdown_engine, m)?)?;
    m.add_function(wrap_pyfunction!(check_connection, m)?)?;
    m.add_function(wrap_pyfunction!(add_image_asset, m)?)?;
    m.add_function(wrap_pyfunction!(search_assets, m)?)?;
    m.add_function(wrap_pyfunction!(patch_asset_metadata, m)?)?;
    m.add_function(wrap_pyfunction!(delete_asset, m)?)?;
    m.add_function(wrap_pyfunction!(restore_asset, m)?)?;
>
    Ok(())
}

Python 코드

삭제하고자 하는 에셋 아이디가 존재하지 않거나 이미 삭제되어 있어서 삭제 연산이 이루어지지 않은 경우에는 404 오류를 반환하도록 구현한다. 복구할 때도 마찬가지로 처리할 수 없다며 400 오류를 반환하도록 한다.

app/main.py

from fastapi import FastAPI, HTTPException
from fastapi.responses import ORJSONResponse
from contextlib import asynccontextmanager
import rust_engine
import logging
import json
from pydantic import BaseModel
>
# 기존 함수 생략
>
@app.delete("/assets/{asset_id}")
async def delete_asset(asset_id: str):
    success = rust_engine.delete_asset(asset_id)
>
    if not success:
        raise HTTPException(
            status_code=404,
            detail="존재하지 않거나 이미 삭제되었습니다."
        )
>
    return {
        "status": "success",
        "message": f"에셋 {asset_id} 삭제 완료"
    }
>
@app.post("/assets/{asset_id}/restore")
async def restore_asset(asset_id: str):
    success = rust_engine.restore_asset(asset_id)
>
    if not success:
        raise HTTPException(
            status_code=400,
            detail="삭제되지 않았거나 없는 파일입니다."
        )
>
    return {
        "status": "success",
        "message": f"에셋 {asset_id} 복구 완료"
    }

빌드 및 실행

Maturin 라이브러리를 통해 Rust 코드를 Python에서 호출 가능한 형태로 컴파일한다. pyproject.toml 파일에 build-backend = "maturin" 이 명시되어 있으니 uv를 사용하는 방식으로 컴파일하겠다.

uvicorn 라이브러리를 통해 FastAPI를 실행한다.

~/workspace/image-assets-management$ uv pip install -e .
~/workspace/image-assets-management$ uv run uvicorn app.main:app --reload

$ curl -i "http://127.0.0.1:8000/assets/search?name_query=hero"
HTTP/1.1 200 OK
date: Tue, 28 Apr 2026 01:18:21 GMT
server: uvicorn
content-length: 739
content-type: application/json; charset=utf-8
>
{"total_count":3,"items":[{"id":"019d9441-cb91-79bb-9bb5-4cd6326f0e68","file_name":"char_hero_run.exr","file_path":"/assets/textures/hero/","width":4096,"height":4096,"metadata":{"format":"EXR","layers":64,"type":"render"},"deleted_at":null},{"id":"019d9441-cb87-7f42-b228-2e3590d90a05","file_name":"char_hero_idle.exr","file_path":"/assets/textures/hero/","width":4096,"height":4096,"metadata":{"format":"EXR","layers":64,"type":"render"},"deleted_at":null},{"id":"019d8eae-5943-76c4-9a85-195ce2c4f81a","file_name":"hero_char_diffuse.exr","file_path":"/assets/textures/hero/","width":4096,"height":4096,"metadata":{"author":"peter","avg_luminance":0.45,"format":"EXR","is_verified":true,"layers":32},"deleted_at":null}],"page":1,"size":5}%  

$ curl -iX DELETE "http://127.0.0.1:8000/assets/019d9441-cb91-79bb-9bb5-4cd6326f0e68"
HTTP/1.1 200 OK
date: Tue, 28 Apr 2026 01:19:11 GMT
server: uvicorn
content-length: 90
content-type: application/json; charset=utf-8
>
{"status":"success","message":"에셋 019d9441-cb91-79bb-9bb5-4cd6326f0e68 삭제 완료"}%

갯수가 줄었다.

$ curl -i "http://127.0.0.1:8000/assets/search?name_query=hero"
HTTP/1.1 200 OK
date: Tue, 28 Apr 2026 01:19:14 GMT
server: uvicorn
content-length: 523
content-type: application/json; charset=utf-8
>
{"total_count":2,"items":[{"id":"019d9441-cb87-7f42-b228-2e3590d90a05","file_name":"char_hero_idle.exr","file_path":"/assets/textures/hero/","width":4096,"height":4096,"metadata":{"format":"EXR","layers":64,"type":"render"},"deleted_at":null},{"id":"019d8eae-5943-76c4-9a85-195ce2c4f81a","file_name":"hero_char_diffuse.exr","file_path":"/assets/textures/hero/","width":4096,"height":4096,"metadata":{"author":"peter","avg_luminance":0.45,"format":"EXR","is_verified":true,"layers":32},"deleted_at":null}],"page":1,"size":5}% 

다시 시도하면 없다고 뜬다.

$ curl -iX DELETE "http://127.0.0.1:8000/assets/019d9441-cb91-79bb-9bb5-4cd6326f0e68"
HTTP/1.1 404 Not Found
date: Tue, 28 Apr 2026 01:19:24 GMT
server: uvicorn
content-length: 47
content-type: application/json
>
{"detail":""존재하지 않거나 이미 삭제되었습니다."}%  

$ curl -iX POST "http://127.0.0.1:8000/assets/019d9441-cb91-79bb-9bb5-4cd6326f0e68/restore"
HTTP/1.1 200 OK
date: Tue, 28 Apr 2026 01:44:26 GMT
server: uvicorn
content-length: 90
content-type: application/json; charset=utf-8
>
{"status":"success","message":"에셋 019d9441-cb91-79bb-9bb5-4cd6326f0e68 복구 완료"}%

갯수가 늘었다.

$ curl -i "http://127.0.0.1:8000/assets/search?name_query=hero"                      
HTTP/1.1 200 OK
date: Tue, 28 Apr 2026 01:44:33 GMT
server: uvicorn
content-length: 739
content-type: application/json; charset=utf-8
>
{"total_count":3,"items":[{"id":"019d9441-cb91-79bb-9bb5-4cd6326f0e68","file_name":"char_hero_run.exr","file_path":"/assets/textures/hero/","width":4096,"height":4096,"metadata":{"format":"EXR","layers":64,"type":"render"},"deleted_at":null},{"id":"019d9441-cb87-7f42-b228-2e3590d90a05","file_name":"char_hero_idle.exr","file_path":"/assets/textures/hero/","width":4096,"height":4096,"metadata":{"format":"EXR","layers":64,"type":"render"},"deleted_at":null},{"id":"019d8eae-5943-76c4-9a85-195ce2c4f81a","file_name":"hero_char_diffuse.exr","file_path":"/assets/textures/hero/","width":4096,"height":4096,"metadata":{"author":"peter","avg_luminance":0.45,"format":"EXR","is_verified":true,"layers":32},"deleted_at":null}],"page":1,"size":5}%  

다시 하면 안 된다고 뜬다.

$ curl -iX POST "http://127.0.0.1:8000/assets/019d9441-cb91-79bb-9bb5-4cd6326f0e68/restore"
HTTP/1.1 400 Bad Request
date: Tue, 28 Apr 2026 01:45:31 GMT
server: uvicorn
content-length: 62
content-type: application/json
>
{"detail":"삭제되지 않았거나 없는 파일입니다."}%

비정형 데이터 JSON의 내용을 수정하는 방법을 알아보자.

에셋의 모든 정보를 바꾸기보다, 특정 메타데이터만 추가하거나 수정하는 일이 훨씬 많다. PostgreSQL의 || (병합) 연산자를 사용하면 기존 JSON 데이터를 유지하면서 특정 키만 덮어쓰거나 추가할 수 있다.

최상위 키만 비교하기 때문에 중첩된 데이터가 있을 경우 일부 데이터가 누락될 수 있는데 이를 방지하려면 jsonb_set 함수를 중첩으로 사용해야 하지만 일반적인 태그 추가나 상태 변경은 || 로도 충분하다.

주의! 기존 메타데이터가 {"info": {"color": "red"}} 인데 {"info": {"size": 10}} 을 병합하면 info 내부가 합쳐지는 게 아니라 info 전체가 새 객체로 덮어씌워짐.

코드 작성

Rust 코드

sqlx::query_as! 를 사용하면 두 번째 인자로 전달된 쿼리의 결과를 첫 번째 인자로 전달된 변수에 바로 대입할 수 있다. 세 번째 인자부터는 sqlx::query! 와 마찬가지로 쿼리에 대입되는 값이다. UPDATE 연산은 RETURNING 을 통해 결과를 반환한다.

src/image.rs

use pyo3::prelude::*;
use pyo3::exceptions::PyRuntimeError;
use sqlx::{Pool, Postgres};
use uuid::Uuid;
use serde_json::Value as JsonValue;
use tracing::{info, instrument, error};
>
#[pyclass(dict)]
#[derive(Clone, serde::Serialize, serde::Deserialize, sqlx::FromRow)]
pub struct ImageAsset {
    pub id: Uuid,
    #[pyo3(get)]
    pub file_name: String,
    #[pyo3(get)]
    pub file_path: String,
    #[pyo3(get)]
    pub width: i32,
    #[pyo3(get)]
    pub height: i32,
    pub metadata: JsonValue,
}
>
#[pymethods]
impl ImageAsset {
    #[getter]
    fn id(&self) -> String {
        self.id.to_string()
    }
>
    #[getter]
    fn metadata<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
        pythonize::pythonize(py, &self.metadata)
            .map_err(|e| PyRuntimeError::new_err(e.to_string()))
    }
>
    fn to_dict<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
        pythonize::pythonize(py, self)
            .map_err(|e| PyRuntimeError::new_err(e.to_string()))
    }
}
>
// 기존 함수 생략
>
#[instrument(skip(pool))]
pub async fn update_metadata(
    pool: &Pool<Postgres>,
    id: uuid::Uuid,
    new_meta: serde_json::Value,
) -> Result<ImageAsset, sqlx::Error> {
    let updated = sqlx::query_as!(
        ImageAsset,
        r#"
        UPDATE image_assets
        SET metadata = metadata || $1
        WHERE id = $2
        RETURNING id, file_name, file_path, width, height, metadata
        "#,
        new_meta,
        id
    )
    .fetch_one(pool)
    .await?;
>
    Ok(updated)
}

src/lib.rs

mod db;
mod logger;
mod image;
>
use pyo3::prelude::*;
use tokio::runtime::Runtime;
use dotenvy::dotenv;
use std::env;
use std::sync::OnceLock;
use tracing::{debug, error, info, instrument};
use tracing_appender::non_blocking::WorkerGuard;
>
static LOG_GUARD: OnceLock<WorkerGuard> = OnceLock::new();
static TOKIO_RUNTIME: OnceLock<Runtime> = OnceLock::new();
>
fn get_runtime() -> &'static Runtime {
    TOKIO_RUNTIME.get_or_init(|| {
        Runtime::new().expect("Tokio 런타임 생성 실패")
    })
}
>
// 기존 함수 생략
>
#[pyfunction]
fn patch_asset_metadata(
    id_str: String,
    metadata_json: String
) -> PyResult<image::ImageAsset> {
    let rt = get_runtime();
    let pool = db::DB_POOL.get()
        .ok_or(pyo3::exceptions::PyRuntimeError::new_err("DB 연결 풀이 없습니다."))?;
>
    let id = uuid::Uuid::parse_str(&id_str)
        .map_err(|e| pyo3::exceptions::PyValueError::new_err(e.to_string()))?;
>
    let metadata: serde_json::Value = serde_json::from_str(&metadata_json)
        .map_err(|e| pyo3::exceptions::PyValueError::new_err(e.to_string()))?;
>
    let asset = rt.block_on(async {
        image::update_metadata(pool, id, metadata).await
    }).map_err(|e| pyo3::exceptions::PyValueError::new_err(e.to_string()))?;
>
    Ok(asset)
}
>
#[pymodule]
fn rust_engine(m: &Bound<'_, PyModule>) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(init_engine, m)?)?;
    m.add_function(wrap_pyfunction!(shutdown_engine, m)?)?;
    m.add_function(wrap_pyfunction!(check_connection, m)?)?;
    m.add_function(wrap_pyfunction!(add_image_asset, m)?)?;
    m.add_function(wrap_pyfunction!(search_assets, m)?)?;
    m.add_function(wrap_pyfunction!(patch_asset_metadata, m)?)?;
>
    Ok(())
}

Python 코드

app/main.py

from fastapi import FastAPI, HTTPException
from fastapi.responses import ORJSONResponse
from contextlib import asynccontextmanager
import rust_engine
import logging
import json
from pydantic import BaseModel
>
# 기존 함수 생략
>
@app.patch("/assets/{asset_id}/metadata")
async def update_asset(asset_id: str, payload: dict):
    updated = rust_engine.patch_asset_metadata(asset_id, json.dumps(payload))
>
    return {
        "status": "success",
        "data": updated.to_dict()
    }

빌드 및 실행

Maturin 라이브러리를 통해 Rust 코드를 Python에서 호출 가능한 형태로 컴파일한다. pyproject.toml 파일에 build-backend = "maturin" 이 명시되어 있으니 uv를 사용하는 방식으로 컴파일하겠다.

uvicorn 라이브러리를 통해 FastAPI를 실행한다.

~/workspace/image-assets-management$ uv pip install -e .
~/workspace/image-assets-management$ uv run uvicorn app.main:app --reload

프론트엔드 단에서 /assets/search 로 조회하여 원하는 이미지 에셋의 id 를 파악하였고 그것에 대한 수정을 하는 것으로 가정하여 id 는 호출하는 쪽에서 알고 있는 상황을 상정한다.

실행 예시에서는 데이터 변화를 확인하기 위해 조회도 수행하겠다.

~$ curl -i "http://127.0.0.1:8000/assets/search?name_query=hero_char"
HTTP/1.1 200 OK
date: Wed, 22 Apr 2026 01:39:55 GMT
server: uvicorn
content-length: 252
content-type: application/json; charset=utf-8
>
{"total_count":1,"items":[{"id":"019d8eae-5943-76c4-9a85-195ce2c4f81a","file_name":"hero_char_diffuse.exr","file_path":"/assets/textures/hero/","width":4096,"height":4096,"metadata":{"avg_luminance":0.45,"format":"EXR","layers":32}}],"page":1,"size":5}%         ``` 

>
```bash
~$ curl -iX PATCH "http://127.0.0.1:8000/assets/019d8eae-5943-76c4-9a85-195ce2c4f81a/metadata" \
     -H "Content-Type: application/json" \
     -d '{"author": "peter", "is_verified": true}'
HTTP/1.1 200 OK
date: Wed, 22 Apr 2026 01:40:13 GMT
server: uvicorn
content-length: 270
content-type: application/json; charset=utf-8
>
{"status":"success","data":{"id":"019d8eae-5943-76c4-9a85-195ce2c4f81a","file_name":"hero_char_diffuse.exr","file_path":"/assets/textures/hero/","width":4096,"height":4096,"metadata":{"author":"peter","avg_luminance":0.45,"format":"EXR","is_verified":true,"layers":32}}}%

~$ curl -i "http://127.0.0.1:8000/assets/search?name_query=hero_char"                           
HTTP/1.1 200 OK
date: Wed, 22 Apr 2026 01:40:20 GMT
server: uvicorn
content-length: 288
content-type: application/json; charset=utf-8
>
{"total_count":1,"items":[{"id":"019d8eae-5943-76c4-9a85-195ce2c4f81a","file_name":"hero_char_diffuse.exr","file_path":"/assets/textures/hero/","width":4096,"height":4096,"metadata":{"author":"peter","avg_luminance":0.45,"format":"EXR","is_verified":true,"layers":32}}],"page":1,"size":5}% 

메타데이터가 수정된 것을 확인할 수 있다.

지난 시간에 만든 데이터베이스를 활용하는 실습을 진행해 보자. 원하는 조건에 따라 데이터를 뽑아내는 코드를 작성하겠다.

DB

인덱스

검색 효율을 높이기 위해 인덱스를 추가한다.

scripts/init.sql

CREATE TABLE IF NOT EXISTS image_assets (
    id UUID PRIMARY KEY DEFAULT uuidv7(),
    file_name TEXT NOT NULL,
    file_path TEXT NOT NULL,
    width INTEGER NOT NULL,
    height INTEGER NOT NULL,
    metadata JSONB,
    created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
);
>
-- 이상 기존 코드
>
CREATE INDEX IF NOT EXISTS idx_image_assets_metadata ON image_assets USING GIN (metadata);
CREATE INDEX IF NOT EXISTS idx_image_assets_file_name ON image_assets (file_name);

IF NOT EXISTS 를 사용하면 이 파일이 여러 번 전달되어도 각 작업을 중복해서 수행하지 않는다.

docker-compose.yml 파일의 volumns 에 scripts/init.sql:/docker-entrypoint-initdb.d/init.sql 를 추가하여 docker 서비스 실행 시 스크립트가 자동으로 실행되게 할 수도 있지만 여기서는 수동으로 전달하겠다.

~/workspace/image-assets-management$ docker exec -i image_assets_management_postgres psql -U peter -d image_db < scripts/init.sql

테이블은 이미 만들어져 있으니 skipping 으로 뜨고 인덱스가 생성될 것이다.

테스트용 데이터 셋

테스트용 데이터 셋을 넣을 스크립트를 작성한다.

seed_data.py

import httpx
import json
>
BASE_URL = "http://localhost:8000/assets"
>
data_set = [
    {"file_name": "char_hero_idle.exr", "file_path": "/assets/textures/hero/", "width": 4096, "height": 4096, "metadata": {"format": "EXR", "layers": 64, "type": "render"}},
    {"file_name": "char_hero_run.exr", "file_path": "/assets/textures/hero/", "width": 4096, "height": 4096, "metadata": {"format": "EXR", "layers": 64, "type": "render"}},
    {"file_name": "env_forest_ground.exr", "file_path": "/assets/textures/env/", "width": 8192, "height": 4096, "metadata": {"format": "EXR", "layers": 12, "type": "environment"}},
    {"file_name": "ui_button_normal.png", "file_path": "/assets/ui/buttons/", "width": 256, "height": 256, "metadata": {"format": "PNG", "alpha": True, "theme": "dark"}},
    {"file_name": "ui_icon_star.png", "file_path": "/assets/ui/icons/", "width": 128, "height": 128, "metadata": {"format": "PNG", "alpha": True, "theme": "gold"}},
    {"file_name": "ref_lighting_01.jpg", "file_path": "/assets/refs/lighting/", "width": 1920, "height": 1080, "metadata": {"format": "JPEG", "quality": 90, "source": "outdoor"}},
    {"file_name": "ref_material_metal.jpg", "file_path": "/assets/refs/materials/", "width": 2048, "height": 2048, "metadata": {"format": "JPEG", "quality": 95, "source": "substance"}},
    {"file_name": "fx_explosion_01.exr", "file_path": "/assets/fx/explosions/", "width": 2048, "height": 2048, "metadata": {"format": "EXR", "layers": 32, "type": "vfx"}},
    {"file_name": "skybox_sunset.exr", "file_path": "/assets/env/skybox/", "width": 16384, "height": 8192, "metadata": {"format": "EXR", "layers": 1, "type": "hdr"}},
    {"file_name": "char_npc_bakery.png", "file_path": "/assets/textures/npc/", "width": 1024, "height": 1024, "metadata": {"format": "PNG", "alpha": False, "type": "concept"}}
]
>
def seed_data():
    with httpx.Client() as client:
        print(f"{len(data_set)}개의 데이터 삽입 시작...")
 >       
        for item in data_set:
            try:
                response = client.post(BASE_URL, json=item)
                response.raise_for_status()
                result = response.json()
                print(f"성공 {item['file_name']} (ID: {result['id']}")
            except Exception as e:
                print(f"실패 {item['file_name']} - {str(e)}")
>
    print("모든 작업 완료")
>
if __name__ == "__main__":
    seed_data()

~/workspace/image-assets-management$ uv add --dev httpx
~/workspace/image-assets-management$ uv run python seed_data.py

코드 작성

Rust 코드

작업에 필요한 크레이트를 추가하기 위해 Cargo.toml 파일을 수정한다.

Cargo.toml

[package]
name = "image-assets-management"
version = "0.1.0"
edition = "2024"
>
[lib]
name = "rust_engine"
crate-type = ["cdylib"]
>
[dependencies]
# Python 연결
pyo3 = "0.28.0"
pythonize = "0.28"
# 비동기 엔진
tokio = { version = "1.43", features = ["full"] }
# 데이터베이스 연결
dotenvy = "0.15"
sqlx = { version = "0.8", features = ["runtime-tokio", "tls-rustls", "postgres", "macros", "uuid", "chrono", "json"] }
# 데이터베이스 호환
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
uuid = { version = "1.23", features = ["v7", "serde"] }
chrono = { version = "0.4", features = ["serde"] }
# 로그 수집
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }
tracing-appender = "0.2"

대소문자 구분 없이 파일 이름의 일부를 통해 검색하거나 metadata 에 format 이 있다면 이를 통해 검색하는 기본적인 검색 쿼리를 작성한다.

페이지네이션을 통해 응답 개수가 많아져도 부담되지 않게 작성하겠다.

이미지 에셋의 정보를 담을 객체는 #[pyclass] 속성을 붙여야 한다. 구조체 필드는 Python에서 읽을 수 있게 #[pyo3[get)] 속성을 붙이되, 바로 사용할 수 없는 필드에 대해서는 적절히 변환하여 반환하는 함수를 따로 구현해준다. dict 는 Pydantic이 모델을 직렬화할 때 사용하는 __dict__ 속성을 생성한다.

구조체를 JSON으로 변환하고 반대로도 변환하기 위해 serde::Serialize, serde::Deserialize 가 필요하며 sqlx를 사용하기 위해 sqlx::FromRow 도 필요하다.

응답 데이터를 담는 객체를 위한 구조체도 생성한다. #[pyclass] 속성에 get_all 을 붙이면 필드에 일일이 작성해 주지 않아도 Python에서 각 필드에 직접 접근할 수 있다.

src/image.rs

use pyo3::prelude::*;
use pyo3::exceptions::PyRuntimeError;
use sqlx::{Pool, Postgres};
use uuid::Uuid;
use serde_json::Value as JsonValue;
use tracing::{info, instrument, error};
>
#[pyclass(dict)]
#[derive(Clone, serde::Serialize, serde::Deserialize, sqlx::FromRow)]
pub struct ImageAsset {
    pub id: Uuid,
    #[pyo3(get)]
    pub file_name: String,
    #[pyo3(get)]
    pub file_path: String,
    #[pyo3(get)]
    pub width: i32,
    #[pyo3(get)]
    pub height: i32,
    pub metadata: JsonValue,
}
>
#[pymethods]
impl ImageAsset {
    #[getter]
    fn id(&self) -> String {
        self.id.to_string()
    }
>
    #[getter]
    fn metadata<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
        pythonize::pythonize(py, &self.metadata)
            .map_err(|e| PyRuntimeError::new_err(e.to_string()))
    }
}
>
#[pyclass(get_all, dict)]
#[derive(Clone, serde::Serialize, serde::Deserialize, sqlx::FromRow)]
pub struct AssetListResponse {
    pub total_count: i64,
    pub items: Vec<ImageAsset>,
    pub page: i32,
    pub size: i32,
}
>
#[pymethods]
impl AssetListResponse {
    fn to_dict<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
        pythonize::pythonize(py, self)
            .map_err(|e| PyRuntimeError::new_err(e.to_string()))
    }
}
>
#[instrument(skip(pool, asset), fields(file = %asset.file_name))]
pub async fn insert_asset(pool: &Pool<Postgres>, asset: ImageAsset) -> Result<Uuid, sqlx::Error> {
    info!("이미지 자산 저장 중...");
>
    let row = sqlx::query!(
        r#"
        INSERT INTO image_assets (file_name, file_path, width, height, metadata)
        VALUES ($1, $2, $3, $4, $5)
        RETURNING id
        "#,
        asset.file_name,
        asset.file_path,
        asset.width,
        asset.height,
        asset.metadata
    )
    .fetch_one(pool)
    .await
    .map_err(|e| {
        error!("Insert 쿼리 실패: {}", e);
        e
    })?;
>
    info!(asset_id = %row.id, "자산 저장 완료");
>
    Ok(row.id)
}
>
#[instrument(skip(pool))]
pub async fn search_assets(
    pool: &Pool<Postgres>,
    name_query: Option<String>,
    format_query: Option<String>,
    limit: i64,
    offset: i64
) -> Result<AssetListResponse, sqlx::Error> {
    let rows = sqlx::query!(
        r#"
        SELECT id, file_name, file_path, width, height, metadata, count(*) OVER() as "total_count!"
        FROM image_assets
        WHERE ($1::TEXT IS NULL OR file_name ILIKE $1)
            AND ($2::JSONB IS NULL OR metadata @> $2)
        ORDER BY id DESC
        LIMIT $3 OFFSET $4
        "#,
        name_query.map(|n| format!("%{}%", n)),
        format_query.map(|f| serde_json::json!({"format": f})),
        limit,
        offset
    )
    .fetch_all(pool)
    .await?;
>
    let total_count = rows.first().map(|r| r.total_count).unwrap_or(0);
    let items = rows.into_iter().map(|r| ImageAsset {
        id: r.id,
        file_name: r.file_name,
        file_path: r.file_path,
        width: r.width,
        height: r.height,
        metadata: r.metadata.unwrap_or(serde_json::Value::Null),
    }).collect();
>
    let size = limit as i32;
    let page = (offset as i32 / size) + 1;
>
    Ok(AssetListResponse{ total_count, items, page, size })
}

#[pyo3(signature = ())] 를 사용하여 기본값을 지정하면 Python에서 인자를 선택적으로 넘길 수 있다.

src/lib.rs

mod db;
mod logger;
mod image;
>
use pyo3::prelude::*;
use tokio::runtime::Runtime;
use dotenvy::dotenv;
use std::env;
use std::sync::OnceLock;
use tracing::{debug, error, info, instrument};
use tracing_appender::non_blocking::WorkerGuard;
>
static LOG_GUARD: OnceLock<WorkerGuard> = OnceLock::new();
static TOKIO_RUNTIME: OnceLock<Runtime> = OnceLock::new();
>
fn get_runtime() -> &'static Runtime {
    TOKIO_RUNTIME.get_or_init(|| {
        Runtime::new().expect("Tokio 런타임 생성 실패")
    })
}
>
#[pyfunction]
fn init_engine() -> PyResult<String> {
    info!("엔진 초기화 절차 시작");
>
    debug!("로깅 시작");
    if LOG_GUARD.get().is_none() {
        let guard = logger::init_tracing();
        let _ = LOG_GUARD.set(guard);
    }
>
    debug!("환경 변수 로드");
    dotenv().ok();
    let url = env::var("DATABASE_URL")
        .map_err(|e| {
            error!("DATABASE_URL 환경 변수를 찾을 수 없음: {}", e);
            pyo3::exceptions::PyRuntimeError::new_err("DATABASE_URL not found in .env")
        })?;
 >   
    debug!("데이터베이스 연결 시작");
    let rt = get_runtime();
    rt.block_on(async {
        db::connect(&url).await
    })
    .map_err(|e| {
        error!("DB 연결 풀 생성 실패: {}", e);
        pyo3::exceptions::PyRuntimeError::new_err(format!("DB connection failed: {}", e))
    })?;
>
    Ok("엔진 초기화 및 DB 연결 성공".to_string())
}
>
#[pyfunction]
fn shutdown_engine() -> PyResult<()> {
    info!("엔진 종료 절차 시작");
 >   
    let rt = get_runtime();
    rt.block_on(async {
        db::close().await;
    });
>
    Ok(())
}
>
#[pyfunction]
fn check_connection() -> PyResult<bool> {
    debug!("DB 연결 상태 확인");
>
    Ok(db::is_alive())
}
>
#[pyfunction]
#[instrument(skip(metadata))]
fn add_image_asset(
    file_name: String,
    file_path: String,
    width: i32,
    height: i32,
    metadata: String
) -> PyResult<String> {
    let rt = get_runtime();
>
    let meta_json: serde_json::Value = serde_json::from_str(&metadata)
        .map_err(|e| pyo3::exceptions::PyValueError::new_err(e.to_string()))?;
>
    let asset = image::ImageAsset {
        id: uuid::Uuid::nil(),
        file_name,
        file_path,
        width,
        height,
        metadata: meta_json,
    };
>
    let pool = db::DB_POOL.get()
        .ok_or(pyo3::exceptions::PyRuntimeError::new_err("DB 연결 풀이 없습니다."))?;
>
    let id = rt.block_on(async {
        image::insert_asset(pool, asset).await
    }).map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))?;
>
    Ok(id.to_string())
}
>
#[pyfunction]
#[pyo3(signature = (name=None, format=None, limit=10, offset=0))]
fn search_assets(
    name: Option<String>,
    format: Option<String>,
    limit: i64,
    offset: i64
) -> PyResult<image::AssetListResponse> {
    let rt = get_runtime();
    let pool = db::DB_POOL.get()
        .ok_or(pyo3::exceptions::PyRuntimeError::new_err("DB 연결 풀이 없습니다."))?;
>
    let results = rt.block_on(async {
        image::search_assets(pool, name, format, limit, offset).await
    }).map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))?;
>
    Ok(results)
}
>
#[pymodule]
fn rust_engine(m: &Bound<'_, PyModule>) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(init_engine, m)?)?;
    m.add_function(wrap_pyfunction!(shutdown_engine, m)?)?;
    m.add_function(wrap_pyfunction!(check_connection, m)?)?;
    m.add_function(wrap_pyfunction!(add_image_asset, m)?)?;
    m.add_function(wrap_pyfunction!(search_assets, m)?)?;
>
    Ok(())
}

Python 코드

기본값을 설정해 놓고 쿼리로 넘어온 값에 대해서만 값을 지정한다.

app/main.py

from fastapi import FastAPI, HTTPException
from fastapi.responses import ORJSONResponse
from contextlib import asynccontextmanager
import rust_engine
import logging
import json
from pydantic import BaseModel
>
logger = logging.getLogger("uvicorn.error")
>
class UTF8ORJSONResponse(ORJSONResponse):
    media_type = "application/json; charset=utf-8"
>
@asynccontextmanager
async def lifespan(app: FastAPI):
    logger.info("Rust 엔진 초기화 중...")
    try:
        msg = rust_engine.init_engine()
        logger.info(msg)
    except Exception as e:
        logger.info(f"Rust 엔진 초기화 실패: {e}")
>
    yield # 앱 가동
>
    # [SHUTDOWN]
>
    logger.info("서버 종료 감지: 자원 정리 중...")
    try:
        rust_engine.shutdown_engine()
        logger.info("모든 연결 안전하게 종료")
    except Exception as e:
        logger.error(f"종료 중 오류 발생: {e}")
>
app = FastAPI(default_response_class=UTF8ORJSONResponse, lifespan=lifespan)
>
@app.get("/")
def read_root():
    return {
        "status": "200",
        "info": "서버 가동 중입니다."
    }
>
@app.get("/db-status")
def get_db_status():
    logger.info("DB 상태 체크 요청")
    is_alive = rust_engine.check_connection()
    if is_alive:
        logger.info("DB 연결 상태 양호")
        return {
            "status": "online",
            "message": "Rust 엔진이 PostgreSQL을 사용합니다."
        }
    else:
        logger.error("DB 연결 끊김 감지")
        raise HTTPException(
            status_code=500,
            detail="DB 연결이 끊겼거나 초기화되지 않았습니다."
        )
>
class ImageRequest(BaseModel):
    file_name: str
    file_path: str
    width: int
    height: int
    metadata: dict
>
@app.post("/assets")
async def create_asset(req: ImageRequest):
    try:
        asset_id = rust_engine.add_image_asset(
            req.file_name,
            req.file_path,
            req.width,
            req.height,
            json.dumps(req.metadata)
        )
>
        return {
            "status": "success",
            "id": asset_id
        }
    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail=str(e)
        )
>
@app.get("/assets/search")
async def get_assets(
    name_query: str = None,
    format_query: str = None,
    page: int = 1,
    size: int = 5
):
    offset = (page - 1) * size
    response = rust_engine.search_assets(
        name=name_query,
        format=format_query,
        limit=size,
        offset=offset
    )
>
    return response.to_dict()

빌드 및 실행

Maturin 라이브러리를 통해 Rust 코드를 Python에서 호출 가능한 형태로 컴파일한다. pyproject.toml 파일에 build-backend = "maturin" 이 명시되어 있으니 uv를 사용하는 방식으로 컴파일하겠다.

uvicorn 라이브러리를 통해 FastAPI를 실행한다.

~/workspace/image-assets-management$ uv pip install -e .
~/workspace/image-assets-management$ uv run uvicorn app.main:app --reload

curl 명령어 또는 브라우저를 통해 다음과 같은 테스트를 해볼 수 있다.

• http://127.0.0.1:8000/assets/search
• http://127.0.0.1:8000/assets/search?fomrat_query=EXR
• http://127.0.0.1:8000/assets/search?name_query=hero

아무 인자를 전달하지 않으면 전체 데이터가 출력되며 format_query 및 name_query 를 통해 쿼리를 전달할 수 있다. curl 명령어로 쿼리 문자열을 전달할 때는 URL을 따옴표로 묶어야 한다.

~$ curl -i http://127.0.0.1:8000/assets/search                   
HTTP/1.1 200 OK
date: Tue, 21 Apr 2026 01:06:41 GMT
server: uvicorn
content-length: 1053
content-type: application/json; charset=utf-8
>
{"total_count":11,"items":[{"id":"019d9441-cbba-74b9-9667-1b06b469045e","file_name":"char_npc_bakery.png","file_path":"/assets/textures/npc/","width":1024,"height":1024,"metadata":{"alpha":false,"format":"PNG","type":"concept"}},{"id":"019d9441-cbb4-7f20-b5e8-285e9d3ced04","file_name":"skybox_sunset.exr","file_path":"/assets/env/skybox/","width":16384,"height":8192,"metadata":{"format":"EXR","layers":1,"type":"hdr"}},{"id":"019d9441-cbaf-7961-86d0-fe5c565d96d0","file_name":"fx_explosion_01.exr","file_path":"/assets/fx/explosions/","width":2048,"height":2048,"metadata":{"format":"EXR","layers":32,"type":"vfx"}},{"id":"019d9441-cbaa-7eb2-9c45-59a183cdb110","file_name":"ref_material_metal.jpg","file_path":"/assets/refs/materials/","width":2048,"height":2048,"metadata":{"format":"JPEG","quality":95,"source":"substance"}},{"id":"019d9441-cba5-7fe1-a000-000623614320","file_name":"ref_lighting_01.jpg","file_path":"/assets/refs/lighting/","width":1920,"height":1080,"metadata":{"format":"JPEG","quality":90,"source":"outdoor"}}],"page":1,"size":5}%    

~$ curl -i "http://127.0.0.1:8000/assets/search?page=2"          
HTTP/1.1 200 OK
date: Tue, 21 Apr 2026 01:08:27 GMT
server: uvicorn
content-length: 1035
content-type: application/json; charset=utf-8
>
{"total_count":11,"items":[{"id":"019d9441-cba0-7d0e-9008-e2201e853806","file_name":"ui_icon_star.png","file_path":"/assets/ui/icons/","width":128,"height":128,"metadata":{"alpha":true,"format":"PNG","theme":"gold"}},{"id":"019d9441-cb9c-7208-bcda-5061bbaf2cdc","file_name":"ui_button_normal.png","file_path":"/assets/ui/buttons/","width":256,"height":256,"metadata":{"alpha":true,"format":"PNG","theme":"dark"}},{"id":"019d9441-cb97-7099-ab84-e6cc1c087381","file_name":"env_forest_ground.exr","file_path":"/assets/textures/env/","width":8192,"height":4096,"metadata":{"format":"EXR","layers":12,"type":"environment"}},{"id":"019d9441-cb91-79bb-9bb5-4cd6326f0e68","file_name":"char_hero_run.exr","file_path":"/assets/textures/hero/","width":4096,"height":4096,"metadata":{"format":"EXR","layers":64,"type":"render"}},{"id":"019d9441-cb87-7f42-b228-2e3590d90a05","file_name":"char_hero_idle.exr","file_path":"/assets/textures/hero/","width":4096,"height":4096,"metadata":{"format":"EXR","layers":64,"type":"render"}}],"page":2,"size":5}% 

~$ curl -i "http://127.0.0.1:8000/assets/search?page=3"
HTTP/1.1 200 OK
date: Tue, 21 Apr 2026 01:08:49 GMT
server: uvicorn
content-length: 253
content-type: application/json; charset=utf-8
>
{"total_count":11,"items":[{"id":"019d8eae-5943-76c4-9a85-195ce2c4f81a","file_name":"hero_char_diffuse.exr","file_path":"/assets/textures/hero/","width":4096,"height":4096,"metadata":{"avg_luminance":0.45,"format":"EXR","layers":32}}],"page":3,"size":5}% 

~$ curl -i "http://127.0.0.1:8000/assets/search?format_query=EXR"
HTTP/1.1 200 OK
date: Tue, 21 Apr 2026 01:06:44 GMT
server: uvicorn
content-length: 1037
content-type: application/json; charset=utf-8
>
{"total_count":6,"items":[{"id":"019d9441-cbb4-7f20-b5e8-285e9d3ced04","file_name":"skybox_sunset.exr","file_path":"/assets/env/skybox/","width":16384,"height":8192,"metadata":{"format":"EXR","layers":1,"type":"hdr"}},{"id":"019d9441-cbaf-7961-86d0-fe5c565d96d0","file_name":"fx_explosion_01.exr","file_path":"/assets/fx/explosions/","width":2048,"height":2048,"metadata":{"format":"EXR","layers":32,"type":"vfx"}},{"id":"019d9441-cb97-7099-ab84-e6cc1c087381","file_name":"env_forest_ground.exr","file_path":"/assets/textures/env/","width":8192,"height":4096,"metadata":{"format":"EXR","layers":12,"type":"environment"}},{"id":"019d9441-cb91-79bb-9bb5-4cd6326f0e68","file_name":"char_hero_run.exr","file_path":"/assets/textures/hero/","width":4096,"height":4096,"metadata":{"format":"EXR","layers":64,"type":"render"}},{"id":"019d9441-cb87-7f42-b228-2e3590d90a05","file_name":"char_hero_idle.exr","file_path":"/assets/textures/hero/","width":4096,"height":4096,"metadata":{"format":"EXR","layers":64,"type":"render"}}],"page":1,"size":5}% 

~$ curl -i "http://127.0.0.1:8000/assets/search?name_query=hero" 
HTTP/1.1 200 OK
date: Tue, 21 Apr 2026 01:06:47 GMT
server: uvicorn
content-length: 649
content-type: application/json; charset=utf-8
>
{"total_count":3,"items":[{"id":"019d9441-cb91-79bb-9bb5-4cd6326f0e68","file_name":"char_hero_run.exr","file_path":"/assets/textures/hero/","width":4096,"height":4096,"metadata":{"format":"EXR","layers":64,"type":"render"}},{"id":"019d9441-cb87-7f42-b228-2e3590d90a05","file_name":"char_hero_idle.exr","file_path":"/assets/textures/hero/","width":4096,"height":4096,"metadata":{"format":"EXR","layers":64,"type":"render"}},{"id":"019d8eae-5943-76c4-9a85-195ce2c4f81a","file_name":"hero_char_diffuse.exr","file_path":"/assets/textures/hero/","width":4096,"height":4096,"metadata":{"avg_luminance":0.45,"format":"EXR","layers":32}}],"page":1,"size":5}%    

현대 백엔드 아키텍처의 핵심은 모든 것을 표에 넣는 것이 아니라, 데이터의 성격에 따라 저장 방식을 결정하는 것이다. 정형 데이터와 비정형 데이터라는 두 방식을 한 테이블 안에서 어떻게 효율적으로 관리할 수 있는지 알아보도록 하자.

비정형 데이터를 포함하기 위해 이미지 파일의 메타데이터를 관리하는 데이터베이스를 상정하고 실습을 진행하겠다. 메타데이터의 구성은 이미지 파일마다 다르다. JSONB를 사용하면 스키마 재정의 없이 다양한 구성의 메타데이터를 관리할 수 있다.

이미지 자체를 DB에 넣을 경우(BLOB) 백업과 복구 성능이 $O(n)$ 으로 선형 증가하여 효율성이 떨어지므로 실제 데이터는 파일 시스템이나 스토리지에 두고, 그 메타데이터와 상태값만 DB에서 초고속으로 동기화하도록 한다.

작업공간 생성 및 구조 확인

기존에 만든 데이터베이스 템플릿을 복제해서 사용해 보겠다. .env 파일과 docker-compose.yml 파일은 ⟨데이터베이스 연결⟩이 최신 버전이고 Rust 코드 및 Python 코드는 ⟨데이터베이스 연결 재구조화⟩가 최신 버전이고 Cargo.toml 파일과 pyproject.toml 파일은 ⟨(여담) uv 패키지 관리자⟩가 최신 버전이다.

~/workspace$ cp -r db-template image-assets-management
~/workspace$ cd image-assets-management
~/workspace/image-assets-management$ tree -a -I .venv
.
├── .env                # 우리 프로젝트의 DB로 연결되게 수정 필요
├── .github
│   └── workflows
│       └── CI.yml
├── .gitignore
├── app
│   └── main.py            # 우리 프로젝트의 라우트 함수 추가 필요
├── Cargo.toml            # 우리 프로젝트의 정보로 수정 필요
├── docker-compose.yml    # 우리 프로젝트의 DB로 연결되게 수정 필요
├── pyproject.toml        # 우리 프로젝트의 정보로 수정 필요
└── src
    ├── db.rs            # 우리 프로젝트에 필요한 함수 추가 필요
    ├── lib.rs            # 우리 프로젝트에 필요한 함수 추가 필요
    └── logger.rs        # 그대로 유지해도 무방

pyproject.toml 파일을 열어 프로젝트 이름을 수정해 주겠다. 나머지 의존성 부분은 이번 프로젝트에서도 사용하는 것들이니 가만히 둔다.

~/workspace/image-assets-management$ vi pyproject.toml

pyproject.toml

[build-system]
requires = ["maturin>=1.12,<2.0"]
build-backend = "maturin"
>
[project]
name = "image-assets-management"
requires-python = ">=3.8"
classifiers = [
    "Programming Language :: Rust",
    "Programming Language :: Python :: Implementation :: CPython",
    "Programming Language :: Python :: Implementation :: PyPy",
]
dynamic = ["version"]
dependencies = [
    "fastapi>=0.124.4",
    "orjson>=3.10.15",
    "uvicorn>=0.33.0",
]
>
[dependency-groups]
dev = [
    "maturin>=1.12.6",
]

다음과 같이 uv 를 통해 pyproject.toml 파일의 모든 의존성을 설치할 수 있다.

~/workspace/image-assets-management$ uv sync

마찬가지로 Cargo.toml 파일도 수정해 준다.

이 때, 필요한 크레이트도 추가해 주도록 하자. 데이터베이스의 기본값으로 사용할 UUID를 위해 uuid 크레이트를 추가하고 (UUID가 뭔지 궁금하다면 해당 크레이트에도 설명이 나와 있으니 클릭해 보자) 생성 일시를 기록하기 위해 chrono 크레이트를 추가하고 JSON 데이터를 다루기 위해 serde_json 크레이트도 추가한다.

~/workspace/image-assets-management$ vi Cargo.toml

Cargo.toml

[package]
name = "image-assets-management"
version = "0.1.0"
edition = "2024"
>
[lib]
name = "rust_engine"
crate-type = ["cdylib"]
>
[dependencies]
pyo3 = "0.28.0"
sqlx = { version = "0.8", features = ["runtime-tokio", "tls-rustls", "postgres", "macros", "uuid", "chrono", "json"] }
tokio = { version = "1.43", features = ["full"] }
tracing-appender = "0.2"
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }
dotenvy = "0.15"
uuid = { version = "1.23", features = ["v7", "serde"] }
chrono = { version = "0.4", features = ["serde"] }
serde_json = "1.0"

템플릿 프로젝트의 DB 정보를 그대로 사용하면 DB가 꼬이게 될 테니 DB 정보도 수정해 준다.

docker-compose.yml 의 변경 필요한 부분은 전부 환경변수로 되어 있으니 .env 파일만 편집하면 된다.

계정 정보는 그대로 사용해도 무방하지만 COMPOSE_PROJECT_NAME 과 POSTGRES_DB 는 수정해 주어야 한다.

~/workspace/image-assets-management$ vi .env

.env

# Docker Compose Project
COMPOSE_PROJECT_NAME=image_assets_management
>
# PostgreSQL
POSTGRES_USER=peter
POSTGRES_PASSWORD=ku201711424
POSTGRES_DB=image_db
POSTGRES_HOST=127.0.0.1
POSTGRES_PORT=5432
>
# pgAdmin
PGADMIN_EMAIL=admin@pjos.dev
PGADMIN_PASSWORD=admin201711424
>
# SQLX
DATABASE_URL=postgres://peter:ku201711424@127.0.0.1:5432/image_db?sslmode=disable

재차 말하지만 학습 기록용이니까 업로드하지 .env 파일의 내용은 유출하는 거 아니다. 보안 이슈!

DB 설정

스키마

이미지 정보를 담을 데이터베이스 테이블을 만들 것이다. 사용할 데이터베이스 테이블의 구성은 다음과 같다.

이 데이터베이스 테이블이 존재하지 않는 경우에만 새로 생성하는 코드를 작성해 보자. 테이블 이름은 image_assets 로 하겠다.

~/workspace/image-assets-management$ mkdir scripts
~/workspace/image-assets-management$ vi scripts/init.sql

scripts/init.sql

CREATE TABLE IF NOT EXISTS image_assets {
    id UUID PRIMARY KEY DEFAULT uuidv7(),
    file_name TEXT NOT NULL,
    file_path TEXT NOT NULL,
    width INTEGER NOT NULL,
    height INTEGER NOT NULL,
    metadata JSONB,
    created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
};

uuidv7() 은 UUID v7을 사용하겠다고 하는 것이고 UUID v4를 사용하고자 한다면 gen_random_uuid() 를 사용한다.

파일을 생성하지 않고 터미널에서 직접 쿼리를 날려도 되지만 따로 파일을 작성하는 편이 유지보수 측면에서 효과적이다.

docker

PostgreSQL을 docker 서비스로 실행한다.

~/workspace/image-assets-management$ docker compose up -d
~/workspace/image-assets-management$ docker ps                    
CONTAINER ID   IMAGE                COMMAND                   CREATED         STATUS         PORTS                                         NAMES
64808bc6afba   dpage/pgadmin4       "/entrypoint.sh"          5 seconds ago   Up 5 seconds   0.0.0.0:8080->80/tcp, [::]:8080->80/tcp       pgadmin_ui
0ca8e65aa3bd   postgres:18-alpine   "docker-entrypoint.s…"   5 seconds ago   Up 5 seconds   0.0.0.0:5432->5432/tcp, [::]:5432->5432/tcp   image_assets_management_postgres

image_assets_management_postgres 내부에서 작업을 수행할 것이다.

대화형 터미널을 열어 쿼리를 직접 날리고 싶을 땐 다음과 같이 -it 를 붙여 사용하지만

~/workspace/image-assets-management$ docker exec -it image_assets_management_postgres psql -U peter -d image_db

우리는 대화형으로 쿼리를 날리지 않고 기존에 작성해 놓은 쿼리 파일을 전달할 것이므로 -i 를 사용하고 파이프라인으로 전달한다.

~/workspace/image-assets-management$ docker exec -i image_assets_management_postgres psql -U peter -d image_db < scripts/init.sql

-U 뒤에 오는 건 .env 의 POSTGRES_USER 이며 -d 뒤에 오는 건 .env 의 POSTGRES_DB 이다. 실무에서 여럿이서 작업할 경우 POSTGRES_USER 는 postgres 로 설정하는 경우가 많다.

PostgreSQL 컨테이너는 데이터 볼륨이 이미 존재하면 .env 의 설정이 바뀌어도 새로운 사용자를 추가로 만들지 않으니 다른 설정으로 컨테이너를 띄운 적이 있을 때 사용자 이름을 변경하고자 한다면 볼륨을 지우고 다시 시작해야 바뀐 설정이 적용된다.

다음과 같이 데이터베이스 테이블이 생성된 것을 확인할 수 있다.

~/workspace/image-assets-management$ docker exec -it image_assets_management_postgres psql -U peter -d image_db -c "\dt"
            List of tables
 Schema |     Name     | Type  | Owner 
--------+--------------+-------+-------
 public | image_assets | table | peter
(1 row)

그리고 그것은 아직 비어 있다.

~/workspace/image-assets-management$ docker exec -it image_assets_management_postgres psql -U peter -d image_db -c "SELECT id, file_name, created_at FROM image_assets;"
 id | file_name | created_at 
----+-----------+------------
(0 rows)

여기에 데이터를 넣는 코드를 작성해 보자.

코드 작성

Rust 코드

기존의 로그 관리 모듈, 데이터베이스 관리 모듈, 인터페이스 모듈 외에 이미지 데이터를 다루는 Rust 파일을 새로 생성한다.

src/image.rs

use sqlx::{Pool, Postgres};
use uuid::Uuid;
use serde_json::Value as JsonValue;
use tracing::{info, instrument, error};
>
pub struct ImageAsset {
    pub file_name: String,
    pub file_path: String,
    pub width: i32,
    pub height: i32,
    pub metadata: JsonValue,
}
>
#[instrument(skip(pool, asset), fields(file = %asset.file_name))]
pub async fn insert_asset(pool: &Pool<Postgres>, asset: ImageAsset) -> Result<Uuid, sqlx::Error> {
    info!("이미지 자산 저장 중...");
>
    let row = sqlx::query!(
        r#"
        INSERT INTO image_assets (file_name, file_path, width, height, metadata)
        VALUES ($1, $2, $3, $4, $5)
        RETURNING id
        "#,
        asset.file_name,
        asset.file_path,
        asset.width,
        asset.height,
        asset.metadata
    )
    .fetch_one(pool)
    .await
    .map_err(|e| {
        error!("Insert 쿼리 실패: {}", e);
        e
    })?;
>
    info!(asset_id = %row.id, "자산 저장 완료");
>
    Ok(row.id)
}

새로 추가한 이미지 모듈에 대한 정보를 인터페이스 모듈에 등록한다.

src/lib.rs

mod db;
mod logger;
mod image; // NEW!
>
use pyo3::prelude::*;
use tokio::runtime::Runtime;
use dotenvy::dotenv;
use std::env;
use std::sync::OnceLock;
use tracing::{debug, error, info, instrument}; // MODIFIED!
use tracing_appender::non_blocking::WorkerGuard;
>
static LOG_GUARD: OnceLock<WorkerGuard> = OnceLock::new();
static TOKIO_RUNTIME: OnceLock<Runtime> = OnceLock::new(); // NEW!
>
> // 추가
fn get_runtime() -> &'static Runtime {
    TOKIO_RUNTIME.get_or_init(|| {
        Runtime::new().expect("Tokio 런타임 생성 실패")
    })
}
>
#[pyfunction]
fn init_engine() -> PyResult<String> {
    info!("엔진 초기화 절차 시작");
>
    debug!("로깅 시작");
    if LOG_GUARD.get().is_none() {
        let guard = logger::init_tracing();
        let _ = LOG_GUARD.set(guard);
    }
>
    debug!("환경 변수 로드");
    dotenv().ok();
    let url = env::var("DATABASE_URL")
        .map_err(|e| {
            error!("DATABASE_URL 환경 변수를 찾을 수 없음: {}", e);
            pyo3::exceptions::PyRuntimeError::new_err("DATABASE_URL not found in .env")
        })?;
 >   
    debug!("데이터베이스 연결 시작");
    let rt = get_runtime(); // MODIFIED!
    rt.block_on(async {
        db::connect(&url).await
    })
    .map_err(|e| {
        error!("DB 연결 풀 생성 실패: {}", e);
        pyo3::exceptions::PyRuntimeError::new_err(format!("DB connection failed: {}", e))
    })?;
>
    Ok("엔진 초기화 및 DB 연결 성공".to_string())
}
>
#[pyfunction]
fn shutdown_engine() -> PyResult<()> {
    info!("엔진 종료 절차 시작");
 >   
    let rt = Runtime::new().unwrap();
    rt.block_on(async {
        db::close().await;
    });
>
    Ok(())
}
>
#[pyfunction]
fn check_connection() -> PyResult<bool> {
    debug!("DB 연결 상태 확인");
>
    Ok(db::is_alive())
}
>
// 이상 기존 코드
>
#[pyfunction]
#[instrument(skip(metadata))]
fn add_image_asset(
    file_name: String,
    file_path: String,
    width: i32,
    height: i32,
    metadata: String
) -> PyResult<String> {
    let rt = get_runtime(); // MODIFIED!
>
    let meta_json: serde_json::Value = serde_json::from_str(&metadata)
        .map_err(|e| pyo3::exceptions::PyValueError::new_err(e.to_string()))?;
>
    let asset = image::ImageAsset {
        file_name, file_path, width, height, metadata: meta_json
    };
>
    let pool = db::DB_POOL.get()
        .ok_or(pyo3::exceptions::PyRuntimeError::new_err("DB 연결 풀이 없습니다."))?;
>
    let id = rt.block_on(async {
        image::insert_asset(pool, asset).await
    }).map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))?;
>
    Ok(id.to_string())
}
>
#[pymodule]
fn rust_engine(m: &Bound<'_, PyModule>) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(init_engine, m)?)?;
    m.add_function(wrap_pyfunction!(shutdown_engine, m)?)?;
    m.add_function(wrap_pyfunction!(check_connection, m)?)?;
    m.add_function(wrap_pyfunction!(add_image_asset, m)?)?; // NEW!
>
    Ok(())
}

Python 코드

app/main.py

from fastapi import FastAPI, HTTPException
from fastapi.responses import ORJSONResponse
from contextlib import asynccontextmanager
import rust_engine
import logging
import json # NEW!
from pydantic import BaseModel # NEW!
>
logger = logging.getLogger("uvicorn.error")
>
class UTF8ORJSONResponse(ORJSONResponse):
    media_type = "application/json; charset=utf-8"
>
@asynccontextmanager
async def lifespan(app: FastAPI):
    logger.info("Rust 엔진 초기화 중...")
    try:
        msg = rust_engine.init_engine()
        logger.info(msg)
    except Exception as e:
        logger.info(f"Rust 엔진 초기화 실패: {e}")
>
    yield # 앱 가동
>
    # [SHUTDOWN]
>
    logger.info("서버 종료 감지: 자원 정리 중...")
    try:
        rust_engine.shutdown_engine()
        logger.info("모든 연결 안전하게 종료")
    except Exception as e:
        logger.error(f"종료 중 오류 발생: {e}")
>
app = FastAPI(default_response_class=UTF8ORJSONResponse, lifespan=lifespan)
>
@app.get("/")
def read_root():
    return {
        "status": "200",
        "info": "서버 가동 중입니다."
    }
>
@app.get("/db-status")
def get_db_status():
    logger.info("DB 상태 체크 요청")
    is_alive = rust_engine.check_connection()
    if is_alive:
        logger.info("DB 연결 상태 양호")
        return {
            "status": "online",
            "message": "Rust 엔진이 PostgreSQL을 사용합니다."
        }
    else:
        logger.error("DB 연결 끊김 감지")
        raise HTTPException(
            status_code=500,
            detail="DB 연결이 끊겼거나 초기화되지 않았습니다."
        )
>
class ImageRequest(BaseModel):
    file_name: str
    file_path: str
    width: int
    height: int
    metadata: dict
>
# 이상 기존 코드
>
@app.post("/assets")
async def create_asset(req: ImageRequest):
    try:
        asset_id = rust_engine.add_image_asset(
            req.file_name,
            req.file_path,
            req.width,
            req.height,
            json.dumps(req.metadata)
        )
        return {
            "status": "success",
            "id": asset_id
        }
    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail=str(e)
        )

빌드 및 실행

테스트용 더미 파일을 생성한다.

create_assets.py

import os
>
def create_dummy_assets():
    # 에셋 저장 폴더 생성
    base_dir = "assets/textures/hero"
    os.makedirs(base_dir, exist_ok=True)
>  
    # 생성할 파일 목록 (파일명, 크기)
    files = {
        "hero_char_diffuse.exr": 1024 * 10, # 10KB dummy
        "ui_button_hover.png": 1024 * 2,
        "env_forest_01_norm.tga": 1024 * 5,
        "emotion_icon_joy.svg": 1024 * 1
    }
>   
    print(f"'{base_dir}' 폴더에 테스트 에셋을 생성합니다...")
    for name, size in files.items():
        path = os.path.join(base_dir, name)
        with open(path, "wb") as f:
            f.write(os.urandom(size)) # 실제 바이너리 데이터처럼 보이기 위해 난수 입력
        print(f"생성 완료: {name} ({size} bytes)")
>
if __name__ == "__main__":
    create_dummy_assets()

~/workspace/image-assets-management$ uv run create_assets.py 
'assets/textures/hero' 폴더에 테스트 에셋을 생성합니다...
생성 완료: hero_char_diffuse.exr (10240 bytes)
생성 완료: ui_button_hover.png (2048 bytes)
생성 완료: env_forest_01_norm.tga (5120 bytes)
생성 완료: emotion_icon_joy.svg (1024 bytes)

Maturin 라이브러리를 통해 Rust 코드를 Python에서 호출 가능한 형태로 컴파일한다.

uvicorn 라이브러리를 통해 FastAPI를 실행한다.

~/workspace/image-assets-management$ uv run maturin develop
~/workspace/image-assets-management$ uv run uvicorn app.main:app --reload

curl 명령어 또는 브라우저를 통해 다음과 같은 테스트를 해볼 수 있다.

• http://127.0.0.1:8000/assets

~$ curl -X POST "http://localhost:8000/assets" \
     -H "Content-Type: application/json" \
     -d '{
       "file_name": "hero_char_diffuse.exr",
       "file_path": "/assets/textures/hero/",
       "width": 4096,
       "height": 4096,
       "metadata": {"format": "EXR", "layers": 32, "avg_luminance": 0.45}
     }'
{"status":"success","id":"019d8eae-5943-76c4-9a85-195ce2c4f81a"}%

~/workspace/image-assets-management$ docker exec -it image_assets_management_postgres psql -U peter -d image_db -c "SELECT id, file_name, created_at FROM image_assets;"
                  id                  |       file_name       |          created_at          
--------------------------------------+-----------------------+------------------------------
 019d8eae-5943-76c4-9a85-195ce2c4f81a | hero_char_diffuse.exr | 2026-04-15 01:08:04.03512+00
(1 row)

여담

• 실무적 팁: 최근 가장 현대적인 패키지 관리자인 uv는 속도가 빠른 대신 '캐싱(Caching)'이 매우 공격적입니다. uv run maturin develop 을 실행할 때 가상환경은 활성화되지만, 파이썬 쪽 스크립트 시그니처나 의존성이 크게 바뀌지 않았다고 판단하면 uv의 캐시나 이전 빌드 잔재 때문에 .so 나 .pyd 확장이 제대로 덮어씌워지지 않는 경우가 잦습니다.
• 해결책: 이럴 때는 억지로 원인을 찾으며 스트레스(자형)를 받기보다, 아주 물리적인 방식으로 접근하는 것이 좋습니다. uv cache clean 으로 캐시를 강제로 날려버리거나, Rust의 target/ 폴더를 통째로 날리고 재빌드하거나, 아예 uv pip install -e . (pyproject.toml 에 maturin 백엔드가 설정된 경우) 방식으로 링킹을 강제하는 것이 멘탈 방어에 유리합니다.

• ➔ Rust 코드를 컴파일해도 rust_engine 이 변하지 않는다면 uv pip install -e . 를 하자.*

pip에서 uv로 갈아타려고 찾아보다가 이 게시물을 발견했다면 맨 아래 여담 부분 확인 바란다.

uv 패키지 관리자

지금까지 우리는 pip 패키지 관리자를 사용하여 프로젝트를 관리했다. 그런데 Gemini 녀석이 프로젝트 관리와 관련해서 더 알아보아야 할 게 있다면 설명해 달라고 하니까 그제서야 다음과 같은 이야기를 한다.

1. 현대적인 의존성 관리 (uv & Cargo)
   파이썬 생태계에서는 최근 uv라는 도구가 표준으로 자리 잡고 있습니다. [cite: 2026-03-20]
   현대적 방식: pip 대신 Rust로 작성된 초고속 패키지 매니저인 uv를 사용하여 pyproject.toml로 의존성을 관리하세요. [cite: 2026-03-20]
   이점: Rust의 Cargo와 철학이 비슷하여 패키지 설치 속도가 압도적이며, 프로젝트별 가상 환경 관리가 매우 견고합니다. [cite: 2026-03-20]

처음 듣는 녀석인데, 하고 살펴보니 2024년에 출시되어 Python 생태계의 새로운 사실상 표준으로서 자리 잡아가고 있는 녀석이라고 한다.

Rust로 작성되어 매우 빠르고, (역시 Rust!) 기존에 여러 도구로 분산되어 있던 핵심 기능들을 단일 CLI로 통합하여 제공하고, 여러 가지 현대적인 기법과 최적화 알고리즘이 적용되어 있다고...

Gemini 녀석, 그걸 왜 이제 알려주는 거람. 오래된 버전 제시해 주는 거라거나 내가 인지한 이슈에 대해서는 보완해서 예제를 수정해 달라고 요청하며 학습하고 있었는데 uv는 내가 이 분야를 완전히 등지고 있던 사이에 출시된 녀석이라 전혀 인지하지 못하고 있었다.

하여간 이런 녀석을 인지한 순간 굳이 pip를 사용할 이유는 없지. 작업공간 생성 프로세스를 수정해 보자.

uv 명령어

먼저 uv를 설치해야 한다. 명령어 한 줄이면 금방 설치된다.

~$ curl -LsSf https://astral.sh/uv/install.sh | sh
downloading uv 0.11.5 aarch64-apple-darwin
installing to /Users/edenjint3927/.local/bin
  uv
  uvx
everything's installed!
>
To add $HOME/.local/bin to your PATH, either restart your shell or run:
>
    source $HOME/.local/bin/env (sh, bash, zsh)
    source $HOME/.local/bin/env.fish (fish)

(우리는 uv init 를 사용하지 않을 거지만) uv init 명령으로 프로젝트를 초기화한 뒤 uv add 로 필요한 패키지를 설치하면 자동으로 가상환경을 위한 .venv 디렉토리가 생성되며 이 프로젝트에서 사용 가능하게 로컬로 설치된다.

런타임에서 필요한 패키지는 그냥 uv add 로 설치하면 되고 빌드 타임 도구는 런타임까지 남겨 놓으면 프로젝트가 불필요하게 커지므로 uv add --dev 로 설치하여 빌드 타임에만 필요함을 명시한다.

기존에 추가한 패키지를 제거하고 싶으면 uv remove 를 사용하면 된다.

uv add 로 설치한 패키지들의 정보는 pyproject.toml 파일에 기록되며 uv sync 를 통해 현재 설치된 버전과 기록된 버전이 일치하는지 검증하고 로컬에 설치되지 않은 패키지를 설치할 수 있다. 나중에 프로젝트를 템플릿으로 복제해서 사용할 때는 uv add 과정을 생략하고 uv sync 를 통해 pyproject.toml 파일에 기록된 패키지를 설치하면 된다.

프로젝트를 처음 초기화하면 다음과 같은 구조가 자동 생성된다.

~/project$ ls -a
.        ..        .git        .gitignore    .python-version    main.py        pyproject.toml    README.md

uv 환경에서는 source .venv/bin/activate 와 같은 명령어를 통해 가상환경을 실행하여 직접 들어갈 필요 없이 uv run 명령어로 가상환경에서 Python을 실행할 수 있다.

uv run main.py 로 Python 서버를 실행할 수 있다지만 우리는 바닐라 Python 서버를 사용할 게 아니므로 그 실습은 패스한다. 언젠가 Python 테스트 코드를 작성하고 나서야 uv run test.py 라는 식으로 스크립트를 실행하게 되지 않을까.

Maturin 빌드는 마찬가지로 uv run 을 붙여 uv run maturin develop 로 실행하며 uvicorn 실행도 uv run uvicorn app.main:app 로 실행한다.

파일 상단에 #! /usr/bin/env uv run 같은 Shebang을 넣어서 실행 권한만 주면 uv run 을 생략할 수도 있다.

IDE 환경에서는 에디터 우측 하단의 파이썬 인터프리터 설정에서 프로젝트 폴더 내의 .venv/bin/python 을 지정해 주어야 한다고 하는데 프로젝터 규모가 커지기 전까지는 zsh+vim 환경에서 실습할 것이므로 그냥 알아만 두도록 하자.

앞서 우리는 uv init 을 사용하지 않을 거라고 했는데, uv init 으로 프로젝트의 뼈대를 잡은 후에 maturin init 을 하고자 하면 이미 생성된 프로젝트에는 maturin init 을 못 한다고 오류가 날 테니 uvx maturin init 으로 프로젝트 구조를 잡고 시작한다. uvx 는 아직 uv init 하지 않은 프로젝트에서 특정 패키지를 임시로 사용하고 보내준다. uvx maturin init 을 해주었으면 uv init 는 생략한다. 둘 다 pyproject.toml 를 생성하여 설계도를 만드는 역할이라 (그러면서 부수적인 파일들도 생성하지만) 중복해서 사용할 경우 충돌이 일어난다.

작업을 하다 환경이 꼬일 경우 rm -rf target .venv 실행 후 uv sync 하여 패키지를 다시 설치할 수 있다.

작업공간 생성 및 구조 확인

데이터베이스를 사용하는 프로젝트의 템플릿 만들었던 것부터 uv 활용으로 바꾸면 좋을 것 같다. 그 전의 예제들은 어차피 지난 예제니까 그대로 두고 지금부터 uv를 사용하면 되는 거니까.

~/workspace$ mkdir db-template & cd db-template
~/workspace/db-template$ uvx maturin init # 프로젝트 초기화
~/workspace/db-template$ # 선택지 중 기본값인 PyO3 선택
~/workspace/db-template$ uv add fastapi uvicorn orjson # 실행 의존성 추가
~/workspace/db-template$ uv add --dev maturin # 개발 의존성 추가
~/workspace/db-template$ uv sync # 의존성 검증 및 누락된 패키지 설치
~/workspace/db-template$ mkdir app && touch app/main.py # Python 파일 직접 생성
~/workspace/db-template$ tree -a -I .venv -I target
.
├── .github
│   └── workflows
│       └── CI.yml
├── .gitignore
├── app
│   └── main.py
├── Cargo.lock
├── Cargo.toml
├── pyproject.toml
├── src
│   └── lib.rs
└── uv.lock

사실 여기서는 기존에 작성한 코드를 가져올 것이기 때문에 app/main.py 를 생성하지 않아도 무방하지만 전체적인 작업 공간 확인을 위해 생성했다.

uv add 및 uv add --dev 를 통해 의존성 패키지를 설치하면 다음과 같이 pyproject.toml 파일의 적절한 위치에 해당 내용이 자동으로 기록된다.

pyproject.toml

[build-system]
requires = ["maturin>=1.12,<2.0"]
build-backend = "maturin"
>
[project]
name = "db-template"
requires-python = ">=3.8"
classifiers = [
    "Programming Language :: Rust",
    "Programming Language :: Python :: Implementation :: CPython",
    "Programming Language :: Python :: Implementation :: PyPy",
]
dynamic = ["version"]
dependencies = [
    "fastapi>=0.124.4",
    "orjson>=3.10.15",
    "uvicorn>=0.33.0",
]
>
[dependency-groups]
dev = [
    "maturin>=1.12.6",
]

코드 이전

Cargo.toml

[lib] 의 name 을 늘 하던 대로 "rust_engine" 으로 수정하고 기존 프로젝트의 [dev] 부분을 가져온다.

Cargo.toml

[package]
name = "db-template"
version = "0.1.0"
edition = "2024"
>
[lib]
name = "rust_engine"
crate-type = ["cdylib"]
>
[dependencies]
pyo3 = "0.28.0"
sqlx = { version = "0.8", features = ["runtime-tokio", "tls-rustls", "postgres", "macros"] }
tokio = { version = "1.43", features = ["full"] }
tracing-appender = "0.2"
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }
dotenvy = "0.15"

Rust 코드 및 Python 코드

기존 프로젝트에 있던 것을 긁어오면 된다.

~/workspace/db-template$ cp ../db-connection/src/* src
~/workspace/db-template$ cp ../db-connection/app/* app

Docker와 관련된 파일도 추가로 긁어오도록 하자.

~/workspace/db-template$ cp ../db-connection/.env .
~/workspace/db-template$ cp ../db-connection/docker-compose.yml .

빌드 및 실행

~/workspace/db-template$ docker compose up -d # PostgreSQL docker 서비스 실행
~/workspace/db-template$ uv run maturin develop # uv 가상환경에서 Rust 컴파일
~/workspace/db-template$ uv run uvicorn app.main:app --reload # uv 가상환경에서 서버 실행

INFO:     Started reloader process [6247] using StatReload
INFO:     Started server process [6249]
INFO:     Waiting for application startup.
INFO:     Rust 엔진 초기화 중...
2026-04-09T00:53:24.617492Z  INFO ThreadId(17) connect: rust_engine::db: PostgreSQL 18 연결 풀 초기화 완료
INFO:     엔진 초기화 및 DB 연결 성공
INFO:     Application startup complete.
^CINFO:     Shutting down
INFO:     Waiting for application shutdown.
INFO:     서버 종료 감지: 자원 정리 중...
2026-04-09T00:53:38.260590Z  INFO ThreadId(17) rust_engine: 엔진 종료 절차 시작
2026-04-09T00:53:38.260906Z  INFO ThreadId(17) close: rust_engine::db: PostgreSQL 18 연결 안전하게 종료
INFO:     모든 연결 안전하게 종료
INFO:     Application shutdown complete.
INFO:     Finished server process [6249]
INFO:     Stopping reloader process [6247]

기존 실습에서처럼 잘 작동하는 것을 확인할 수 있다.

여담

사실 이렇게 프로젝트를 새로 만들어서 복사해 올 거 없이 기존 프로젝트에서 uv add 및 uv add --dev 명령어로 pyproject.toml 파일에 의존성을 명시하고 바로 uv run uvicorn app.main:app --reload 해도 정상적으로 작동한다.

하지만 uvx maturin init 부터 시작하여 작업 환경을 구축하는 것도 필요한 과정이니 여기서는 처음부터 프로젝트를 다시 구축하는 걸로 진행했다.

만약 기존 프로젝트가 존재하고 pip에서 uv로 갈아타려고 하는 거라면 번거롭게 새로 만들 필요 없이 uv add 와 uv run 을 사용하자. (uv run 은 내부적으로 uv sync 를 먼저 실행한다.)

이전 실습에서는 데이터베이스 연결 자체에 집중하느라 구조적인 확장에 대해 고려하지 않고 코드를 작성해 보았다.

하지만 모든 기능이 src/lib.rs 파일 하나에 들어가면 프로젝트를 관리하기 어려워진다는 것을 ⟨모듈화⟩에서 언급한 바 있다.

따라서 본격적인 데이터베이스 실습에 들어가기에 앞서 프로젝트를 재구조화하도록 하겠다.

작업공간 조정 및 구조 확인

Tracing 및 로그 아카이빙 로직 관리를 담당할 src/logger.rs 파일과 PostgreSQL 연결 및 Pool 관리를 담당할 src/db.rs 파일을 생성하고 src/lib.rs 파일은 Python과의 인터페이스만 담당하도록 가볍게 유지할 것이다.

필요에 따라 공통 유틸리티 및 에러 처리를 담당할 src/common.rs 파일도 생성할 수 있지만 여기선 생략한다.

이후 각 프로젝트에서 필요한 로직이 있다면 이것을 템플릿삼아 해당 프로젝트를 위한 파일을 생성하여 사용하면 된다.

~/workspace/db-connection$ # 기존 구조 확인
~/workspace/db-connection$ tree -a -I venv -I target -I postgres_data -I logs
.
├── .env
├── .github
│   └── workflows
│       └── CI.yml
├── .gitignore
├── app
│   └── main.py
├── Cargo.lock
├── Cargo.toml
├── docker-compose.yml
├── pyproject.toml
└── src
    └── lib.rs
~/workspace/db-connection$ touch src/logger.rs src/db.rs
~/workspace/db-connection$ tree -a -I venv -I target -I postgres_data -I logs
.
├── .env
├── .github
│   └── workflows
│       └── CI.yml
├── .gitignore
├── app
│   └── main.py
├── Cargo.lock
├── Cargo.toml
├── docker-compose.yml
├── pyproject.toml
└── src
    ├── db.rs
    ├── lib.rs
    └── logger.rs

코드 분리

로그 관리 모듈

기존 코드에서는 server.log prefix가 붙은 로그 파일이 한 시간마다 생성되도록 구현되어 있었는데 Python 로그와 구분하기 위해 rust-engine.log prefix를 사용하고 하루 단위로 로그를 생성하도록 수정하겠다.

기존에는 콘솔에 출력할 것과 파일로 저장할 것에 대한 정보를 tracing_subscriber::registry() 에 with() 로 바로 전달하였는데 코드를 알아보기 쉽게 변수로 따로 작성하도록 하겠다. 추가로, 텍스트 파일에 ANSI 색상 코드가 들어가면 가독성이 떨어지므로 콘솔에서만 색상 코드를 적용하고 파일로 저장 시 생략하도록 하겠다.

src/logger.rs

use tracing_appender::non_blocking::WorkerGuard;
use tracing_subscriber::{fmt, prelude::*, EnvFilter};
>
pub fn init_tracing() -> WorkerGuard {
    let file_appender = tracing_appender::rolling::daily("logs", "rust-engine.log");
    let (non_blocking, guard) = tracing_appender::non_blocking(file_appender);
>
    // 로그 수준
    let filter = EnvFilter::try_from_default_env()
        .unwrap_or_else(|_| EnvFilter::new("info"));
>
    // 콘솔 출력
    let stdout_layer = fmt::layer()
        .with_thread_ids(true)
        .with_thread_names(true)
        .with_ansi(true);
>
    // 파일 저장
    let file_layer = fmt::layer()
        .with_writer(non_blocking)
        .json()
        .with_thread_ids(true)
        .with_ansi(false);
>
    tracing_subscriber::registry()
        .with(filter)
        .with(stdout_layer)
        .with(file_layer)
        .init();
>
    guard
}

데이터베이스 관리 모듈

데이터베이스 작업은 #[instrument] 속성을 붙여 로그를 남기되 연결 함수에는 skip(url) 을 사용하여 보안 이슈가 발생하지 않도록 한다.

src/db.rs

use sqlx::postgres::PgPoolOptions;
use sqlx::{Pool, Postgres};
use std::sync::OnceLock;
use tracing::{info, warn, error, instrument};
>
pub static DB_POOL: OnceLock<Pool<Postgres>> = OnceLock::new();
>
#[instrument(skip(url))]
pub async fn connect(url: &str) -> Result<(), sqlx::Error> {
    let pool = PgPoolOptions::new()
        .max_connections(10)
        .connect(url)
        .await
        .map_err(|e| {
            error!("PostgreSQL 18 연결 실패");
            e   
        })?;
>
    if DB_POOL.set(pool).is_ok() {
        info!("PostgreSQL 18 연결 풀 초기화 완료");
    } else {
        warn!("PostgreSQL 18 연결 풀이 이미 초기화되어 있음");
    }   
>
    Ok(())
}
>
#[instrument]
pub async fn close() {
    if let Some(pool) = DB_POOL.get() {
        pool.close().await;
>
        info!("PostgreSQL 18 연결 안전하게 종료");
    } else {
        warn!("종료할 DB 연결 풀이 존재하지 않음");
    }   
}
>
#[instrument]
pub fn is_alive() -> bool {
    match DB_POOL.get() {
        Some(pool) => !pool.is_closed(),
        None => false,
    }   
}

인터페이스 모듈

로그 관리 모듈과 데이터베이스 관리 모듈을 불러와 사용한다. 이후에 추가되는 로직들도 이곳에서 Python과 연결될 것이다.

src/lib.rs

mod db;
mod logger;
>
use pyo3::prelude::*;
use tokio::runtime::Runtime;
use dotenvy::dotenv;
use std::env;
use std::sync::OnceLock;
use tracing::{debug, error, info};
use tracing_appender::non_blocking::WorkerGuard;
>
static LOG_GUARD: OnceLock<WorkerGuard> = OnceLock::new();
>
#[pyfunction]
fn init_engine() -> PyResult<String> {
    info!("엔진 초기화 절차 시작");
>
    debug!("로깅 시작");
    if LOG_GUARD.get().is_none() {
        let guard = logger::init_tracing();
        let _ = LOG_GUARD.set(guard);
    }
>
    debug!("환경 변수 로드");
    dotenv().ok();
    let url = env::var("DATABASE_URL")
        .map_err(|e| {
            error!("DATABASE_URL 환경 변수를 찾을 수 없음: {}", e);
            pyo3::exceptions::PyRuntimeError::new_err("DATABASE_URL not found in .env")
        })?;
>   
    debug!("데이터베이스 연결 시작");
    let rt = Runtime::new().unwrap();
    rt.block_on(async {
        db::connect(&url).await
    })
    .map_err(|e| {
        error!("DB 연결 풀 생성 실패: {}", e);
        pyo3::exceptions::PyRuntimeError::new_err(format!("DB connection failed: {}", e))
    })?;
>
    Ok("엔진 초기화 및 DB 연결 성공".to_string())
}
>
#[pyfunction]
fn shutdown_engine() -> PyResult<()> {
    info!("엔진 종료 절차 시작");
>   
    let rt = Runtime::new().unwrap();
    rt.block_on(async {
        db::close().await;
    });
>
    Ok(())
}
>
#[pyfunction]
fn check_connection() -> PyResult<bool> {
    debug!("DB 연결 상태 확인");
>
    Ok(db::is_alive())
}
>
#[pymodule]
fn rust_engine(m: &Bound<'_, PyModule>) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(init_engine, m)?)?;
    m.add_function(wrap_pyfunction!(shutdown_engine, m)?)?;
    m.add_function(wrap_pyfunction!(check_connection, m)?)?;
>
    Ok(())
}

Python 코드

Rust 엔진의 수정된 함수명에 따라 Python 코드도 수정한다.

app/main.py

from fastapi import FastAPI, HTTPException
from fastapi.responses import ORJSONResponse
from contextlib import asynccontextmanager
import rust_engine
import logging
>
logger = logging.getLogger("uvicorn.error")
>
class UTF8ORJSONResponse(ORJSONResponse):
    media_type = "application/json; charset=utf-8"
>
@asynccontextmanager
async def lifespan(app: FastAPI):
    logger.info("Rust 엔진 초기화 중...")
    try:
        msg = rust_engine.init_engine()
        logger.info(msg)
    except Exception as e:
        logger.info(f"Rust 엔진 초기화 실패: {e}")
>
    yield # 앱 가동
>
    # [SHUTDOWN]
>
    logger.info("서버 종료 감지: 자원 정리 중...")
    try:
        rust_engine.shutdown_engine()
        logger.info("모든 연결 안전하게 종료")
    except Exception as e:
        logger.error(f"종료 중 오류 발생: {e}")
>
app = FastAPI(default_response_class=UTF8ORJSONResponse, lifespan=lifespan)
>
@app.get("/")
def read_root():
    return {
        "status": "200",
        "info": "서버 가동 중입니다."
    }
>
@app.get("/db-status")
def get_db_status():
    logger.info("DB 상태 체크 요청")
    is_alive = rust_engine.check_connection()
>
    if is_alive:
        logger.info("DB 연결 상태 양호")
        return {
            "status": "online",
            "message": "Rust 엔진이 PostgreSQL을 사용합니다."
        }
    else:
        logger.error("DB 연결 끊김 감지")
        raise HTTPException(
            status_code=500,
            detail="DB 연결이 끊겼거나 초기화되지 않았습니다."
        )

빌드 및 실행

docker 서비스가 내려가 있다면 다시 실행해 준다.

~/workspace/db-connection$ docker compose up -d 

docker 서비스를 내리고 싶을 땐 다음 명령어를 사용하면 된다.

~/workspace/db-connection$ docker compose down

Maturin 라이브러리를 통해 Rust 코드를 다시 컴파일한 후 uvicorn 라이브러리를 통해 FastAPI를 실행하면 이전과 동일하게 실행되는 것을 확인할 수 있다.

~/workspace/db-connection$ maturin develop
~/workspace/db-connection$ uvicorn app.main:app --reload

앞으로 데이터베이스를 다루는 예제에서는 이 프로젝트 디렉토리를 복제 후 수정하여 사용할 것이다.

많은 서버는 데이터베이스를 사용한다. 데이터베이스를 어떻게 사용할 수 있는지 알아보자.

여기서는 PostgreSQL을 사용할 것이다. PostgreSQL은 객체-관계형(ORDBMS) 데이터베이스의 표준이다. SONB 타입을 지원하여 NoSQL의 장점을 흡수했고, PostGIS(지리정보), pgvector(AI 벡터 검색) 등 확장이 무궁무진하다. 데이터 무결성에 있어서도 엄격하며 안정적이기에 가장 복잡한 비즈니스 로직을 가장 안정적인 성능으로 처리할 수 있다.

작업공간 생성 및 구조 확인

~/workspace$ mkdir db-connection && cd db-connection
~/workspace/db-connection$ python3 -m venv venv
~/workspace/db-connection$ source venv/bin/activate
~/workspace/db-connection$ # 평소에 설치하던 것 외에 추가된 라이브러리를 놓치지 말자
~/workspace/db-connection$ pip install maturin fastapi uvicorn orjson
~/workspace/db-connection$ maturin init
~/workspace/db-connection$ # 선택지 중 기본값인 PyO3 선택
~/workspace/db-connection$ # Cargo.toml과 src/lib.rs가 자동 생성된다
~/workspace/db-connection$ # Python 코드는 직접 생성해 주어야 한다
~/workspace/db-connection$ mkdir app && touch app/main.py
~/workspace/db-connection$ # ProgreSQL Docker를 위한 파일도 생성한다
~/workspace/db-connection$ touch docker-compose.yml .env
~/workspace/db-connection$ tree -a -I venv
.
├── .env
├── .github
│   └── workflows
│       └── CI.yml
├── .gitignore
├── app
│   └── main.py
├── Cargo.toml
├── docker-compose.yml
├── pyproject.toml
└── src
    └── lib.rs

Cargo.toml 파일을 열어 라이브러리 이름을 수정해 주겠다.

데이터베이스 연결을 위해서는 sqlx 크레이트가 필요하다. 그리고 데이터베이스 입출력은 비동기로 수행하는 게 효율적이므로 비동기 작업을 위한 tokio 크레이트도 사용한다. 각 크레이트에는 적절한 features 도 설정해 주겠다.

데이터베이스 설정 파일과 계정 정보를 분리하기 위해 환경변수 사용을 위한 dotenvy 크레이트도 사용한다.

로그를 기록하는 것을 배웠으니 tracing 크레이트를 비롯한 로그 크레이트도 추가하여 로그도 남겨 보겠다.

Cargo.toml

[package]
name = "image-processor"
version = "0.1.0"
edition = "2024"
>
[lib]
name = "rust_engine"
crate-type = ["cdylib"]
>
[dependencies]
pyo3 = "0.28.0"
sqlx = { version = "0.8", features = ["runtime-tokio", "tls-rustls", "postgres", "macros"] }
tokio = { version = "1.43", features = ["full"] }
tracing-appender = "0.2"
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }
dotenvy = "0.15"

코드 작성

Docker 파일

docker compose 를 통해 데이터베이스를 관리할 것이다.

현 시점 가장 안정적인 PostgreSQL 버전을 사용하며, 성능 모니터링을 위해 유용한 도구들을 함께 띄운다.

계정 정보는 .env 환경변수 파일에 작성하여 하드웨어 자원 할당과 계정 정보를 분리하여 관리할 것이다.

.env

# Docker Compose Project
COMPOSE_PROJECT_NAME=db_template
>
# PostgreSQL
POSTGRES_USER=peter
POSTGRES_PASSWORD=ku201711424
POSTGRES_DB=rust_python_db
POSTGRES_HOST=localhost
POSTGRES_PORT=5432
>
# pgAdmin
PGADMIN_EMAIL=admin@pjos.dev
PGADMIN_PASSWORD=admin201711424
>
# SQLX
DATABASE_URL=postgres://${POSTGRES_USER}:${POSTGRES_PASSWORD}@${POSTGRES_HOST}:${POSTGRES_PORT}/${POSTGRES_DB}

실무에서는 .env 파일의 내용을 절대 어딘가에 업로드하거나 유출하지 말 것. ...이라는 기본 보안 수칙을 인지하지 못한 채 개발을 하는 바이브코더들이 종종 이슈가 되더라. 당장 Gemini에게 설정 파일 작성하라고 해도 .env 파일을 따로 빼지 않고 설정 파일에 보안 관련 정보까지 다 집어넣어 버리더라.

데이터베이스 설정에 io_method=worker 커맨드를 추가하여 PostgreSQL 18의 비동기 IO를 사용한다. 또한 io_workers=8 커맨드로 코어 활용을 최적화한다.

데이터의 변화를 시각적으로 확인하기 위해 pgAdmin을 사용한다. 이를 통해 PostgreSQL 18에서 도입된 UUID v7이 실제 시간 순서대로 정렬되어 인덱스 효율을 높이는지 직접 눈으로 확인할 수 있으며, 복잡한 메타데이터(JSON)를 계층 구조로 편하게 분석할 수 있다. DB의 세션 상태와 I/O 부하를 실시간 그래프로 보며 성능 모니터링을 할 수 있다.

컨테이너 이름은 임의의 문자열을 사용하면 된다.

docker-compose.yml

services:
  db:
    image: postgres:18-alpine
    container_name: ${COMPOSE_PROJECT_NAME}_postgres
    environment:
      POSTGRES_USER: ${POSTGRES_USER}
      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
      POSTGRES_DB: ${POSTGRES_DB}
    ports:
      - "${POSTGRES_PORT}:5432"
    command:
      - "-c"
      - "io_method=worker"
      - "-c"
      - "io_workers=8"
    volumes:
      - ./postgres_data:/var/lib/postgresql
    restart: always
>
  pgadmin:
    image: dpage/pgadmin4
    container_name: pgadmin_ui
    environment:
      PGADMIN_DEFAULT_EMAIL: ${PGADMIN_EMAIL}
      PGADMIN_DEFAULT_PASSWORD: ${PGADMIN_PASSWORD}
    ports:
      - "8080:80"
    depends_on:
      - db
    restart: always
>
volumes:
  db_data:
    name: ${COMPOSE_PROJECT_NAME}_db_data

Rust 코드

PostgreSQL에 연결된 데이터베이스 커넥션 풀을 미리 생성해 놓고 필요할 때 하나씩 꺼내 사용하는 방식으로 구현한다. 데이터베이스 커넥션 풀은 OnceLock 을 이용하여 전역 선언하여 FastAPI 서버가 가동되는 동안 살아있도록 한다.

src/lib.rs

use pyo3::prelude::*;
use sqlx::postgres::PgPoolOptions;
use sqlx::{Pool, Postgres};
use std::sync::OnceLock;
use tokio::runtime::Runtime;
use dotenvy::dotenv;
use std::env;
use tracing::{info, debug, warn, error, instrument};
use tracing_appender::non_blocking::WorkerGuard;
use tracing_subscriber::{fmt, prelude::*, EnvFilter};
>
static DB_POOL: OnceLock<Pool<Postgres>> = OnceLock::new();
static LOG_GUARD: OnceLock<WorkerGuard> = OnceLock::new();
>
fn init_tracing() {
    let filter_appender = tracing_appender::rolling::hourly("./logs", "server.log");
    let (non_blocking, guard) = tracing_appender::non_blocking(filter_appender);
>
    let filter = EnvFilter::try_from_default_env()
        .unwrap_or_else(|_| EnvFilter::new("info"));
>
    let _ = LOG_GUARD.set(guard);
>
    tracing_subscriber::registry()
        .with(filter)
        .with(fmt::layer().with_thread_ids(true))               // 콘솔 출력
        .with(fmt::layer().with_writer(non_blocking).json())    // 파일 저장
        .init();
}
>
#[pyfunction]
#[instrument]
fn init_database() -> PyResult<String> {
    dotenv().ok();
>
    debug!("환경 변수 로드");
    let database_url = env::var("DATABASE_URL")
        .map_err(|e| {
            error!("DATABASE_URL 환경 변수를 찾을 수 없음: {}", e);
            pyo3::exceptions::PyRuntimeError::new_err("DATABASE_URL not found in .env")
        })?;
>
    let rt = Runtime::new().unwrap();
>
    rt.block_on(async {
        info!("PostgreSQL 연결 풀 생성 시작");
        let pool = PgPoolOptions::new()
            .max_connections(10)
            .connect(&database_url)
            .await
            .map_err(|e| {
                error!("데이터베이스 연결 풀 생성 실패: {}", e);
                pyo3::exceptions::PyRuntimeError::new_err(format!("DB 연결 실패: {}", e))
            })?;
>
        if DB_POOL.set(pool).is_ok() {
            info!("데이터베이스 연결 풀 성공적으로 초기화");
        } else {
            warn!("데이터베이스 연결 풀이 이미 초기화되어 있음");
        }
>
        Ok("PostgreSQL 연결 성공 및 Pool 초기화 완료".to_string())
    })
}
>
#[pyfunction]
#[instrument]
fn close_database() -> PyResult<()> {
    if let Some(pool) = DB_POOL.get() {
        info!("데이터베이스 연결 종료");
>
        let rt = Runtime::new().unwrap();
>
        rt.block_on(async {
            pool.close().await;
            info!("PostgreSQL 연결 풀 안전하게 닫음");
        });
    } else {
        warn!("정리할 데이터베이스 연결 풀이 존재하지 않음");
    }
>
    Ok(())
}
>
#[pyfunction]
fn check_connection() -> PyResult<bool> {
    debug!("데이터베이스 연결 상태 확인");
>
    if let Some(pool) = DB_POOL.get() {
        let closed = pool.is_closed();
        if closed {
            warn!("데이터베이스 풀이 닫혀 있음");
        } else {
            debug!("데이터베이스 풀 활성화 상태");
        }
        return Ok(!closed);
    }
>
    error!("데이터베이스 연결 풀이 존재하지 않음");
    Ok(false)
}
>
#[pymodule]
fn rust_engine(m: &Bound<'_, PyModule>) -> PyResult<()> {
    init_tracing();
    tracing::info!("Rust Engine 가동 및 Tracing 시스템 초기화 완료");
>
    m.add_function(wrap_pyfunction!(init_database, m)?)?;
    m.add_function(wrap_pyfunction!(close_database, m)?)?;
    m.add_function(wrap_pyfunction!(check_connection, m)?)?;
>
    Ok(())
}

Python 코드

서버 시작 시 데이터베이스 연결을 자동으로 수행하도록 설정한다. 서버 중단 시 데이터베이스 연결을 안전하게 종료하도록 설정한다. 이 작업은 lifespan 을 통해 작성할 수 있다.

app/main.py

from fastapi import FastAPI, HTTPException
from fastapi.responses import ORJSONResponse
from contextlib import asynccontextmanager
import rust_engine
import logging
>
logger = logging.getLogger("uvicorn.error")
>
class UTF8ORJSONResponse(ORJSONResponse):
    media_type = "application/json; charset=utf-8"
>
@asynccontextmanager
async def lifespan(app: FastAPI):
    logger.info("DB 연결 초기화 중...")
    try:
        msg = rust_engine.init_database()
        logger.info(msg)
    except Exception as e:
        logger.info(f"DB 연결 오류: {e}")
>
    yield # 앱 가동
>
    # [SHUTDOWN]
    logger.info("서버 종료 감지")
    try:
        rust_engine.close_database()
        logger.info("모든 연결 안전하게 종료")
    except Exception as e:
        logger.error(f"종료 중 오류 발생: {e}")
>
app = FastAPI(default_response_class=UTF8ORJSONResponse, lifespan=lifespan)
>
@app.get("/")
def read_root():
    return {
        "status": "200",
        "info": "서버 가동 중입니다."
    }
>
@app.get("/db-status")
def get_db_status():
    logger.info("DB 상태 체크 요청")
    is_alive = rust_engine.check_connection()
    if is_alive:
        logger.info("DB 연결 상태 양호")
        return {
            "status": "online",
            "message": "Rust 엔진이 PostgreSQL을 사용합니다."
        }
    else:
        logger.error("DB 연결 끊김 감지")
        raise HTTPException(
            status_code=500,
            detail="DB 연결이 끊겼거나 초기화되지 않았습니다."
        )

빌드 및 실행

도커 서비스를 실행한다.

~/workspace/db-connection$ docker compose up -d 
[+] up 31/31
 ✔ Image dpage/pgadmin4          Pulled                                                84.6s
 ✔ Image postgres:18-alpine      Pulled                                                95.9s
 ✔ Network db-connection_default Created                                               0.0s
 ✔ Container rust_progres_18     Created                                               0.3s
 ✔ Container pgadmin_ui          Created                                               0.0s

브라우저를 통해 다음 URL로 접속하여 환경변수에 작성한 pgAdmin 계정으로 로그인하면 데이터베이스 관리를 위한 웹 인터페이스에 접속할 수 있다.

• http://localhost:8080

여기서 [Add New Server] 혹은 [새 서버 추가] 버튼을 눌러 데이터베이스 정보를 입력한다.

Maturin 라이브러리를 통해 Rust 코드를 Python에서 호출 가능한 형태로 컴파일한다. 병렬 처리가 포함된 코드는 성능 최적화를 위해 --release 를 붙여 컴파일한다. 컴파일 후 pip list 명령어를 사용해 보면 Cargo.toml 파일에 작성한 패키지 이름을 확인할 수 있다.

uvicorn 라이브러리를 통해 FastAPI를 실행한다.

~/workspace/db-connection$ maturin develop
~/workspace/db-connection$ uvicorn app.main:app --reload

curl 명령어 또는 브라우저를 통해 다음과 같은 테스트를 해볼 수 있다.

• http://127.0.0.1:8000/db-status

~$ curl -i http://127.0.0.1:8000          
HTTP/1.1 200 OK
date: Tue, 07 Apr 2026 01:09:36 GMT
server: uvicorn
content-length: 53
content-type: application/json; charset=utf-8
>
{"status":"200","info":"서버 가동 중입니다."}% 

~$ curl -i http://127.0.0.1:8000/db-status
HTTP/1.1 200 OK
date: Tue, 07 Apr 2026 01:09:39 GMT
server: uvicorn
content-length: 77
content-type: application/json; charset=utf-8
>
{"status":"online","message":"Rust 엔진이 PostgreSQL을 사용합니다."}%

INFO:     Started reloader process [85335] using StatReload
2026-04-07T01:09:29.646312Z  INFO ThreadId(02) rust_engine: Rust Engine 가동 및 Tracing 시스템 초기화 완료
INFO:     Started server process [85337]
INFO:     Waiting for application startup.
INFO:     DB 연결 초기화 중...
2026-04-07T01:09:29.648668Z  INFO ThreadId(02) init_database: rust_engine: PostgreSQL 연결 풀 생성 시작
2026-04-07T01:09:29.688320Z  INFO ThreadId(02) init_database: rust_engine: 데이터베이스 연결 풀 성공적으로 초기화
INFO:     PostgreSQL 연결 성공 및 Pool 초기화 완료
INFO:     Application startup complete.
INFO:     127.0.0.1:55419 - "GET / HTTP/1.1" 200 OK
INFO:     DB 상태 체크 요청
INFO:     DB 연결 상태 양호
INFO:     127.0.0.1:55420 - "GET /db-status HTTP/1.1" 200 OK
^CINFO:     Shutting down
INFO:     Waiting for application shutdown.
INFO:     서버 종료 감지
2026-04-07T01:09:46.667943Z  INFO ThreadId(02) close_database: rust_engine: 데이터베이스 연결 종료
2026-04-07T01:09:46.669614Z  INFO ThreadId(02) close_database: rust_engine: PostgreSQL 연결 풀 안전하게 닫음
INFO:     모든 연결 안전하게 종료
INFO:     Application shutdown complete.
INFO:     Finished server process [85337]
INFO:     Stopping reloader process [85335]

RUST_LOG=debug 일 때

~$ curl -i http://127.0.0.1:8000                  
HTTP/1.1 200 OK
date: Tue, 07 Apr 2026 01:11:16 GMT
server: uvicorn
content-length: 53
content-type: application/json; charset=utf-8
>
{"status":"200","info":"서버 가동 중입니다."}%

~$ curl -i http://127.0.0.1:8000/db-status
HTTP/1.1 200 OK
date: Tue, 07 Apr 2026 01:11:18 GMT
server: uvicorn
content-length: 77
content-type: application/json; charset=utf-8
>
{"status":"online","message":"Rust 엔진이 PostgreSQL을 사용합니다."}%

INFO:     Started reloader process [85375] using StatReload
2026-04-07T01:11:10.113858Z  INFO ThreadId(02) rust_engine: Rust Engine 가동 및 Tracing 시스템 초기화 완료
INFO:     Started server process [85377]
INFO:     Waiting for application startup.
INFO:     DB 연결 초기화 중...
2026-04-07T01:11:10.115043Z DEBUG ThreadId(02) init_database: rust_engine: 환경 변수 로드
2026-04-07T01:11:10.115417Z  INFO ThreadId(02) init_database: rust_engine: PostgreSQL 연결 풀 생성 시작
2026-04-07T01:11:10.154654Z  INFO ThreadId(02) init_database: rust_engine: 데이터베이스 연결 풀 성공적으로 초기화
INFO:     PostgreSQL 연결 성공 및 Pool 초기화 완료
INFO:     Application startup complete.
INFO:     127.0.0.1:55422 - "GET / HTTP/1.1" 200 OK
INFO:     DB 상태 체크 요청
2026-04-07T01:11:18.652556Z DEBUG ThreadId(18) rust_engine: 데이터베이스 연결 상태 확인
2026-04-07T01:11:18.652603Z DEBUG ThreadId(18) rust_engine: 데이터베이스 풀 활성화 상태
INFO:     DB 연결 상태 양호
INFO:     127.0.0.1:55423 - "GET /db-status HTTP/1.1" 200 OK
^CINFO:     Shutting down
INFO:     Waiting for application shutdown.
INFO:     서버 종료 감지
2026-04-07T01:11:25.215181Z  INFO ThreadId(02) close_database: rust_engine: 데이터베이스 연결 종료
2026-04-07T01:11:25.215891Z  INFO ThreadId(02) close_database: rust_engine: PostgreSQL 연결 풀 안전하게 닫음
INFO:     모든 연결 안전하게 종료
INFO:     Application shutdown complete.
INFO:     Finished server process [85377]
INFO:     Stopping reloader process [85375]

이어서 스키마 설계 및 CRUB 단계로 넘어갈 예정이다.

로그를 터미널에 출력하고 넘어가면 실시간으로 관측하고 있지 않은 이상 문제를 파악하기 어렵다. 따라서 로그를 파일의 형태로 아카이빙할 필요가 있다.

여기서는 조금 더 다양한 상황을 연출하기 위해 병렬 처리뿐만 아니라 비동기 처리에 대한 시뮬레이션을 추가했다.

작업공간 생성 및 구조 확인

~/workspace$ mkdir log-archiving && cd log-archiving
~/workspace/log-archiving$ python3 -m venv venv
~/workspace/log-archiving$ source venv/bin/activate
~/workspace/log-archiving$ # 평소에 설치하던 것 외에 추가된 라이브러리를 놓치지 말자
~/workspace/log-archiving$ pip install maturin fastapi uvicorn orjson
~/workspace/log-archiving$ maturin init
~/workspace/log-archiving$ # 선택지 중 기본값인 PyO3 선택
~/workspace/log-archiving$ # Cargo.toml과 src/lib.rs가 자동 생성된다
~/workspace/log-archiving$ # Python 코드는 직접 생성해 주어야 한다
~/workspace/log-archiving$ mkdir app && touch app/main.py
~/workspace/log-archiving$ tree -I venv
.
├── app
│   └── main.py
├── Cargo.toml
├── pyproject.toml
└── src
    └── lib.rs

Cargo.toml 파일을 열어 라이브러리 이름을 수정해 주겠다.

로그 추적을 위한 tracing 크레이트와 로그를 필터링하기 위한 tracing-subscriber 크레이트, 로그를 파일에 기록하기 위한 tracing-appender 크레이트를 사용한다. features 에 "json" 을 추가해 주어야 데이터 분석 도구에서 사용하기 수월한 JSON 형태로 저장할 수 있다.

병렬 처리를 위한 rayon 크레이트와 더불어 비동기 처리를 위한 tokio 크레이트도 추가한다.

Cargo.toml

[package]
name = "log-archiving"
version = "0.1.0"
edition = "2024"
>
[lib]
name = "rust_engine"
crate-type = ["cdylib"]
>
[dependencies]
pyo3 = "0.28.0"
rayon = "1.11"
tokio = { version = "1.50", features = ["full"] }
tracing-appender = "0.2"
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }

코드 작성

Rust 코드

tracing_appender::non_blocking 을 사용하여 별도의 스레드가 매 시간마다 새로운 로그 파일을 생성하고 logs 폴더에 저장하도록 한다.

guard 는 프로그램이 갑자기 종료될 때 버퍼에 남아있던 로그들을 파일에 마저 쓰고 닫는 역할을 하는데, 이를 전역 변수에 저장하여 프로그램 종료 시까지 살려두어야 함수가 끝나는 즉시 사라지지 않고 프로그램이 끝날 때까지 살아 있을 수 있다.

std::sync::OnceLock 을 사용하면 Python이 실행되는 동안 로그 시스템이 죽지 않고 살아있게 된다.

src/lib.rs

use pyo3::prelude::*;
use rayon::prelude::*;
use tracing::{info, debug, warn, error, instrument, span, Level};
use tracing_appender::non_blocking::WorkerGuard;
use tracing_subscriber::{fmt, prelude::*, EnvFilter};
use std::sync::OnceLock;
use std::time::Duration;
>
static LOG_GUARD: OnceLock<WorkerGuard> = OnceLock::new();
>
fn init_tracing() {
    let filter_appender = tracing_appender::rolling::hourly("./logs", "server.log");
    let (non_blocking, guard) = tracing_appender::non_blocking(filter_appender);
>
    let filter = EnvFilter::try_from_default_env()
        .unwrap_or_else(|_| EnvFilter::new("info"));
>
    let _ = LOG_GUARD.set(guard);
>
    tracing_subscriber::registry()
        .with(filter)
        .with(fmt::layer().with_thread_ids(true))               // 콘솔 출력
        .with(fmt::layer().with_writer(non_blocking).json())    // 파일 저장
        .init();
}
>
/// 비동기 IO 시뮬레이션 함수
async fn mock_db_call(id: u32) {
    let _span = span!(Level::INFO, "db_query", id = id).entered();
    debug!("DB 데이터 조회 중...");
    tokio::time::sleep(Duration::from_millis(50)).await;
    info!("DB 조회 완료");
}
>
#[pyfunction]
#[instrument]
fn complex_task_runner(events: Vec<String>) -> PyResult<String> {
    info!("통합 태스크 러너 가동");
>
    // 비동기 처리
    let rt = tokio::runtime::Runtime::new().unwrap();
    rt.block_on(async {
        mock_db_call(101).await;
    });
>
    // 병렬 처리
    events.into_par_iter()
        .enumerate()
        .for_each(|(idx, name)| {
            let _sapn = span!(Level::DEBUG, "worker", id = idx).entered();
>
            if name.contains("error") {
                error!("심각한 페이로드 발견: {}", name);
            } else if name.len() > 20 {
                warn!("비정상적으로 긴 이벤트명 감지: {}", name);
            } else {
                // 이벤트 처리에 걸리는 시간 시뮬레이션
                std::thread::sleep(Duration::from_millis(10));
                debug!("일반 태스크 처리 중: {}", name);
            }
        });
>
    Ok("모든 작업 수행 및 로깅 완료".to_string())
}
>
#[pymodule]
fn rust_engine(m: &Bound<'_, PyModule>) -> PyResult<()> {
    init_tracing();
    tracing::info!("Rust Engine 가동 및 Tracing 시스템 초기화 완료");
>
    m.add_function(wrap_pyfunction!(complex_task_runner, m)?)?;
>
    Ok(())
}

Python 코드

app/main.py

import os
import time
from fastapi import FastAPI
from fastapi.responses import ORJSONResponse
>
os.environ["RUST_LOG"] = "debug"
>
import rust_engine
>
class UTF8ORJSONResponse(ORJSONResponse):
    media_type = "application/json; charset=utf-8"
>
app = FastAPI(default_response_class=UTF8ORJSONResponse)
>
@app.get("/")
def read_root():
    return {
        "status": "200",
        "info": "서버 가동 중입니다."
    }
>
@app.get("/run-complex-task")
async def run_task():
    events = [
        "normal_event_1",
        "critical_error_payload",
        "short",
        "very_long_event_name_for_warning",
        "normal_event_2"
    ]
>
    start = time.perf_counter()
    result = rust_engine.complex_task_runner(events)
    end = time.perf_counter()
>
    duration = end - start
>
    return {
        "status": "ok",
        "rust_result": result,
        "rust_pure_time": f"{duration:.4f} sec",
        "log_check": "프로젝트 루트의 /logs 디렉토리에서 확인"
    }

빌드 및 실행

Maturin 라이브러리를 통해 Rust 코드를 Python에서 호출 가능한 형태로 컴파일한다. 병렬 처리가 포함된 코드는 성능 최적화를 위해 --release 를 붙여 컴파일한다. 컴파일 후 pip list 명령어를 사용해 보면 Cargo.toml 파일에 작성한 패키지 이름을 확인할 수 있다.

uvicorn 라이브러리를 통해 FastAPI를 실행한다.

~/workspace/log-archiving$ maturin develop --release
~/workspace/log-archiving$ uvicorn app.main:app --reload

curl 명령어 또는 브라우저를 통해 다음과 같은 테스트를 해볼 수 있다.

• http://127.0.0.1:8000/run-complex-task

~$ curl -i http://127.0.0.1:8000/run-complex-task 
HTTP/1.1 200 OK
date: Sun, 05 Apr 2026 22:48:13 GMT
server: uvicorn
content-length: 169
content-type: application/json; charset=utf-8
>
{"status":"ok","rust_result":"모든 작업 수행 및 로깅 완료","rust_pure_time":"0.0635 sec","log_check":"프로젝트 루트의 /logs 디렉토리에서 확인"}% 

INFO:     Started reloader process [75975] using StatReload
2026-04-05T22:48:07.792253Z  INFO ThreadId(02) rust_engine: Rust Engine 가동 및 Tracing 시스템 초기화 완료
INFO:     Started server process [75977]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
2026-04-05T22:48:14.300101Z  INFO ThreadId(02) complex_task_runner{events=["normal_event_1", "critical_error_payload", "short", "very_long_event_name_for_warning", "normal_event_2"]}: rust_engine: 통합 태스크 러너 가동
2026-04-05T22:48:14.300434Z DEBUG ThreadId(02) complex_task_runner{events=["normal_event_1", "critical_error_payload", "short", "very_long_event_name_for_warning", "normal_event_2"]}:db_query{id=101}: rust_engine: DB 데이터 조회 중...
2026-04-05T22:48:14.352516Z  INFO ThreadId(02) complex_task_runner{events=["normal_event_1", "critical_error_payload", "short", "very_long_event_name_for_warning", "normal_event_2"]}:db_query{id=101}: rust_engine: DB 조회 완료
2026-04-05T22:48:14.352827Z  WARN ThreadId(26) worker{id=3}: rust_engine: 비정상적으로 긴 이벤트명 감지: very_long_event_name_for_warning
2026-04-05T22:48:14.352837Z ERROR ThreadId(22) worker{id=1}: rust_engine: 심각한 페이로드 발견: critical_error_payload
2026-04-05T22:48:14.362937Z DEBUG ThreadId(29) worker{id=0}: rust_engine: 일반 태스크 처리 중: normal_event_1
2026-04-05T22:48:14.362966Z DEBUG ThreadId(27) worker{id=4}: rust_engine: 일반 태스크 처리 중: normal_event_2
2026-04-05T22:48:14.363032Z DEBUG ThreadId(18) worker{id=2}: rust_engine: 일반 태스크 처리 중: short
INFO:     127.0.0.1:52324 - "GET /run-complex-task HTTP/1.1" 200 OK

로그에 출력된 events 인자가 길어 생략하고 싶다면 저번 실습에서처럼 #[instrument(skip(events))] 를 사용하면 되지만 어떤 인자가 전달되었는지도 로그 분석에 필요할 수 있으니 여기서는 남겨두는 방식으로 작성했다.

~/workspace/log-archiving$ tree -I venv -I target
.
├── app
│   └── main.py
├── Cargo.lock
├── Cargo.toml
├── logs
│   └── server.log.2026-04-05-22
├── pyproject.toml
└── src
    └── lib.rs
>
~/workspace/log-archiving$ cat logs/server.log.2026-04-05-22
{"timestamp":"2026-04-05T22:48:07.792269Z","level":"INFO","fields":{"message":"Rust Engine 가동 및 Tracing 시스템 초기화 완료"},"target":"rust_engine"}
{"timestamp":"2026-04-05T22:48:14.300122Z","level":"INFO","fields":{"message":"통합 태스크 러너 가동"},"target":"rust_engine","span":{"events":"[\"normal_event_1\", \"critical_error_payload\", \"short\", \"very_long_event_name_for_warning\", \"normal_event_2\"]","name":"complex_task_runner"},"spans":[{"events":"[\"normal_event_1\", \"critical_error_payload\", \"short\", \"very_long_event_name_for_warning\", \"normal_event_2\"]","name":"complex_task_runner"}]}
{"timestamp":"2026-04-05T22:48:14.300449Z","level":"DEBUG","fields":{"message":"DB 데이터 조회 중..."},"target":"rust_engine","span":{"id":101,"name":"db_query"},"spans":[{"events":"[\"normal_event_1\", \"critical_error_payload\", \"short\", \"very_long_event_name_for_warning\", \"normal_event_2\"]","name":"complex_task_runner"},{"id":101,"name":"db_query"}]}
{"timestamp":"2026-04-05T22:48:14.352541Z","level":"INFO","fields":{"message":"DB 조회 완료"},"target":"rust_engine","span":{"id":101,"name":"db_query"},"spans":[{"events":"[\"normal_event_1\", \"critical_error_payload\", \"short\", \"very_long_event_name_for_warning\", \"normal_event_2\"]","name":"complex_task_runner"},{"id":101,"name":"db_query"}]}
{"timestamp":"2026-04-05T22:48:14.352843Z","level":"WARN","fields":{"message":"비정상적으로 긴 이벤트명 감지: very_long_event_name_for_warning"},"target":"rust_engine","span":{"id":3,"name":"worker"},"spans":[{"id":3,"name":"worker"}]}
{"timestamp":"2026-04-05T22:48:14.352847Z","level":"ERROR","fields":{"message":"심각한 페이로드 발견: critical_error_payload"},"target":"rust_engine","span":{"id":1,"name":"worker"},"spans":[{"id":1,"name":"worker"}]}
{"timestamp":"2026-04-05T22:48:14.362949Z","level":"DEBUG","fields":{"message":"일반 태스크 처리 중: normal_event_1"},"target":"rust_engine","span":{"id":0,"name":"worker"},"spans":[{"id":0,"name":"worker"}]}
{"timestamp":"2026-04-05T22:48:14.362974Z","level":"DEBUG","fields":{"message":"일반 태스크 처리 중: normal_event_2"},"target":"rust_engine","span":{"id":4,"name":"worker"},"spans":[{"id":4,"name":"worker"}]}
{"timestamp":"2026-04-05T22:48:14.363044Z","level":"DEBUG","fields":{"message":"일반 태스크 처리 중: short"},"target":"rust_engine","span":{"id":2,"name":"worker"},"spans":[{"id":2,"name":"worker"}]}

서버를 가동하다가 문제가 발생했을 때 원인을 찾고 문제를 해결하기 위해서는 어디서 어떤 문제가 발생했는지 알아야 한다.

이를 위해 정상 작동과 그렇지 않은 상황에 대한 로그를 남겨 두는 것이 좋다.

우선 터미널에 로그를 찍는 것을 알아보자.

작업공간 생성 및 구조 확인

~/workspace$ mkdir log-tracing && cd log-tracing
~/workspace/log-tracing$ python3 -m venv venv
~/workspace/log-tracing$ source venv/bin/activate
~/workspace/log-tracing$ # 평소에 설치하던 것 외에 추가된 라이브러리를 놓치지 말자
~/workspace/log-tracing$ pip install maturin fastapi uvicorn orjson
~/workspace/log-tracing$ maturin init
~/workspace/log-tracing$ # 선택지 중 기본값인 PyO3 선택
~/workspace/log-tracing$ # Cargo.toml과 src/lib.rs가 자동 생성된다
~/workspace/log-tracing$ # Python 코드는 직접 생성해 주어야 한다
~/workspace/log-tracing$ mkdir app && touch app/main.py
~/workspace/log-tracing$ tree -I venv
.
├── app
│   └── main.py
├── Cargo.toml
├── pyproject.toml
└── src
    └── lib.rs

Cargo.toml 파일을 열어 라이브러리 이름을 수정해 주겠다.

Rust에서 로그를 남기기 위해서는 tracing 크레이트를 사용해야 한다. 그리고 로그를 출력하고 필터링하기 위해 tracing-subscriber를 사용한다. features = ["env-filter"] 를 명시해 주면 다시 컴파일할 필요 없이 환경변수만으로 필터를 조정할 수 있다.

유의미한 로그가 나오는 환경을 시뮬레이션하기 위해 병렬 처리를 위한 rayon도 사용한다.

Cargo.toml

[package]
name = "log-tracing"
version = "0.1.0"
edition = "2024"
>
[lib]
name = "rust_engine"
crate-type = ["cdylib"]
>
[dependencies]
pyo3 = "0.28.0"
rayon = "1.11"
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }

코드 작성

Rust 코드

Python에서 rust_engine 모듈을 가져올 때 최초로 한 번 로그에 어떤 스레드가 작업하는지 표시하도록 하며 EnvFilter 를 통해 환경변수에서 가져온 로그 수준을 반영하는 초기화 과정을 거친다.

#[instrument] 속성을 사용하면 함수 호출 시 자동으로 맥락(Span)을 생성하며 함수의 인자로 전달된 task_id 가 로그에 자동으로 찍힌다.

span!() 매크로를 사용하면 로그 수준에 따라 로그 앞에 부가적인 정보를 붙일 수도 있다.

지연 및 조건부 경고/에러에 대한 시뮬레이션을 작성한다.

src/lib.rs

use pyo3::prelude::*;
use rayon::prelude::*;
use tracing::{info, debug, warn, error, instrument, span, Level};
use tracing_subscriber::{fmt, prelude::*, EnvFilter};
use std::time::Duration;
>
fn init_tracing() {
    let filter = EnvFilter::try_from_default_env()
        .unwrap_or_else(|_| EnvFilter::new("info");
>
    let _ = tracing_subscriber::registry()
        .with(fmt::layer().with_thread_ids(true))
        .with(filter)
        .try_init();
}
>
#[pyfunction]
#[instrument(skip(events))] // 인자로 전달된 `events` 자체에 대한 건 생략
fn process_events_parallel(events: Vec<String>) -> PyResult<String> {
    info!("총 {}개의 이벤트 병렬 처리 시작", events.len());
>
    let results: Vec<bool> = events
        .into_par_iter()
        .enumerate()
        .map(|(idx, event_name)| {
            let event_span = span!(Level::DEBUG, "event_worker", id = idx);
            let _enter = event_span.enter();
>
            debug!("데이터 검증 중: {}", event_name);
>
            if event_name.contains("error") {
                error!("심각한 페이로드 발견: {}", event_name);
                false
            } else if event_name.len() > 10 {
                warn!("비정상적으로 긴 이벤트명 감지");
                true
            } else {
                // 이벤트 처리에 걸리는 시간 시뮬레이션
                std::thread::sleep(Duration::from_millis(10));
                true
            }
        })
        .collect();
>
    let success_count = results.iter().filter(|&&r| r).count();
>
    if success_count < results.len() {
        warn!("일부 작업이 실패했습니다. (성공: {} / 전체: {})", success_count, results.len());
    } else {
        info!("{}개의 모든 이벤트가 성공적으로 처리되었습니다.", results.len());
    }
>
    Ok(format!("처리 완료: {}건 성공", success_count))
}
>
#[pymodule]
fn rust_engine(m: &Bound<'_, PyModule>) -> PyResult<()> {
    init_tracing();
    tracing::info!("Rust Engine 가동 및 Tracing 시스템 초기화 완료");
>
    m.add_function(wrap_pyfunction!(process_events_parallel, m)?)?;
>
    Ok(())
}

Python 코드

Rust로 작성한 모듈을 불러오기 전에 환경 변수를 설정한다.

여기서 설정해주지 않을 경우 서버를 실행할 때 다음과 같이 환경 변수를 설정해 주어도 된다.

~/workspace/log-tracing$ RUST_LOG=debug uvicorn app.main:app --reload

app/main.py

import os
from fastapi import FastAPI
from fastapi.responses import ORJSONResponse
>
# rust_engine을 로드하기 전에 로그 수준 설정
# debug: 각 스레드의 상세 log 확인 가능
# info: 요약 정보만 확인 가능
os.environ["RUST_LOG"] = "debug"
>
import rust_engine
>
class UTF8ORJSONResponse(ORJSONResponse):
    media_type = "application/json; charset=utf-8"
>
app = FastAPI(default_response_class=UTF8ORJSONResponse)
>
@app.get("/")
def read_root():
    return {
        "status": "200",
        "info": "서버 가동 중입니다."
    }
>
@app.get("/process")
def process():
    test_events = [
        "login_event",
        "purchase_event_with_long_name",
        "system_error_critical",
        "logout_event",
        "something"
    ]
>
    result = rust_engine.process_events_parallel(test_events)
>    
    return {
        "status": "ok",
        "message": result
    }

빌드 및 실행

Maturin 라이브러리를 통해 Rust 코드를 Python에서 호출 가능한 형태로 컴파일한다. 병렬 처리가 포함된 코드는 성능 최적화를 위해 --release 를 붙여 컴파일한다. 컴파일 후 pip list 명령어를 사용해 보면 Cargo.toml 파일에 작성한 패키지 이름을 확인할 수 있다.

uvicorn 라이브러리를 통해 FastAPI를 실행한다.

~/workspace/log-tracing$ maturin develop --release
~/workspace/log-tracing$ uvicorn app.main:app --reload

curl 명령어 또는 브라우저를 통해 다음과 같은 테스트를 해볼 수 있다.

• http://127.0.0.1:8000/process

로그 수준이 DEBUG일 때

~$ curl -i http://127.0.0.1:8000/process
HTTP/1.1 200 OK
date: Thu, 02 Apr 2026 23:21:58 GMT
server: uvicorn
content-length: 54
content-type: application/json; charset=utf-8
>
{"status":"ok","message":"처리 완료: 4건 성공"}% 

INFO:     Started reloader process [65590] using StatReload
2026-04-02T23:21:55.889185Z  INFO ThreadId(01) rust_engine: Rust Engine 가동 및 Tracing 시스템 초기화 완료
INFO:     Started server process [65592]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
2026-04-02T23:21:59.932540Z  INFO ThreadId(02) process_events_parallel: rust_engine: 총 5개의 이벤트 병렬 처리 시작
2026-04-02T23:21:59.932686Z DEBUG ThreadId(16) event_worker{id=1}: rust_engine: 데이터 검증 중: purchase_event_with_long_name
2026-04-02T23:21:59.932693Z DEBUG ThreadId(13) event_worker{id=0}: rust_engine: 데이터 검증 중: login_event
2026-04-02T23:21:59.932686Z DEBUG ThreadId(15) event_worker{id=2}: rust_engine: 데이터 검증 중: system_error_critical
2026-04-02T23:21:59.932699Z DEBUG ThreadId(06) event_worker{id=3}: rust_engine: 데이터 검증 중: logout_event
2026-04-02T23:21:59.932721Z ERROR ThreadId(15) event_worker{id=2}: rust_engine: 심각한 페이로드 발견: system_error_critical
2026-04-02T23:21:59.932722Z  WARN ThreadId(06) event_worker{id=3}: rust_engine: 비정상적으로 긴 이벤트명 감지
2026-04-02T23:21:59.932733Z  WARN ThreadId(13) event_worker{id=0}: rust_engine: 비정상적으로 긴 이벤트명 감지
2026-04-02T23:21:59.932722Z  WARN ThreadId(16) event_worker{id=1}: rust_engine: 비정상적으로 긴 이벤트명 감지
2026-04-02T23:21:59.932737Z DEBUG ThreadId(05) event_worker{id=4}: rust_engine: 데이터 검증 중: something
2026-04-02T23:21:59.945342Z  WARN ThreadId(02) process_events_parallel: rust_engine: 일부 작업이 실패했습니다. (성공: 4 / 전체: 5)
INFO:     127.0.0.1:49647 - "GET /process HTTP/1.1" 200 OK

로그 수준이 INFO일 때 혹은 로그 수준을 명시하지 않았을 때

~$ curl -i http://127.0.0.1:8000/process
HTTP/1.1 200 OK
date: Thu, 02 Apr 2026 23:22:29 GMT
server: uvicorn
content-length: 54
content-type: application/json; charset=utf-8
>
{"status":"ok","message":"처리 완료: 4건 성공"}% 

INFO:     Started reloader process [65603] using StatReload
2026-04-02T23:22:27.262593Z  INFO ThreadId(01) rust_engine: Rust Engine 가동 및 Tracing 시스템 초기화 완료
INFO:     Started server process [65605]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
2026-04-02T23:22:30.232762Z  INFO ThreadId(02) process_events_parallel: rust_engine: 총 5개의 이벤트 병렬 처리 시작
2026-04-02T23:22:30.232913Z ERROR ThreadId(04) rust_engine: 심각한 페이로드 발견: system_error_critical
2026-04-02T23:22:30.232929Z  WARN ThreadId(15) rust_engine: 비정상적으로 긴 이벤트명 감지
2026-04-02T23:22:30.232932Z  WARN ThreadId(14) rust_engine: 비정상적으로 긴 이벤트명 감지
2026-04-02T23:22:30.232950Z  WARN ThreadId(16) rust_engine: 비정상적으로 긴 이벤트명 감지
2026-04-02T23:22:30.245526Z  WARN ThreadId(02) process_events_parallel: rust_engine: 일부 작업이 실패했습니다. (성공: 4 / 전체: 5)
INFO:     127.0.0.1:49648 - "GET /process HTTP/1.1" 200 OK

단순히 개수를 세거나 총합을 구하는 건 비교적 쉽지만 이전 행이나 다음 행과 비교하거나 몇 개 행 단위로 묶어서 연산을 수행할 경우 병렬 처리를 위해 구간을 나누면서 문제가 발생할 수 있다.

몇 개 행 단위로 묶어서 연산을 하는 것은 구간을 나눌 때 기준 행 수의 배수로 나누면 되니 비교적 수월하게 구현할 수 있지만 이전 행과의 비교는 어떻게든 다른 구간과의 비교가 필요하다.

이런 상황에서 어떻게 처리할 수 있는지 알아보기 위해 현재 행의 value 값이 이전 행의 value 값보다 큰지 비교하여 그 결과를 저장하는 예제를 살펴보도록 하겠다.

이번에는 결과를 전송하지 않고 비교 결과 항목이 추가된 새 CSV 파일을 생성해 볼 것이다.

이 실습은 이전 실습에서 이어서 진행한다. 따라서 작업공간 생성 및 구조 확인은 생략한다.

코드 작성

Rust 코드

연산 결과를 새 CSV 파일로 저장할 것이므로 CSV 출력에 필요한 크레이트를 추가로 불러와야 한다.

병렬 처리를 위해 구간을 나누는 함수는 그대로 재사용한다.

맨 앞이 아니라면 이전 청크의 마지막 값을 탐색해야 병렬 처리를 해도 연속적인 결과를 확인할 수 있다.

wtr 를 사용하는 영역을 블록 스코프로 묶었는데 그렇게 하지 않으면 buffer 를 반환할 때 소유권 문제가 생긴다. 블록 스코프로 묶지 않고 drop(wtr); 하는 것도 방법이다.

src/lib.rs

use pyo3::prelude::*;
use memmap2::Mmap;
use std::fs::File;
use std::io::{BufWriter, Write}; // NEW!
use rayon::prelude::*;
use csv::{ReaderBuilder, WriterBuilder, ByteRecord}; // MODIFIED!
>
/// 청크가 줄 중간에서 잘리지 않도록
/// 파일을 코어 수에 맞춰 최적의 경계로 분리
fn get_split_points(bytes: &[u8], num_chunks: usize) -> Vec<usize> {
    let mut points = vec![0];
    let base_size = bytes.len() / num_chunks;
>
    for i in 1..num_chunks {
        let target = i * base_size;
        let pos = bytes[target..].iter()
            .position(|&b| b == b'\n')
            .map(|p| target + p + 1)
            .unwrap_or(bytes.len());
>
        if pos < bytes.len() && !points.contains(&pos) {
            points.push(pos);
        }
    }
 >   
    points.push(bytes.len());
>
    points
}
>
#[pyfunction]
fn process_large_csv(file_path: String) -> PyResult<(f64, usize)> {
    let file = File::open(file_path)?;
    let mmap = unsafe {
        Mmap::map(&file)?
    };
    let bytes = &mmap[..];
>
    let split_points = get_split_points(bytes, rayon::current_num_threads());
>
    let result = split_points.windows(2) // (start, end) 쌍으로 사용
        .enumerate() // 인덱스와 요소를 튜플로 받아 사용: (index, (start, end))
        .collect::<Vec<_>>()
        .into_par_iter()
        .map(|(i, window)| {
            let (start, end) = (window[0], window[1]);
            let mut rdr = ReaderBuilder::new()
                .has_headers(i == 0)
                .from_reader(&bytes[start..end]);
>
            let mut record = ByteRecord::new();
            let (mut l_sum, mut l_count) = (0.0, 0);
>
            while rdr.read_byte_record(&mut record).unwrap_or(false) {
                // 0:id, 1:value, 2:category
                if record.get(2) == Some(b"A") {
                    if let Some(val_bytes) = record.get(1) {
                        if let Ok(s) = std::str::from_utf8(val_bytes) {
                            if let Ok(val) = s.parse::<f64>() {
                                l_sum += val;
                                l_count += 1;
                            }
                        }
                    }
                }
            }
>
            (l_sum, l_count)
        })
        .reduce(
            || (0.0, 0),
            |(s1, c1), (s2, c2)| (s1 + s2, c1 + c2)
        );
>
    Ok(result)
}
>
// 이상 기존 코드
>
#[pyfunction]
fn compare_and_write(input_path: String, output_path: String) -> PyResult<usize> {
    let file = File::open(&input_path)?;
    let mmap = unsafe {
        Mmap::map(&file)?
    };
    let bytes = &mmap[..];
>
    let split_points = get_split_points(bytes, rayon::current_num_threads());
>
    let processed_chunks: Vec<Vec<u8>> = split_points.windows(2) // (start, end) 쌍으로 사용
        .enumerate() // 인덱스와 요소를 튜플로 받아 사용: (index, (start, end))
        .collect::<Vec<_>>()
        .into_par_iter()
        .map(|(i, window)| {
            let (start, end) = (window[0], window[1]);
            let mut buffer = Vec::new();
            {
                let mut wtr = WriterBuilder::new()
                    .has_headers(false)
                    .from_writer(&mut buffer);
>
                let mut prev_value: Option<f64> = None;
>
                // Look-back: 이전 청크의 마지막 값 가져오기
                if i > 0 {
                    let search_start = if start > 1024 { start - 1024 } else { 0 };
                    let mut rdr_prev = ReaderBuilder::new()
                        .has_headers(false)
                        .from_reader(&bytes[search_start..start]); // 이전 청크
>
                    let mut last_rec = ByteRecord::new();
                    let mut last_val = None;
>
                    // 마지막 값 찾기
                    while rdr_prev.read_byte_record(&mut last_rec).unwrap_or(false) {
                        // 0:id, 1:value, 2:category
                        if let Ok(s) = std::str::from_utf8(last_rec.get(1).unwrap_or(b"0")) {
                            last_val = s.parse::<f64>().ok();
                        }
                    }
                    prev_value = last_val;
                }
>
                // 본격적인 작업 시작
>           
                let mut rdr = ReaderBuilder::new()
                    .has_headers(false)
                    .from_reader(&bytes[start..end]);
                let mut record = ByteRecord::new();
>
                while rdr.read_byte_record(&mut record).unwrap_or(false) {
                    // 0:id, 1:value, 2:category
                    let current_value = std::str::from_utf8(record.get(1).unwrap_or(b"0"))
                        .unwrap_or("0").parse::<f64>().unwrap_or(0.0);
>
                    let is_higher = match prev_value {
                        Some(v) if current_value > v => "1",
                        _ => "0",
                    };
>
                    record.push_field(is_higher.as_bytes());
                    wtr.write_byte_record(&record).unwrap();
>
                    prev_value = Some(current_value);
                }
>
                wtr.flush().unwrap();
            }
>
            buffer
        })
        .collect();
>
    let out_file = File::create(output_path)?;
    let mut buffered_writer = BufWriter::with_capacity(1024 * 1024, out_file);
    buffered_writer.write_all(b"id,value,category,is_higher\n")?;
    for chunk in processed_chunks {
        buffered_writer.write_all(&chunk)?;
    }
>
    Ok(mmap.len())
}
>
#[pymodule]
fn rust_engine(m: &Bound<'_, PyModule>) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(process_large_csv, m)?)?;
    m.add_function(wrap_pyfunction!(compare_and_write, m)?)?; // NEW!
>
    Ok(())
}

Python 코드

app/main.py

import pandas as pd
import numpy as np
from fastapi import FastAPI
from fastapi.responses import ORJSONResponse
import rust_engine
import time
import os
>
class UTF8ORJSONResponse(ORJSONResponse):
    media_type = "application/json; charset=utf-8"
>
app = FastAPI(default_response_class=UTF8ORJSONResponse)
>
@app.get("/")
def read_root():
    return {
        "status": "200",
        "info": "서버 가동 중입니다."
    }
>
@app.get("/setup-csv/{rows}")
def setup_csv(rows: int):
    file_path = "large_data.csv"
>
    df = pd.DataFrame({
        "id": range(rows),
        "value": np.random.rand(rows),
        "category": np.random.choice(["A", "B", "C", "D"], rows)
    })
>
    df.to_csv(file_path, index=False)
>
    return {
        "message": f"{rows:,}줄 CSV 파일 생성 완료"
    }
>
@app.get("/process-csv")
def process_csv():
    file_path = "large_data.csv"
>
    if not os.path.exists(file_path):
        return {
            "error": "파일이 없습니다. /setup-csv 라우트를 먼저 호출해주세요."
        }
>
    start_rust = time.perf_counter()
    rust_sum, rust_count = rust_engine.process_large_csv(file_path)
    end_rust = time.perf_counter()
>
    rust_duration = end_rust - start_rust
>
    start_pd = time.perf_counter()
    df = pd.read_csv(file_path)
    pd_sum = df[df["category"] == "A"]["value"].sum()
    end_pd = time.perf_counter()
>
    pd_duration = end_pd - start_pd
>
    return {
        "total_rows": rust_count,
        "rust_time": f"{rust_duration:.4f} sec",
        "pandas_time": f"{pd_duration:.4f} sec",
        "speed_up": f"{pd_duration / rust_duration:.1f}x",
        "results": {
            "rust_sum": rust_sum,
            "pandas_sum": pd_sum
        }
    }
>
# 이상 기존 코드
>
@app.get("/compare-and-write")
def compare_and_write():
    input_file = "large_data.csv"
    output_file = "processed_data.csv"
>
    if not os.path.exists(input_file):
        return {
            "error": "파일이 없습니다. /setup-csv 라우트를 먼저 호출해주세요."
        }
>
    start = time.perf_counter()
    processed_data = rust_engine.compare_and_write(input_file, output_file)
    end = time.perf_counter()
>
    duration = end - start
>
    return {
        "output_file": output_file,
        "rust_pure_time": f"{duration:.4f} sec",
        "file_size": f"{os.path.getsize(output_file) / (1024 * 1024):.2f} MB"
    }

빌드 및 실행

Maturin 라이브러리를 통해 Rust 코드를 Python에서 호출 가능한 형태로 컴파일한다. 병렬 처리가 포함된 코드는 성능 최적화를 위해 --release 를 붙여 컴파일한다. 컴파일 후 pip list 명령어를 사용해 보면 Cargo.toml 파일에 작성한 패키지 이름을 확인할 수 있다.

uvicorn 라이브러리를 통해 FastAPI를 실행한다.

~/workspace/csv-processor$ maturin develop --release
~/workspace/csv-processor$ uvicorn app.main:app --reload

curl 명령어 또는 브라우저를 통해 다음과 같은 테스트를 해볼 수 있다.

• http://127.0.0.1:8000/setup-csv/50000000
• http://127.0.0.1:8000/compare-and-write

~$ column -s, -t < workspace/csv-processor/large_data.csv | head -n 20
id        value                   category
0         0.40051801434356893     B
1         0.2546372973392159      B
2         0.8411876220186603      A
3         0.22976716598349933     C
4         0.2205761919396625      C
5         0.8969404327142123      C
6         0.24805162492147237     C
7         0.6157544951301185      B
8         0.37118755341262744     D
9         0.20077848283153787     C
10        0.7006307580777321      A
11        0.6289105208867691      C
12        0.5645651105257122      A
13        0.07251456319471494     C
14        0.4878721859338948      B
15        0.41749333189497495     B
16        0.793024114133093       B
17        0.684794684186633       C
18        0.7427233817197105      A

~$ curl -i http://127.0.0.1:8000/compare-and-write           
HTTP/1.1 200 OK
date: Thu, 02 Apr 2026 01:41:21 GMT
server: uvicorn
content-length: 91
content-type: application/json; charset=utf-8
>
{"output_file":"processed_data.csv","rust_pure_time":"0.7453 sec","file_size":"1528.15 MB"}%

~$ column -s, -t < workspace/csv-processor//processed_data.csv | head -n 20
id        value                   category  is_higher
0         0.40051801434356893     B         0
1         0.2546372973392159      B         0
2         0.8411876220186603      A         1
3         0.22976716598349933     C         0
4         0.2205761919396625      C         0
5         0.8969404327142123      C         1
6         0.24805162492147237     C         0
7         0.6157544951301185      B         1
8         0.37118755341262744     D         0
9         0.20077848283153787     C         0
10        0.7006307580777321      A         1
11        0.6289105208867691      C         0
12        0.5645651105257122      A         0
13        0.07251456319471494     C         0
14        0.4878721859338948      B         1
15        0.41749333189497495     B         0
16        0.793024114133093       B         1
17        0.684794684186633       C         0
18        0.7427233817197105      A         1

CSV 파일은 DBMS를 사용할 정도의 규모와 복잡도를 갖지 않은 간단하지만 많은 양의 데이터를 다룰 때 유용하게 사용된다.

CSV 파일의 행 단위 독립 연산을 빠르게 수행하는 방법을 알아보자.

작업공간 생성 및 구조 확인

~/workspace$ mkdir csv-processor && cd csv-processor
~/workspace/csv-processor$ python3 -m venv venv
~/workspace/csv-processor$ source venv/bin/activate
~/workspace/csv-processor$ # 평소에 설치하던 것 외에 추가된 라이브러리를 놓치지 말자
~/workspace/csv-processor$ pip install maturin fastapi uvicorn orjson numpy pandas
~/workspace/csv-processor$ maturin init
~/workspace/csv-processor$ # 선택지 중 기본값인 PyO3 선택
~/workspace/csv-processor$ # Cargo.toml과 src/lib.rs가 자동 생성된다
~/workspace/csv-processor$ # Python 코드는 직접 생성해 주어야 한다
~/workspace/csv-processor$ mkdir app && touch app/main.py
~/workspace/csv-processor$ tree -I venv
.
├── app
│   └── main.py
├── Cargo.toml
├── pyproject.toml
└── src
    └── lib.rs

Cargo.toml 파일을 열어 라이브러리 이름을 수정해 주겠다.

CSV 처리를 위한 Rust 생태계의 표준인 csv 크레이트와 데이터 직렬화를 담당하는 serde 크레이트를 추가한다.

features = ["derive"] 를 명시해 주어야 serde 크레이트의 매크로를 사용할 수 있음을 유의하자.

CSV 파일의 크기가 클 경우 MMAP을 사용하는 게 효율적이다. 여기서는 대용량 CSV 파일을 상정하고 실습을 진행할 것이므로 MMAP을 사용하기 위해 memmap2 크레이트도 사용한다.

CSV 파일 다루는 것 자체는 순차적으로 진행되지만 데이터를 파싱하고 계산하는 과정은 병렬화가 가능하므로 rayon도 사용한다.

Cargo.toml

[package]
name = "csv-processor"
version = "0.1.0"
edition = "2024"
>
[lib]
name = "rust_engine"
crate-type = ["cdylib"]
>
[dependencies]
pyo3 = "0.28.0"
rayon = "1.11"
memmap2 = "0.9"
csv = "1.4"
serde = { version = "1.0", features = ["derive"] }

코드 작성

Rust 코드

연습 수준에서는 모든 데이터를 벡터로 수집하여 처리해도 괜찮지만 데이터 양이 많아질 경우에는 메모리 이슈가 발생하므로 데이터를 메모리에 쌓지 않고 읽는 즉시 병렬 파이프라인으로 던지는 방식으로 스트리밍 병렬 처리를 하는 것이 효율적이다.

코어 수에 맞게 청크를 나눌 때 라인 중간에서 끊기면 안 되므로 최적의 경계로 분리하는 기준점을 제시하는 헬퍼 함수를 따로 작성한다.

Rust의 csv 생태계에서는 관습적으로 Reader를 rdr, Writer를 wtr 라는 약어로 사용한다는 건 여담.

src/lib.rs

use pyo3::prelude::*;
use memmap2::Mmap;
use std::fs::File;
use rayon::prelude::*;
use csv::{ReaderBuilder, ByteRecord};
>
/// 청크가 줄 중간에서 잘리지 않도록 파일을 코어 수에 맞춰 최적의 경계로 분리
fn get_split_points(bytes: &[u8], num_chunks: usize) -> Vec<usize> {
    let mut points = vec![0];
    let base_size = bytes.len() / num_chunks;
>
    for i in 1..num_chunks {
        let target = i * base_size;
        let pos = bytes[target..].iter()
            .position(|&b| b == b'\n')
            .map(|p| target + p + 1)
            .unwrap_or(bytes.len());
>
        if pos < bytes.len() && !points.contains(&pos) {
            points.push(pos);
        }
    }
 >   
    points.push(bytes.len());
>
    points
}
>
#[pyfunction]
fn process_large_csv(file_path: String) -> PyResult<(f64, usize)> {
    let file = File::open(file_path)?;
    let mmap = unsafe {
        Mmap::map(&file)?
    };
    let bytes = &mmap[..];
>
    let split_points = get_split_points(bytes, rayon::current_num_threads());
>
    let result = split_points.windows(2) // (start, end) 쌍으로 사용
        .enumerate() // 인덱스와 요소를 튜플로 받아 사용: (index, (start, end))
        .collect::<Vec<_>>()
        .into_par_iter()
        .map(|(i, window)| {
            let (start, end) = (window[0], window[1]);
            let mut rdr = ReaderBuilder::new()
                .has_headers(i == 0)
                .from_reader(&bytes[start..end]);
>
            let mut record = ByteRecord::new();
            let (mut l_sum, mut l_count) = (0.0, 0);
>
            while rdr.read_byte_record(&mut record).unwrap_or(false) {
                // 0:id, 1:value, 2:category
                if record.get(2) == Some(b"A") {
                    if let Some(val_bytes) = record.get(1) {
                        if let Ok(s) = std::str::from_utf8(val_bytes) {
                            if let Ok(val) = s.parse::<f64>() {
                                l_sum += val;
                                l_count += 1;
                            }
                        }
                    }
                }
            }
>
            (l_sum, l_count)
        })
        .reduce(
            || (0.0, 0),
            |(s1, c1), (s2, c2)| (s1 + s2, c1 + c2)
        );
>
    Ok(result)
}
>
#[pymodule]
fn rust_engine(m: &Bound<'_, PyModule>) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(process_large_csv, m)?)?;
>
    Ok(())
}

Python 코드

C언어 기반으로 최적화되어 있는 Pandas도 충분히 빠르지만 우리가 작성한 Rust 코드를 사용하여 데이터를 처리하는 것과 Pandas를 사용하여 데이터를 처리하는 것의 시간을 비교해 보자.

app/main.py

import pandas as pd
import numpy as np
from fastapi import FastAPI
from fastapi.responses import ORJSONResponse
import rust_engine
import time
import os
>
class UTF8ORJSONResponse(ORJSONResponse):
    media_type = "application/json; charset=utf-8"
>
app = FastAPI(default_response_class=UTF8ORJSONResponse)
>
@app.get("/")
def read_root():
    return {
        "status": "200",
        "info": "서버 가동 중입니다."
    }
>
@app.get("/setup-csv/{rows}")
def setup_csv(rows: int):
    file_path = "large_data.csv"
>
    df = pd.DataFrame({
        "id": range(rows),
        "value": np.random.rand(rows),
        "category": np.random.choice(["A", "B", "C", "D"], rows)
    })
>
    df.to_csv(file_path, index=False)
>
    return {
        "message": f"{rows:,}줄 CSV 파일 생성 완료"
    }
>
@app.get("/process-csv")
def process_csv():
    file_path = "large_data.csv"
>
    if not os.path.exists(file_path):
        return {
            "error": "파일이 없습니다. /setup-csv 라우트를 먼저 호출해주세요."
        }
>
    start_rust = time.perf_counter()
    rust_sum, rust_count = rust_engine.process_large_csv(file_path)
    end_rust = time.perf_counter()
>
    rust_duration = end_rust - start_rust
>
    start_pd = time.perf_counter()
    df = pd.read_csv(file_path)
    pd_sum = df[df["category"] == "A"]["value"].sum()
    end_pd = time.perf_counter()
>
    pd_duration = end_pd - start_pd
>
    return {
        "total_rows": rust_count,
        "rust_time": f"{rust_duration:.4f} sec",
        "pandas_time": f"{pd_duration:.4f} sec",
        "speed_up": f"{pd_duration / rust_duration:.1f}x",
        "results": {
            "rust_sum": rust_sum,
            "pandas_sum": pd_sum
        }
    }

빌드 및 실행

Maturin 라이브러리를 통해 Rust 코드를 Python에서 호출 가능한 형태로 컴파일한다. 병렬 처리가 포함된 코드는 성능 최적화를 위해 --release 를 붙여 컴파일한다. 컴파일 후 pip list 명령어를 사용해 보면 Cargo.toml 파일에 작성한 패키지 이름을 확인할 수 있다.

uvicorn 라이브러리를 통해 FastAPI를 실행한다.

~/workspace/csv-processor$ maturin develop --release
~/workspace/csv-processor$ uvicorn app.main:app --reload

curl 명령어 또는 브라우저를 통해 다음과 같은 테스트를 해볼 수 있다.

• http://127.0.0.1:8000/setup-csv/50000000
• http://127.0.0.1:8000/process-csv

~$ curl -i http://127.0.0.1:8000/setup-csv/50000000
HTTP/1.1 200 OK
date: Wed, 01 Apr 2026 05:11:26 GMT
server: uvicorn
content-length: 52
content-type: application/json; charset=utf-8
>
{"message":"50,000,000줄 CSV 파일 생성 완료"}%

~$ curl -i http://127.0.0.1:8000/process-csv       
HTTP/1.1 200 OK
date: Wed, 01 Apr 2026 05:12:02 GMT
server: uvicorn
content-length: 166
content-type: application/json; charset=utf-8
>
{"total_rows":12502433,"rust_time":"0.2692 sec","pandas_time":"4.8297 sec","speed_up":"17.9x","results":{"rust_sum":6252556.523183651,"pandas_sum":6252556.523183713}}%

부동소수점 연산 특성 상 연산 순서에 따라 소숫점 아래 낮은 자리 값은 오차가 있을 수 있다.

제로 카피 연산에서 단순히 더하고 곱하는 건 메모리 대역폭 싸움이었다면, 행렬 곱셈은 CPU 연산 능력과 데이터 재사용의 싸움이다.

64GB 램 안에서 가장 거대하고 빠른 행렬 연산을 구현해 보도록 하자.

작업공간 생성 및 구조 확인

~/workspace$ mkdir matrix-multiply && cd matrix-multiply
~/workspace/matrix-multiply$ python3 -m venv venv
~/workspace/matrix-multiply$ source venv/bin/activate
~/workspace/matrix-multiply$ # 평소에 설치하던 것 외에 추가된 라이브러리를 놓치지 말자
~/workspace/matrix-multiply$ pip install maturin fastapi uvicorn orjson numpy
~/workspace/matrix-multiply$ maturin init
~/workspace/matrix-multiply$ # 선택지 중 기본값인 PyO3 선택
~/workspace/matrix-multiply$ # Cargo.toml과 src/lib.rs가 자동 생성된다
~/workspace/matrix-multiply$ # Python 코드는 직접 생성해 주어야 한다
~/workspace/matrix-multiply$ mkdir app && touch app/main.py
~/workspace/matrix-multiply$ tree -I venv
.
├── app
│   └── main.py
├── Cargo.toml
├── pyproject.toml
└── src
    └── lib.rs

Cargo.toml 파일을 열어 라이브러리 이름을 수정해 주겠다.

이번 실습에서는 numpy 크레이트와 함께 ndarray 크레이트를 사용한다고 명시해 둔다. numpy 크레이트가 내부적으로 ndarray 크레이트를 가지고 사용하고 있어도 Rust의 의존성 관리 규칙에 의해 다음과 같은 이유로 명시적으로 작성해주는 것을 권장한다.

Cargo.toml

[package]
name = "matrix-multiply"
version = "0.1.0"
edition = "2024"
>
[lib]
name = "rust_engine"
crate-type = ["cdylib"]
>
[dependencies]
pyo3 = "0.28.0"
numpy = "0.28"
ndarray = "0.17"

코드 작성

Rust 코드

지금까지는 1차원 배열을 사용했지만 이번에는 행렬 연산을 수행할 것이기에 2차원 배열을 가져온다.

src/lib.rs

use pyo3::prelude::*;
use numpy::{PyReadonlyArray2, PyArray2};
>
#[pyfunction]
fn fast_matrix_multiply(
    py: Python<'_>,
    mat_a: PyReadonlyArray2<f64>,
    mat_b: PyReadonlyArray2<f64>
) -> Py<PyArray2<f64>> {
    let a = mat_a.as_array();
    let b = mat_b.as_array();
>
    let result = a.dot(&b);
>
    PyArray2::from_owned_array(py, result).unbind()
}
>
#[pymodule]
fn rust_engine(m: &Bound<'_, PyModule>) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(fast_matrix_multiply, m)?)?;
>
    Ok(())
}

Python 코드

app/main.py

import numpy as np
from fastapi import FastAPI
from fastapi.responses import ORJSONResponse
import rust_engine
import time
>
class UTF8ORJSONResponse(ORJSONResponse):
    media_type = "application/json; charset=utf-8"
>
app = FastAPI(default_response_class=UTF8ORJSONResponse)
>
@app.get("/")
def read_root():
    return {
        "status": "200",
        "info": "서버 가동 중입니다."
    }
>
@app.get("/matrix/{dim}")
def matrix_test(dim: int):
    a = np.random.rand(dim, dim).astype(np.float64)
    b = np.random.rand(dim, dim).astype(np.float64)
>
    start = time.perf_counter()
    result = rust_engine.fast_matrix_multiply(a, b)
    end = time.perf_counter()
>
    rust_duration = end - start
>
    return {
        "dimension": f"{dim}x{dim}",
        "total_elements": dim*dim,
        "rust_pure_time": f"Rust 연산에 걸린 시간: {rust_duration:.4f} sec",
        "result_sample": result[0, :3].tolist()
    }

빌드 및 실행

Maturin 라이브러리를 통해 Rust 코드를 Python에서 호출 가능한 형태로 컴파일한다. ndarray 크레이트의 dot() 연산이 내부적으로 병렬 처리를 지원하므로 성능 최적화를 위해 --release 를 붙여 컴파일한다. 컴파일 후 pip list 명령어를 사용해 보면 Cargo.toml 파일에 작성한 패키지 이름을 확인할 수 있다.

uvicorn 라이브러리를 통해 FastAPI를 실행한다.

~/workspace/matrix-multiply$ maturin develop --release
~/workspace/matrix-multiply$ uvicorn app.main:app --reload

curl 명령어 또는 브라우저를 통해 다음과 같은 테스트를 해볼 수 있다.

• http://127.0.0.1:8000/matrix/10000

~$ curl -i http://127.0.0.1:8000/matrix/10000
HTTP/1.1 200 OK
date: Mon, 30 Mar 2026 21:52:19 GMT
server: uvicorn
content-length: 190
content-type: application/json; charset=utf-8
>
{"dimension":"10000x10000","total_elements":100000000,"rust_pure_time":"Rust 연산에 걸린 시간: 37.1463 sec","result_sample":[2493.7760400667707,2484.6190913959954,2469.6911490233115]}%   

행렬 연산은 O(N³) 연산이므로 데이터 크기가 늘어날수록 세제곱에 비례하게 속도가 느려진다. 이 테스트는 약 2.4GB의 데이터를 다루며 1조 번의 곱셈과 덧셈을 수행한다. 데이터가 큰 만큼 오래 걸린 것 같아 보이지만 초당 약 538억 번의 부동 소수점 연산이 있었다. Accelerate 프레임워크 같은 하드웨어 가속 라이브러리를 연결하여 하드웨어 최적화를 하면 속도를 훨씬 더 줄일 수 있겠지만 여기서는 다루지 않겠다.

가변카피 첫 예제에서 데이터 양이 너무 많아지면 SWAP 공간에 대한 오버헤드가 발생한다는 사실을 확인해 보았다. 파일을 마치 메모리인 것처럼 다루는 MMAP을 사용하면 그런 오버헤드를 완화할 수 있다.

일반적인 파일 읽기는 File -> OS Buffer -> User Buffer 순으로 데이터를 복사하여 사용하는 반면, MMAP은 파일을 프로세스의 가상 주소 공간에 직접 매핑한다.

데이터를 명시적으로 read() 하지 않아도 되며 RAM 용량보다 훨씬 큰 파일을 읽는 경우에도 파일 크기만큼의 메모리 공간이 있는 것처럼 사용할 수 있다.

작업공간 생성 및 구조 확인

~/workspace$ mkdir memory-mapping && cd memory-mapping
~/workspace/memory-mapping$ python3 -m venv venv
~/workspace/memory-mapping$ source venv/bin/activate
~/workspace/memory-mapping$ # 평소에 설치하던 것 외에 추가된 라이브러리를 놓치지 말자
~/workspace/memory-mapping$ pip install maturin fastapi uvicorn orjson numpy
~/workspace/memory-mapping$ maturin init
~/workspace/memory-mapping$ # 선택지 중 기본값인 PyO3 선택
~/workspace/memory-mapping$ # Cargo.toml과 src/lib.rs가 자동 생성된다
~/workspace/memory-mapping$ # Python 코드는 직접 생성해 주어야 한다
~/workspace/memory-mapping$ mkdir app && touch app/main.py
~/workspace/memory-mapping$ tree -I venv
.
├── app
│   └── main.py
├── Cargo.toml
├── pyproject.toml
└── src
    └── lib.rs

Cargo.toml 파일을 열어 라이브러리 이름을 수정해 주겠다.

MMAP을 위해서는 memmap2 크레이트를 불러와야 한다.

Cargo.toml

[package]
name = "memory-mapping"
version = "0.1.0"
edition = "2024"
>
[lib]
name = "rust_engine"
crate-type = ["cdylib"]
>
[dependencies]
pyo3 = "0.28.0"
numpy = "0.28"
rayon = "1.11"
memmap2 = "0.9"

코드 작성

Rust 코드

파일 경로를 받아 그 파일을 MMAP으로 열고, 안에 들어있는 f64 데이터를 제로카피로 읽어 합계를 계산한다.

바이트 데이터를 f64 슬라이스로 변환하여 사용할 것이다. f64는 8바이트이므로, 전체 바이트를 8로 나눈 만큼의 슬라이스를 만들고 각 슬라이스의 합을 Rayon을 통해 병렬 연산한다.

src/lib.rs

use pyo3::prelude::*;
use memmap2::Mmap;
use std::fs::File;
use rayon::prelude::*;
>
#[pyfunction]
fn sum_mmap_file(file_path: String) -> PyResult<f64> {
    let file = File::open(file_path)?;
>
    let mmap = unsafe {
        Mmap::map(&file)?
    };
>
    let raw_ptr = mmap.as_ptr() as *const f64;
    let len = mmap.len() / 8;
>
    let slice = unsafe {
        std::slice::from_raw_parts(raw_ptr, len)
    };
>
    let total_sum = slice.par_iter().sum();
>
    Ok(total_sum)
}
>
#[pymodule]
fn rust_engine(m: &Bound<'_, PyModule>) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(sum_mmap_file, m)?)?;
>
    Ok(())
}

Python 코드

MMAP 테스트를 위해 거대한 바이너리 파일을 만드는 함수와 실제 MMAP 연산을 Rust에게 요청하는 함수를 각각 작성한다.

여기서는 전부 1.0 으로 채워진 GB 단위의 바이너리 파일을 사용하겠다.

app/main.py

import numpy as np
from fastapi import FastAPI
from fastapi.responses import ORJSONResponse
import rust_engine
import os
import time
>
class UTF8ORJSONResponse(ORJSONResponse):
    media_type = "application/json; charset=utf-8"
>
app = FastAPI(default_response_class=UTF8ORJSONResponse)
>
@app.get("/")
def read_root():
    return {
        "status": "200",
        "info": "서버 가동 중입니다."
    }
>
@app.get("/setup-mmap-file/{gb}")
def setup_file(gb: int):
    file_path = "huge_data.bin"
    size = gb * 1024 * 1024 * 1024
    elements = size // 8
>
    data = np.ones(elements, dtype=np.float64)
    data.tofile(file_path)
>
    return {
        "message": f"{gb}GB 파일 생성 완료",
        "path": file_path
    }
>
@app.get("/sum-mmap")
def sum_mmap():
    file_path = "huge_data.bin"
    if not os.path.exists(file_path):
        return {
            "error": "파일이 없습니다. /setup-mmap-file 라우트를 먼저 호출하세요."
        }
>
    start = time.perf_counter()
    result = rust_engine.sum_mmap_file(file_path)
    end = time.perf_counter()
>   
    rust_duration = end - start
>
    return {
        "file_size": f"{os.path.getsize(file_path) / (1024**3)} GB",
        "result": result,
        "rust_pure_time": f"Rust 연산에 걸린 시간: {rust_duration:.4f} sec"
    }

빌드 및 실행

Maturin 라이브러리를 통해 Rust 코드를 Python에서 호출 가능한 형태로 컴파일한다. 병렬 처리가 포함된 코드는 성능 최적화를 위해 --release 를 붙여 컴파일한다. 컴파일 후 pip list 명령어를 사용해 보면 Cargo.toml 파일에 작성한 패키지 이름을 확인할 수 있다.

uvicorn 라이브러리를 통해 FastAPI를 실행한다.

~/workspace/memory-mapping$ maturin develop --release
~/workspace/memory-mapping$ uvicorn app.main:app --reload

curl 명령어 또는 브라우저를 통해 다음과 같은 테스트를 해볼 수 있다.

• http://127.0.0.1:8000/setup-mmap-file/8
• http://127.0.0.1:8000/sum-mmap

~$ curl -i http://127.0.0.1:8000/setup-mmap-file/8
HTTP/1.1 200 OK
date: Mon, 30 Mar 2026 01:19:13 GMT
server: uvicorn
content-length: 61
content-type: application/json; charset=utf-8
>
{"message":"8GB 파일 생성 완료","path":"huge_data.bin"}%
>
~$ curl -i http://127.0.0.1:8000/sum-mmap         
HTTP/1.1 200 OK
date: Mon, 30 Mar 2026 01:19:18 GMT
server: uvicorn
content-length: 104
content-type: application/json; charset=utf-8
>
{"file_size":"8.0 GB","result":1073741824.0,"rust_pure_time":"Rust 연산에 걸린 시간: 0.5097 sec"}%

데이터 크기가 작으면 OS 페이지 캐시의 도움으로 매우 빠르게 완료된다. (약 15.6 GB/s)

~$ curl -i http://127.0.0.1:8000/setup-mmap-file/16
HTTP/1.1 200 OK
date: Mon, 30 Mar 2026 01:19:29 GMT
server: uvicorn
content-length: 62
content-type: application/json; charset=utf-8
>
{"message":"16GB 파일 생성 완료","path":"huge_data.bin"}%
>
~$ curl -i http://127.0.0.1:8000/sum-mmap          
HTTP/1.1 200 OK
date: Mon, 30 Mar 2026 01:19:34 GMT
server: uvicorn
content-length: 105
content-type: application/json; charset=utf-8
>
{"file_size":"16.0 GB","result":2147483648.0,"rust_pure_time":"Rust 연산에 걸린 시간: 2.8441 sec"}%

캐시의 도움을 받지 못하는 순수 SSD 읽기 속도로 수행된다. (약 5.6 GB/s)

~$ curl -i http://127.0.0.1:8000/setup-mmap-file/80
HTTP/1.1 200 OK
date: Mon, 30 Mar 2026 01:20:02 GMT
server: uvicorn
content-length: 62
content-type: application/json; charset=utf-8
>
{"message":"80GB 파일 생성 완료","path":"huge_data.bin"}%
>
~$ curl -i http://127.0.0.1:8000/sum-mmap          
HTTP/1.1 200 OK
date: Mon, 30 Mar 2026 01:20:51 GMT
server: uvicorn
content-length: 107
content-type: application/json; charset=utf-8
>
{"file_size":"80.0 GB","result":10737418240.0,"rust_pure_time":"Rust 연산에 걸린 시간: 34.3357 sec"}%

RAM(64GB) 크기를 초과하여 시간이 많이 걸린다. (약 2.3 GB/s) 100억 개의 데이터를 약 55초만에 처리했던 것(약 1.44 GB/s)에 비하면 많이 빨라졌지만.

크기가 작은 데이터는 메모리에 먼저 올려 놓고 사용하는 일반적인 제로카피가 더 빠른 연산이 가능하지만 데이터 크기가 커질수록 MMAP을 사용하는 게 이득이다.

이 실습은 이전 실습에서 이어서 진행한다. 따라서 작업공간 생성 및 구조 확인은 생략한다.

조건에 맞는 데이터의 인덱스를 뽑아내는 실습을 해보겠다.

코드 작성

Rust 코드

Rayon으로 병렬 필터링을 진행하며 enumerate() 로 인덱스와 데이터 쌍을 만들고 filter_map() 으로 조건에 맞는 인덱스만 수집한 뒤 결과 인덱스 배열을 Python 객체로 변환하여 반환한다.

src/lib.rs

use pyo3::prelude::*;
use numpy::{PyArray1, PyReadonlyArray1, PyArrayMethods};
use rayon::prelude::*;
>
#[pyfunction]
fn zero_copy_sum(array: PyReadonlyArray1<f64>) -> f64 {
    let slice = array.as_slice().expect("Numpy 슬라이스를 가져오는 데 실패했습니다.");
>
    slice.par_iter().sum()
}
>
#[pyfunction]
fn zero_copy_multiply(array: &Bound<'_, PyArray1<f64>>, factor: f64) {
    unsafe {
        let slice = array.as_slice_mut().expect("가변 슬라이스를 가져오는 데 실패했습니다.");
>
        slice.par_iter_mut().for_each(|x| *x *= factor);
    }
}
>
// 이상 기존 코드
>
#[pyfunction]
fn zero_copy_filter(py: Python<'_>, array: PyReadonlyArray1<f64>, threshold: f64) ->Py<PyArray1<u64>> {
    let slice = array.as_slice().expect("Numpy 슬라이스를 가져오는 데 실패했습니다.");
>
    let indices: Vec<u64> = slice
        .par_iter()
        .enumerate()
        .filter_map(|(idx, &val)| {
            if val > threshold {
                Some(idx as u64)
            } else {
                None
            }
        }).collect();
>
    PyArray1::from_vec(py, indices).unbind()
}
>
#[pymodule]
fn rust_engine(m: &Bound<'_, PyModule>) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(zero_copy_sum, m)?)?;
    m.add_function(wrap_pyfunction!(zero_copy_multiply, m)?)?;
    m.add_function(wrap_pyfunction!(zero_copy_filter, m)?)?; // NEW!
>
    Ok(())
}

Python 코드

app/main.py

import numpy as np
from fastapi import FastAPI
from fastapi.responses import ORJSONResponse
import rust_engine
import time
>
class UTF8ORJSONResponse(ORJSONResponse):
    media_type = "application/json; charset=utf-8"
>
app = FastAPI(default_response_class=UTF8ORJSONResponse)
>
@app.get("/")
def read_root():
    return {
        "status": "200",
        "info": "서버 가동 중입니다."
    }
>
@app.get("/compute/{size}")
def compute(size: int):
    data = np.random.rand(size)
>
    start = time.perf_counter()
    result = rust_engine.zero_copy_sum(data)
    end = time.perf_counter()
>
    rust_duration = end - start
>
    return {
        "size": len(data),
        "result": result,
        "rust_pure_time": f"Rust 연산에 걸린 시간: {rust_duration:.4f} sec"
    }
>
@app.get("/multiply/{size}/{factor}")
def multiply_inplace(size: int, factor: float):
    data = np.random.rand(size)
>
    start = time.perf_counter()
    result = rust_engine.zero_copy_multiply(data, factor)
    end = time.perf_counter()
>
    rust_duration = end - start
>
    return {
        "size": len(data),
        "factor": factor,
        "result_sample": data[:3].tolist(),
        "rust_pure_time": f"Rust 연산에 걸린 시간: {rust_duration:.4f} sec"
    }
>
# 이상 기존 코드
>
@app.get("/filter/{size}/{threshold}")
def filter_data(size: int, threshold: float):
    data = np.random.rand(size)
>
    start = time.perf_counter()
    indices = rust_engine.zero_copy_filter(data, threshold)
    end = time.perf_counter()
>
    rust_duration = end - start
>
    return {
        "original size": len(data),
        "filtered count": len(indices),
        "threshold": threshold,
        "sample_indices": indices[:5].tolist(),
        "rust_pure_time": f"Rust 연산에 걸린 시간: {rust_duration:.4f} sec"
    }

빌드 및 실행

Maturin 라이브러리를 통해 Rust 코드를 Python에서 호출 가능한 형태로 컴파일한다. 병렬 처리가 포함된 코드는 성능 최적화를 위해 --release 를 붙여 컴파일한다. 컴파일 후 pip list 명령어를 사용해 보면 Cargo.toml 파일에 작성한 패키지 이름을 확인할 수 있다.

uvicorn 라이브러리를 통해 FastAPI를 실행한다.

~/workspace/zero-copy$ maturin develop --release
~/workspace/zero-copy$ uvicorn app.main:app --reload

curl 명령어 또는 브라우저를 통해 다음과 같은 테스트를 해볼 수 있다.

• http://127.0.0.1:8000/filter/1000000000/0.9

~$ curl -i http://127.0.0.1:8000/filter/1000000000/0.9
HTTP/1.1 200 OK
date: Sat, 28 Mar 2026 00:09:11 GMT
server: uvicorn
content-length: 163
content-type: application/json; charset=utf-8
>
{"original size":1000000000,"filtered count":99997758,"threshold":0.9,"sample_indices":[8,36,39,56,57],"rust_pure_time":"Rust 연산에 걸린 시간: 0.1645 sec"}%

값이 0.9보다 큰 데이터만 뽑아낼 때 평균적으로 전체의 10% 정도의 데이터가 추출된다. 10억 번의 조건 분기가 0.1초대에 완료되었다.

이 실습은 이전 실습에서 이어서 진행한다. 따라서 작업공간 생성 및 구조 확인은 생략한다.

이전 실습에서는 Python의 Numpy 배열을 읽어서 사용하는 것만 다뤘다. 하지만 때로는 새로 생성하거나 수정하는 일도 있을 수 있다.

기존에 사용했던 PyReadonlyArray1 는 쓰기에 대한 고려를 하지 않아 빠르게 작업을 처리할 수 있지만 생성 및 수정 연산을 수행할 수 없다. 따라서 PyArray1 을 사용해야 한다.

코드 작성

Rust 코드

원본 Numpy 배열을 복사하지 않고 곱셈 연산을 하는 코드를 추가한다. &Bound 를 사용하여 Python 객체에 직접 접근할 수 있다.

컴파일러가 메모리 안전성을 보장할 수 없는 로우 레벨 작업을 할 때 사용되는 unsafe 블록 내부에서 Python 메모리 버퍼에 직접 가변 슬라이스로 접근한다.

호출 시점에 GIL(Global Interpreter Lock)이 보장되므로 Rust의 안전 가이드라인 안에서 Python 메모리를 직접 건드릴 수 있다.

as_slice_mut() 메서드를 사용하기 위해서는 PyArrayMethods 도 같이 불러와야 한다.

src/lib.rs

use pyo3::prelude::*;
use numpy::{PyArray1, PyReadonlyArray1, PyArrayMethods}; // MODIFIED!
use rayon::prelude::*;
>
#[pyfunction]
fn zero_copy_sum(array: PyReadonlyArray1<f64>) -> f64 {
    let slice = array.as_slice().expect("Numpy 슬라이스를 가져오는 데 실패했습니다.");
>
    slice.par_iter().sum()
}
>
// 이상 기존 코드
>
#[pyfunction]
fn zero_copy_multiply(array: &Bound<'_, PyArray1<f64>>, factor: f64) {
    unsafe {
        let slice = array.as_slice_mut().expect("가변 슬라이스를 가져오는 데 실패했습니다.");
>
        slice.par_iter_mut().for_each(|x| *x *= factor);
    }
}
>
#[pymodule]
fn rust_engine(m: &Bound<'_, PyModule>) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(zero_copy_sum, m)?)?;
    m.add_function(wrap_pyfunction!(zero_copy_multiply, m)?)?; // NEW!
>
    Ok(())
}

Python 코드

app/main.py

import numpy as np
from fastapi import FastAPI
from fastapi.responses import ORJSONResponse
import rust_engine
import time
>
class UTF8ORJSONResponse(ORJSONResponse):
    media_type = "application/json; charset=utf-8"
>
app = FastAPI(default_response_class=UTF8ORJSONResponse)
>
@app.get("/")
def read_root():
    return {
        "status": "200",
        "info": "서버 가동 중입니다."
    }
>
@app.get("/compute/{size}")
def compute(size: int):
    data = np.random.rand(size)
>
    start = time.perf_counter()
    result = rust_engine.zero_copy_sum(data)
    end = time.perf_counter()
>
    rust_duration = end - start
>
    return {
        "size": len(data),
        "result": result,
        "rust_pure_time": f"Rust 연산에 걸린 시간: {rust_duration:.4f} sec"
    }
>
# 이상 기존 코드
>
@app.get("/multiply/{size}/{factor}")
def multiply_inplace(size: int, factor: float):
    data = np.random.rand(size)
>
    start = time.perf_counter()
    result = rust_engine.zero_copy_multiply(data, factor)
    end = time.perf_counter()
>
    rust_duration = end - start
>
    return {
        "size": len(data),
        "factor": factor,
        "result_sample": data[:3].tolist(),
        "rust_pure_time": f"Rust 연산에 걸린 시간: {rust_duration:.4f} sec"
    }

빌드 및 실행

Maturin 라이브러리를 통해 Rust 코드를 Python에서 호출 가능한 형태로 컴파일한다. 병렬 처리가 포함된 코드는 성능 최적화를 위해 --release 를 붙여 컴파일한다. 컴파일 후 pip list 명령어를 사용해 보면 Cargo.toml 파일에 작성한 패키지 이름을 확인할 수 있다.

uvicorn 라이브러리를 통해 FastAPI를 실행한다.

~/workspace/zero-copy$ maturin develop --release
~/workspace/zero-copy$ uvicorn app.main:app --reload

curl 명령어 또는 브라우저를 통해 다음과 같은 테스트를 해볼 수 있다.

• http://127.0.0.1:8000/multiply/1000000000/0.5

~$ curl -i http://127.0.0.1:8000/compute/1000000000
HTTP/1.1 200 OK
date: Fri, 27 Mar 2026 00:54:42 GMT
server: uvicorn
content-length: 107
content-type: application/json; charset=utf-8
>
{"size":1000000000,"result":499981729.21686363,"rust_pure_time":"Rust 연산에 걸린 시간: 0.0686 sec"}%

~$ curl -i http://127.0.0.1:8000/multiply/1000000000/0.5
HTTP/1.1 200 OK
date: Fri, 27 Mar 2026 00:55:05 GMT
server: uvicorn
content-length: 169
content-type: application/json; charset=utf-8
>
{"size":1000000000,"factor":0.5,"result_sample":[0.48982013585054657,0.42512184525374297,0.2544803303002875],"rust_pure_time":"Rust 연산에 걸린 시간: 0.0833 sec"}% 

값을 수정할 때는 CPU 캐시와 실제 RAM 사이의 데이터를 동기화하는 오버헤드가 발생하여 같은 크기의 데이터에 대해 읽기 연산보다 조금 더 오래 걸리는 걸 확인할 수 있다.

큰 데이터를 다루게 될 경우 Python에서 Rust로 데이터를 옮겨 오는 게 병목 현상을 야기할 수 있다.

따라서 큰 데이터를 다룰 때는 데이터 자체를 복사하는 과정을 생략하고 메모리 주소만 넘겨 받아 사용하는 편이 효율적이다.

Rust의 rust-numpy 를 사용하면 Python의 Numpy 배열을 복사 없이 직접 읽고 쓸 수 있다.

작업공간 생성 및 구조 확인

~/workspace$ mkdir zero-copy && cd zero-copy
~/workspace/zero-copy$ python3 -m venv venv
~/workspace/zero-copy$ source venv/bin/activate
~/workspace/zero-copy$ # 평소에 설치하던 것 외에 추가된 라이브러리를 놓치지 말자
~/workspace/zero-copy$ pip install maturin fastapi uvicorn orjson numpy
~/workspace/zero-copy$ maturin init
~/workspace/zero-copy$ # 선택지 중 기본값인 PyO3 선택
~/workspace/zero-copy$ # Cargo.toml과 src/lib.rs가 자동 생성된다
~/workspace/zero-copy$ # Python 코드는 직접 생성해 주어야 한다
~/workspace/zero-copy$ mkdir app && touch app/main.py
~/workspace/zero-copy$ tree -I venv
.
├── app
│   └── main.py
├── Cargo.toml
├── pyproject.toml
└── src
    └── lib.rs

Cargo.toml 파일을 열어 라이브러리 이름을 수정해 주겠다.

Python의 Numpy 배열을 가져다 쓰기 위한 numpy 크레이트를 추가하고, 큰 데이터를 빠르게 처리할 수 있도록 병렬 처리를 위한 Rayon 크레이트, 그리고 데이터를 처리하는 시간동안 다른 작업이 멈추어 있지 않도록 비동기 처리를 위한 takio 크레이트와 pyo3_async_runtimes 크레이트도 추가해 준다.

여담이지만 numpy, pyo3_async_runtimes 와 같이 Python과 연관된 크레이트의 경우 대체로 PyO3와 버전이 함께 올라간다. PyO3의 버전이 올라감에 따라 변경점에 최적화하여 같이 변경되기 때문에 되도록이면 이들끼리 버전을 맞춰주는 게 좋다.

Cargo.toml

[package]
name = "zero-copy"
version = "0.1.0"
edition = "2024"
>
[lib]
name = "rust_engine"
crate-type = ["cdylib"]
>
[dependencies]
pyo3 = "0.28.0"
numpy = "0.28"
rayon = "1.11"
tokio = { version = "1.50", features = ["full"] }
pyo3-async-runtimes = { version = "0.28", features = ["tokio-runtime"] }

코드 작성

Rust 코드

Python의 1차원 Numpy 배열을 사용하기 위한 자료형을 가져온다. 실무에서는 PyArray1 과 PyReadonlyArray1 을 세트로 가져오는 경우가 많지만 우리는 배열을 생성하지는 않고 사용만 할 것이므로 후자만 가져와도 충분하다.

Numpy 배열을 복사 없이 그대로 사용하여 병렬 연산을 수행할 것이다.

src/lib.rs

use pyo3::prelude::*;
use numpy::PyReadonlyArray1;
use rayon::prelude::*;
>
#[pyfunction]
fn zero_copy_sum(array: PyReadonlyArray1<f64>) -> f64 {
    let slice = array.as_slice().expect("Numpy 슬라이스를 가져오는 데 실패했습니다.");
>
    slice.par_iter().sum()
}
>
#[pymodule]
fn rust_engine(m: &Bound<'_, PyModule>) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(zero_copy_sum, m)?)?;
>
    Ok(())
}

Python 코드

사용자로부터 입력받은 크기의 Numpy 배열을 임의로 생성하여 Rust에게 모든 배열 원소의 합을 구하라고 던져 준다.

데이터 생성 등으로 인한 오버헤드를 제외하고 순수 Rust 연산 시간을 알아보기 위해 time 라이브러리를 사용한다.

app/main.py

import numpy as np
from fastapi import FastAPI
from fastapi.responses import ORJSONResponse
import rust_engine
import time
>
class UTF8ORJSONResponse(ORJSONResponse):
    media_type = "application/json; charset=utf-8"
>
app = FastAPI(default_response_class=UTF8ORJSONResponse)
>
@app.get("/")
def read_root():
    return {
        "status": "200",
        "info": "서버 가동 중입니다."
    }
>
@app.get("/compute/{size}")
def compute(size: int):
    data = np.random.rand(size)
>
    start = time.perf_counter()
    result = rust_engine.zero_copy_sum(data)
    end = time.perf_counter()
>
    rust_duration = end - start
>
    return {
        "size": len(data),
        "result": result,
        "rust_pure_time": f"Rust 연산에 걸린 시간: {rust_duration:.4f} sec"
    }

빌드 및 실행

Maturin 라이브러리를 통해 Rust 코드를 Python에서 호출 가능한 형태로 컴파일한다. 병렬 처리가 포함된 코드는 성능 최적화를 위해 --release 를 붙여 컴파일한다. 컴파일 후 pip list 명령어를 사용해 보면 Cargo.toml 파일에 작성한 패키지 이름을 확인할 수 있다.

uvicorn 라이브러리를 통해 FastAPI를 실행한다.

~/workspace/zero-copy$ maturin develop --release
~/workspace/zero-copy$ uvicorn app.main:app --reload

curl 명령어 또는 브라우저를 통해 다음과 같은 테스트를 해볼 수 있다.

• http://127.0.0.1:8000/compute/100000000

~$ curl -i http://127.0.0.1:8000/compute/100000000 
HTTP/1.1 200 OK
date: Thu, 26 Mar 2026 00:49:03 GMT
server: uvicorn
content-length: 105
content-type: application/json; charset=utf-8
>
{"size":100000000,"result":50003983.53816322,"rust_pure_time":"Rust 연산에 걸린 시간: 0.0074 sec"}%    

~$ curl -i http://127.0.0.1:8000/compute/1000000000
HTTP/1.1 200 OK
date: Thu, 26 Mar 2026 00:49:08 GMT
server: uvicorn
content-length: 107
content-type: application/json; charset=utf-8
>
{"size":1000000000,"result":499998781.80647945,"rust_pure_time":"Rust 연산에 걸린 시간: 0.0681 sec"}%

~$ curl -i http://127.0.0.1:8000/compute/10000000000
HTTP/1.1 200 OK
date: Thu, 26 Mar 2026 00:49:22 GMT
server: uvicorn
content-length: 108
content-type: application/json; charset=utf-8
>
{"size":10000000000,"result":5000036618.422572,"rust_pure_time":"Rust 연산에 걸린 시간: 55.5093 sec"}% 

데이터 양이 너무 많아지면 SWAP 공간에 대한 오버헤드가 발생한다. 그런 경우에는 MMAP을 사용할 수 있는데 그것에 대해서는 이후에 알아보도록 하자.

FastAPI로부터 응답을 받았을 때 웹 브라우저에서 한글이 깨지는 경우 어떻게 해결할 수 있는지 알아보자.

원인

지금까지 우리가 작성한 코드는 텍스트 인코딩 방식을 명시하지 않았다.

보통의 웹사이트는 헤더 부분에 charset=utf-8 라는 메타 데이터를 가지고 있어 웹 브라우저가 웹사이트 정보를 받았을 때 UTF-8 방식으로 인코딩되어 있음을 인지할 수 있지만 그것이 명시되어 있지 않을 경우 시스템 기본 인코딩으로 보여준다.

이 '시스템 기본 인코딩'이 UTF-8이 아닌 다른 방식이었을 경우 인코딩 방식이 맞지 않아 잘못 해석된 텍스트가 출력된다.

해결책

어차피 개발 단계가 아닌 프로젝트 완성 단계에서는 백엔드의 출력을 웹 브라우저에서 직접 보여줄 일이 없으니 프론트엔드에서 메타 데이터를 제대로 명시하면 문제 없긴 하다.

하지만 개발 단계에서 출력된 데이터를 확인할 수 있어야 정상적으로 작동하는지 원활하게 확인할 수 있으니 적절한 코드로 변경한다.

우리의 이전 실습에서 한글 텍스트를 반환했으니 그 코드를 기준으로 코드를 수정해 보겠다.

수정 전

app/main.py (수정 전)

from fastapi import FastAPI
import rust_engine
import time
>
app = FastAPI()
>
@app.get("/")
def read_root():
    return {
        "status": "200",
        "info": "비동기 작업 중에도 응답할 수 있습니다."
    }
>
@app.get("/async-rust/{seconds}")
async def run_rust_task(seconds: int):
    start = time.time()
>
    message = await rust_engine.async_compute(seconds)
>
    end = time.time()
>
    return {
        "result": message,
        "duration": f"{end - start:.2f}s",
        "engine": "Rust (via pyo3-async-runtimes)"
    }

~/workspace/asynchronous$ curl -i http://127.0.0.1:8000/
HTTP/1.1 200 OK
date: Tue, 24 Mar 2026 23:01:06 GMT
server: uvicorn
content-length: 80
content-type: application/json
>
{"status":"200","info":"비동기 작업 중에도 응답할 수 있습니다."}% 

수정 후

ORJSONResponse를 불러온다. FastAPI의 기본 응답을 ORJSONResponse로 바꿔주기만 하면 된다.

이를 위해 orjson 라이브러리를 설치해야 한다.

~/workspace/asynchronous$ pip install orjson

기본 응답을 바꾸지 않고 각 함수의 반환값을 일일이

return ORJSONResponse(
    content=content,
    media_type="application/json; charset=utf-8")

와 같이 변경할 수도 있지만 기본 응답 자체를 바꿔주는 편이 효율적이다.

ORJSONResponse 클래스를 상속받아 media_type 값을 덮어씌운 클래스를 생성하고 FastAPI의 기본 응답을 그것으로 변경해 준다.

app/main.py (수정 후)

from fastapi import FastAPI
from fastapi.responses import ORJSONResponse
import rust_engine
import time
>
class UTF8ORJSONResponse(ORJSONResponse):
    media_type = "application/json; charset=utf-8"
>
app = FastAPI(default_response_class=UTF8ORJSONResponse)
>
@app.get("/")
def read_root():
    return {
        "status": "200",
        "info": "비동기 작업 중에도 응답할 수 있습니다."
    }
>
@app.get("/async-rust/{seconds}")
async def run_rust_task(seconds: int):
    start = time.time()
>
    message = await rust_engine.async_compute(seconds)
>
    end = time.time()
>
    return {
        "result": message,
        "duration": f"{end - start:.2f}s",
        "engine": "Rust (via pyo3-async-runtimes)"
    }

~$ curl -i http://127.0.0.1:8000/
HTTP/1.1 200 OK
date: Tue, 24 Mar 2026 23:04:15 GMT
server: uvicorn
content-length: 80
content-type: application/json; charset=utf-8
>
{"status":"200","info":"비동기 작업 중에도 응답할 수 있습니다."}%   

응답의 content-type 에 인코딩 정보가 추가되었다. 웹 브라우저로 테스트 해보아도 한글이 정상 출력되는 것을 확인할 수 있을 것이다.

비교

실습의 단순화를 위해 기본값을 사용했지만 사실 성능 측면에서도 ORJSONReponse를 사용하는 편이 이득이다.

여담

코드를 수정하기 전에도 웹 브라우저에는 한글이 깨져도 Swagger UI로 실행하면 응답의 한글이 잘 나온다.

이는 Swaager UI가 웹 브라우저 위에서 실행되는 웹앱인데 내부적으로 fetch() API를 사용하여 데이터를 가져온 뒤 UTF-8 기반으로 HTML 문서 안에 보기 좋게 출력해주기 때문이다.

Swagger UI의 기본 인코딩이 UTF-8이니 따로 명시해주지 않았어도 UTF-8로 해석하고 보여준 것이다.

지금까지 우리가 작성했던 예제 코드는 비동기 처리를 하는 FastAPI에서 동기 처리를 하는 Rust 코드를 호출해 Rust 연산이 끝날 때까지 Python 이벤트 루프가 멈출 위험이 있다.

Rust에서 무거운 작업을 실행할 경우 async 로 비동기로 처리하고 이를 Python의 await 와 연결하여 이 문제를 해결할 수 있다.

작업공간 생성 및 구조 확인

~/workspace$ mkdir asynchronous && cd asynchronous
~/workspace/asynchronous$ python3 -m venv venv
~/workspace/asynchronous$ source venv/bin/activate
~/workspace/asynchronous$ pip install maturin fastapi uvicorn
~/workspace/asynchronous$ maturin init
~/workspace/asynchronous$ # 선택지 중 기본값인 PyO3 선택
~/workspace/asynchronous$ # Cargo.toml과 src/lib.rs가 자동 생성된다
~/workspace/asynchronous$ # Python 코드는 직접 생성해 주어야 한다
~/workspace/asynchronous$ mkdir app && touch app/main.py
~/workspace/asynchronous$ tree -I venv
.
├── app
│   └── main.py
├── Cargo.toml
├── pyproject.toml
└── src
    └── lib.rs

Cargo.toml 파일을 열어 라이브러리 이름을 수정해 주겠다. 비동기 처리를 위한 takio 크레이트와 pyo3_async_runtimes 크레이트도 추가해 준다.

가장 최신 takio 크레이트는 1.50 버전인데 1.0으로 작성해도 충분하다. Cargo.toml 는 1.0이라고 적어놓은 버전을 "1.0 이상 2.0 미만의 최신 버전"으로 인식한다. takio는 1.0 버전 이후로 매우 엄격한 하위 호환성을 지키고 있는 크레이트이므로 1.0 버전 이후의 어떤 버전을 내려받더라도 문제 없이 작동할 것이다.

pyo3_async_runtimes 크레이트는 pyo3와 tokio 사이의 다리 역할을 하는데 pyo3 버전이 올라갈 때마다 그것에 맞춰 변경점을 반영하기 때문에 pyo3와 버전을 맞춰주는 게 좋다.

pyo3의 features 항목에는 "extension-module"을 추가하지 않아도 Maturin이 자동으로 추가해서 Python 모듈로 생성할 수 있지만 비동기 크레이트들에는 features 항목을 적어주어야 한다.

Cargo.toml

[package]
name = "asynchronous"
version = "0.1.0"
edition = "2024"
>
[lib]
name = "rust_engine"
crate-type = ["cdylib"]
>
[dependencies]
pyo3 = "0.28.0"
tokio = { version = "1.50", features = ["full"] }
pyo3-async-runtimes = { version = "0.28", features = ["tokio-runtime"] }

코드 작성

Rust 코드

pyo3_async_runtimes를 사용하여 Rust의 Future를 Python이 이해할 수 있는 Awaitable 객체(Coroutine)로 변환한다.

future_into_py 함수가 Rust의 tokio 런타임에서 실행되는 Future를 Python의 asyncio 루프로 던져주기 때문에 Python은 Rust가 일을 끝낼 때까지 기다리지 않고 다른 HTTP 요청을 받을 수 있다.

async 블록 내에서 await 을 사용하면 비동기 처리를 할 수 있다.

src/lib.rs

use pyo3::prelude::*;
use std::time::Duration;
use tokio::time::sleep;
>
#[pyfunction]
fn async_compute(py: Python<'_>, seconds: u64) -> PyResult<Bound<'_, PyAny>> {
    pyo3_async_runtimes::tokio::future_into_py(py, async move {
        sleep(Duration::from_secs(seconds)).await;
>
        let result = format!("코어에서 {}초간 비동기 연산을 무사히 마쳤습니다!", seconds);
>       
        Ok(result)
    })
}
>
#[pymodule]
fn rust_engine(m: &Bound<'_, PyModule>) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(async_compute, m)?)?;
>
    Ok(())
}

Python 코드

Rust에서 작성한 비동기 함수를 Python에서 async def 를 통해 정의한 함수처럼 사용할 수 있다.

app/main.py

from fastapi import FastAPI
import rust_engine
import time
>
app = FastAPI()
>
@app.get("/")
def read_root():
    return {
        "status": "200",
        "info": "비동기 작업 중에도 응답할 수 있습니다."
    }
>
@app.get("/async-rust/{seconds}")
async def run_rust_task(seconds: int):
    start = time.time()
>
    message = await rust_engine.async_compute(seconds)
>
    end = time.time()
>
    return {
        "result": message,
        "duration": f"{end - start:.2f}s",
        "engine": "Rust (via pyo3-async-runtimes)"
    }

빌드 및 실행

Maturin 라이브러리를 통해 Rust 코드를 Python에서 호출 가능한 형태로 컴파일한다. 컴파일 후 pip list 명령어를 사용해 보면 Cargo.toml 파일에 작성한 패키지 이름을 확인할 수 있다.

uvicorn 라이브러리를 통해 FastAPI를 실행한다.

~/workspace/asynchronous$ maturin develop
~/workspace/asynchronous$ uvicorn app.main:app --reload

curl 명령어 또는 브라우저를 통해 다음과 같은 테스트를 해볼 수 있다.

• http://127.0.0.1:8000/async-rust/10

전달한 숫자만큼의 시간이 지나기 전에 다른 창을 열고 다시 접속해 보면 대기 없이 접속되는 것을 확인할 수 있다.

Swagger UI를 통해 /async-rust 를 실행한 후 로딩 중인 동안 / 를 실행해 보는 식으로 테스트할 수도 있다.

프로젝트 규모가 커졌을 때 모든 Rust 코드를 src/lib.rs 파일 하나에서 관리하게 될 경우 스파게티 코드가 될 확률이 높으며 관리하기 어려워진다.

따라서 기능에 따라 파일을 나누어 모듈 단위로 관리할 필요가 있다.

작업공간 생성 및 구조 확인

~/workspace$ mkdir modularization && cd modularization
~/workspace/modularization$ python3 -m venv venv
~/workspace/modularization$ source venv/bin/activate
~/workspace/modularization$ pip install maturin fastapi uvicorn
~/workspace/modularization$ maturin init
~/workspace/modularization$ # 선택지 중 기본값인 PyO3 선택
~/workspace/modularization$ # Cargo.toml과 src/lib.rs가 자동 생성된다
~/workspace/modularization$ # Python 코드는 직접 생성해 주어야 한다
~/workspace/modularization$ mkdir app && touch app/main.py
~/workspace/modularization$ # Rust 모듈을 작성할 파일도 생성해 주어야 한다
~/workspace/modularization$ touch src/math_ops.rs src/data_ops.rs
~/workspace/modularization$ tree -I venv
.
├── app
│   └── main.py
├── Cargo.toml
├── pyproject.toml
└── src
    ├── data_ops.rs
    ├── lib.rs
    └── math_ops.rs

Cargo.toml 파일을 열어 라이브러리 이름을 수정해 주겠다. 우리가 작성할 모듈 중 하나는 병렬 처리 연산을 수행할 것이므로 병렬 처리를 위한 Rayon 크레이트도 추가해 준다.

Cargo.toml

[package]
name = "modularization"
version = "0.1.0"
edition = "2024"
>
[lib]
name = "rust_engine"
crate-type = ["cdylib"]
>
[dependencies]
pyo3 = "0.28.0"
rayon = "1.11"

코드 작성

Rust 코드

다른 파일에 작성된 함수는 #[pyfunction] 속성뿐만 아니라 pub 키워드도 있어야 src/lib.rs 에서 정상적으로 등록할 수 있다. 그 외에는 src/lib.rs 에서 작성하던 것과 크게 다르지 않다.

src/math_ops.rs

use pyo3::prelude::*;
use rayon::prelude::*;
>
#[pyfunction]
pub fn parallel_sum(data: Vec<i32>) -> PyResult<i64> {
    Ok(data.par_iter().map(|&x| x as i64).sum())
}

src/data_ops.rs

use pyo3::prelude::*;
use pyo3::exceptions::PyValueError;
use std::collections::HashMap;
>
#[pyfunction]
pub fn get_stats(data: Vec<i32>) -> PyResult<HashMap<String, i64>> {
    if data.is_empty() {
        return Err(PyValueError::new_err("데이터 비어 있습니다."));
    }
>
    let sum: i32 = data.iter().sum();
    let avg = sum as i64 / data.len() as i64;
>
    let mut map = HashMap::new();
    map.insert("sum".to_string(), sum as i64);
    map.insert("average".to_string(), avg);
>
    Ok(map)
}

다른 파일에서 작성한 코드를 src/lib.rs 에서 모듈로 불러올 땐 불러오고자 하는 파일 경로 앞에 mod 키워드를 사용한다.

모듈 이름과 함수 이름 사이에 :: 을 붙여 해당 모듈 내에 정의된 함수를 사용할 수 있다.

src/lib.rs

use pyo3::prelude::*;
>
mod math_ops;
mod data_ops;
>
#[pymodule]
fn rust_engine(m: &Bound<'_, PyModule>) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(math_ops::parallel_sum, m)?)?;
    m.add_function(wrap_pyfunction!(data_ops::get_stats, m)?)?;
>
    Ok(())
}

Python 코드

이번에도 정수 값을 담는 List 를 속성으로 가진 DataInput 클래스를 만들어 사용할 것이다. 이것은 데이터 검증 라이브러리 pydantic 의 BaseModel 클래스를 상속받아 생성한다.

app/main.py

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import rust_engine
>
app = FastAPI()
>
class NumbersInput(BaseModel):
    values: list[int]
>
@app.post("/compute")
def run_all(data: NumbersInput):
    try:
        p_sum = rust_engine.parallel_sum(data.values)
        stats = rust_engine.get_stats(data.values)
>
        return {
            "parallel_sum": p_sum,
            "statistics": stats
        }
>
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

빌드 및 실행

Maturin 라이브러리를 통해 Rust 코드를 Python에서 호출 가능한 형태로 컴파일한다. 병렬 처리가 포함된 코드는 성능 최적화를 위해 --release 를 붙여 컴파일한다. 컴파일 후 pip list 명령어를 사용해 보면 Cargo.toml 파일에 작성한 패키지 이름을 확인할 수 있다.

uvicorn 라이브러리를 통해 FastAPI를 실행한다.

~/workspace/modularization$ maturin develop --release
~/workspace/modularization$ uvicorn app.main:app --reload

curl 명령어 또는 브라우저를 통해 테스트를 해볼 수 있다.

이 예제의 경우 GET 메서드가 아닌 POST 메서드로 통신하므로 브라우저를 통한 테스트 시 Swagger UI를 사용해야 한다. FastAPI의 경우 다음과 같은 주소로 Swagger UI가 내장되어 있다.

• http://127.0.0.1:8000/docs

이 예제는 병렬 처리 연산이 핵심은 아니므로 따로 클라이언트 파일은 생성하지 않고 Swagger UI를 통해 테스트를 진행해도 충분하다.