rscc-common

RSCC 공통 Python 라이브러리. 응답 봉투, 에러 코드, X-Trace-Id 전파, ASGI 미들웨어, 회복탄력성(재시도·서킷 브레이커·격벽·TTL 캐시), 보안(JWT·AES-GCM·웹훅 서명·로그 리댁션), 한국 도메인 유틸을 제공한다.

• 기본 설치는 의존성 0 (표준 라이브러리만). FastAPI/pydantic 통합과 암호화는 extra 로 opt-in.
• Java(com.rscc:rscc-common-*)·JS(@rscc/common-*) 구현과 와이어 계약(응답 봉투·에러 코드· 헤더 규약 등)과 공유 테스트 벡터를 맞춘 폴리글랏 라이브러리의 Python 구현이다. 계약의 핵심 규칙은 각 모듈 docstring 에 요약돼 있어 이 패키지만으로 사용할 수 있다.
• Python >= 3.11 (상한 없음), 타입 힌트 포함(py.typed), 라이선스 MIT.
• 시간 단위는 초 (Java/JS 는 ms — 초 × 1000 == ms).

설치

pip install rscc-common                    # 기본 — 의존성 0
pip install "rscc-common[fastapi]"         # + fastapi>=0.110, pydantic>=2, httpx>=0.27
pip install "rscc-common[crypto]"          # + cryptography>=42 (AES-256/GCM)
pip install "rscc-common[fastapi,crypto]"  # 둘 다

extra 없이 해당 모듈을 임포트하면 그 모듈만 ImportError/ModuleNotFoundError 이고, 패키지 자체(import rscc_common)와 나머지 모듈은 계속 동작한다. 순수 ASGI 미들웨어 (install_exception_handlers 를 제외한 install_*)와 httpx 훅은 fastapi/httpx 를 임포트하지 않으므로 기본 설치로도 임포트된다.

PyPI 대신 git 태그로 직설치할 수도 있다 (X.Y.Z 는 설치할 릴리스 버전):

pip install "rscc-common[fastapi] @ git+https://github.com/Jeonghyeon-Ryu/r-common.git@py-vX.Y.Z#subdirectory=python/rscc-common"

모듈 개요

* = extra 필요. 무의존 모듈은 패키지 최상위에서 서브모듈 네임스페이스로도 쓸 수 있고 (from rscc_common import jwt), 자주 쓰는 심볼은 최상위에 플랫 재노출된다 (configure_logging, TraceIdFilter, install_* 계열, get_trace_id/set_trace_id/ reset_trace_id/new_trace_id, trace_request_hook, UserContext/get_user_context, ResultCode, CommonException 등).

웹·와이어 계약 (ASGI / FastAPI)

회복탄력성

로깅·보안

한국 도메인 유틸

빠른 시작

import logging

import httpx
from fastapi import FastAPI

from rscc_common import configure_logging, install_trace_middleware, trace_request_hook
from rscc_common.errors import CommonException, ResultCode
from rscc_common.handlers import install_exception_handlers   # [fastapi]
from rscc_common.response import CommonResponse               # [fastapi]

configure_logging(log_dir="logs", level=logging.INFO, app_name="order-api", quiet_uvicorn=True)
# → [2026-09-26 12:00:00] INFO  [3f9a1c2e] order_api - 주문 조회: 7
#   (traceId 는 요청 컨텍스트 밖이면 "-". uvicorn --workers > 1 이면 per_pid_file=True 필수)

app = FastAPI()
install_trace_middleware(app)      # X-Trace-Id 수신 재사용/생성 → contextvar → 응답 에코
install_exception_handlers(app)    # 예외 → 실패 봉투 (CommonException·검증 400·HTTPException·500)

client = httpx.Client(event_hooks={"request": [trace_request_hook]})  # 다운스트림 전파
log = logging.getLogger("order_api")

@app.get("/orders/{order_id}")
def get_order(order_id: int):
    log.info("주문 조회: %s", order_id)
    if order_id > 1000:
        raise CommonException(ResultCode.NOT_FOUND)             # → 404 + {"success": false, "code": "404", ...}
    return CommonResponse.ok({"id": order_id}).to_wire()        # 반드시 to_wire() 반환

회복탄력성 — 재시도 안에서 서킷 브레이커로 게이트하고, OPEN 차단은 재시도하지 않는다:

from rscc_common.circuit_breaker import CircuitBreaker, CircuitOpenError
from rscc_common.retry import retry_call

cb = CircuitBreaker(failure_threshold=5, open_duration=30.0)

data = retry_call(
    lambda attempt: cb.call(fetch_inventory, item_id),   # 비동기는 await cb.call_async(afetch, item_id)
    should_retry=lambda exc, attempt: not isinstance(exc, CircuitOpenError),
)

call/call_async 는 정상 반환 → 성공, Exception → 실패(같은 예외 재전파), 취소·인터럽트 등 그 밖의 BaseException → 판정 없는 반납(on_ignore)으로 정확히 1회 보고한다.

요청 밖(백그라운드 태스크 등)에서 traceId 잇기:

from rscc_common import new_trace_id, reset_trace_id, set_trace_id

token = set_trace_id(captured_trace_id or new_trace_id())
try:
    do_background_work()     # 이 안의 로그 라인·httpx 발신에 traceId 가 실린다
finally:
    reset_trace_id(token)

자체 로깅 구성(dictConfig 등)을 쓰는 앱은 configure_logging 대신 TraceIdFilter 를 핸들러에 붙이고 포맷에 %(trace_id)s 를 넣는다:

from rscc_common.logging import TraceIdFilter

handler = logging.StreamHandler()
handler.addFilter(TraceIdFilter())    # dictConfig: {"()": "rscc_common.logging.TraceIdFilter"}
handler.setFormatter(logging.Formatter("[%(asctime)s] %(levelname)-5s [%(trace_id)s] %(name)s - %(message)s"))

알려진 제약

• 미처리 예외의 500 응답과 X-Trace-Id — 트레이스 미들웨어는 Starlette ServerErrorMiddleware 안쪽에서 실행되므로 500 응답에 에코하지 못한다. install_exception_handlers 의 catch-all 이 수신 X-Trace-Id 가 유효하면 그 값을 500 응답 헤더에 직접 에코해 부분 보완한다. 수신 값이 없어 미들웨어가 새로 생성한 경우엔 500 응답에 헤더가 없고, catch-all 의 에러 로그 라인도 traceId 가 - 로 찍힌다(그 시점엔 contextvar 가 이미 정리됨). HTTPException·검증 오류 등 핸들러 처리 경로(4xx)는 정상 에코된다.
• CommonResponse 모델을 FastAPI 핸들러에서 직접 반환하지 말고 .to_wire() 를 반환할 것 — 직접 반환하면 data: null 이 실려 "데이터 없으면 키 생략" 계약이 깨진다.
• configure_logging
  • 기본 파일명({app_name}.log)은 프로세스 1개 전제 — uvicorn --workers > 1 등 멀티 프로세스면 per_pid_file=True ({app_name}.{pid}.log) 필수.
  • 재호출 시 자신이 부착한 핸들러만 교체하고 다른 루트 핸들러(OTel·Sentry 등)는 보존한다 (clear_root_handlers=True 면 전부 제거 — 이전 버전의 동작). 기본 포맷에 [traceId] 가 들어간다(include_trace_id=False 로 이전 포맷).
  • level 기본값은 logging.DEBUG — 운영은 logging.INFO 권장.
  • install_secret_redaction() 은 configure_logging 이후에 호출(재구성마다 다시 호출).
• 인메모리 상태(install_rate_limit, install_idempotency 기본 스토어, TtlCache, CircuitBreaker)는 프로세스(워커) 단위다 — 인스턴스 간 공유가 필요하면 분산 스토어를 IdempotencyStore Protocol 로 구현해 store= 로 주입한다.
• CircuitBreaker 를 수동 게이트(allow_request + 보고)로 쓸 때 보고가 누락되면 HALF_OPEN 프로브가 풀리지 않는다 — call/call_async 를 쓰면 보고가 보장된다.
• jwt 는 HS256 시크릿이 32바이트(UTF-8) 미만이면 ValueError (fail-fast, allow_weak_secret=True 는 교체 전 임시 옵트아웃).
• log_redact 는 정규식 기반 best-effort — 미탐·오탐이 공존한다. log_sanitize 와 병용 시 반드시 redact → sanitize(절단) 순서.
• rrn.is_valid_rrn 은 체크섬을 검사하지 않는다 — 2020-10 이후 발급분은 뒷자리가 임의번호라 mod-11 이 성립하지 않는다(checksum_ok_legacy 는 그 이전 발급분 전용). 저장 시 암호화 의무.
• text.chosung.of() 의 공백 스킵은 스페이스/탭만(Java 파리티).

버전·릴리스

버전의 단일 소스는 rscc_common.__version__ 이다. py-vX.Y.Z 태그 push 시 CI 가 태그와 버전 정합을 검증한 뒤 PyPI 에 배포한다(Trusted Publishing). 변경 이력은 소스 저장소의 CHANGELOG.md 에 있다.