rainbow-rb-utils 0.0.9.dev5

Rainbow Python Utilities

rb_utils

rb_utils는 백엔드 전반에서 재사용하는 유틸리티 모음 패키지입니다.

모듈/함수 상세

1) rb_utils.parser

• snake_to_camel(s: str) -> str: snake_case 문자열을 CamelCase 형태로 변환
• camel_to_snake(s: str) -> str: CamelCase 문자열을 snake_case로 변환
• t_to_dict(obj: Any) -> Any: 중첩 객체(FlatBuffer T, dict/list/tuple/set, bytes, numpy 배열)를 Python 기본 타입으로 재귀 변환
• _normalize_filter(filter_: Any) -> dict: None/dict/JSON 문자열을 공통 dict로 정규화
• to_json(obj: Any) -> str: t_to_dict 결과를 JSON 문자열로 직렬화
• to_iso(val: str | datetime) -> str: datetime이면 ISO 문자열, 문자열이면 그대로 반환

예시:

from rb_utils.parser import t_to_dict, to_json

payload_dict = t_to_dict(obj_payload)
payload_json = to_json(obj_payload)

2) rb_utils.flow_manager

• safe_eval_expr(expr, variables=None, get_global_variable=None) -> Any: 제한된 AST 규칙으로 안전 평가
• _pick_target_loop(func, provided) -> asyncio.AbstractEventLoop | None: callable 실행 대상 loop 선택
• call_with_matching_args(func, **provided) -> Any: 함수 시그니처에 맞춰 인자만 골라 호출 (sync/async 공통)
• eval_value(value, variables=None, get_global_variable=None) -> Any: 문자열/리스트/튜플/딕셔너리를 재귀 평가
• make_builtins_allow_most() -> dict: 위험 builtins 제외한 builtins dict 생성

예시:

from rb_utils.flow_manager import eval_value

value = eval_value("a + 10", variables={"local": {"a": 5}, "global": {}})
# value == 15

3) rb_utils.asyncio_helper

• spawn(coro, *, name=None, log_ex=True): 현재 loop에 task 생성 후 완료/예외 로깅
• _ensure_private_loop() -> asyncio.AbstractEventLoop: 전용 백그라운드 loop 확보
• fire_and_log(coro, *, name=None, loop=None): 현재 loop/외부 loop/전용 loop에 안전하게 실행
• _attach_done_logging(fut_or_task, *, name=None): Future/Task 완료 시 예외 로깅 callback 연결

예시:

from rb_utils.asyncio_helper import fire_and_log

fire_and_log(socket_client.emit("muscat/program/log", {"msg": "hello"}), name="program-log")

4) rb_utils.service_exception

• class ServiceException(Exception):
  • 필드: message: str, status_code: int | None
  • __str__(): message (status=...) 형태 출력
  • to_dict(): {message, status_code} 반환

예시:

from rb_utils.service_exception import ServiceException

raise ServiceException("invalid request", status_code=400)

5) rb_utils.date

• is_valid_date(d): date/str/int(timestamp) 유효성 검사
• timestamp_ms_to_seconds(timestamp, tz=UTC): ms/seconds timestamp를 seconds로 정규화
• is_has_time(s: str): 문자열에 시간 정보 포함 여부 판단
• parse_date_with_time(mode, d, tz=UTC): 날짜/문자열/타임스탬프를 timezone-aware datetime으로 변환
• ensure_datetime(dt, tz): date | datetime을 지정 tz의 datetime으로 보정
• _tz(tz_name): timezone 캐시 조회
• convert_dt(local, in_tz, out_tz): 입력 tz 기준 datetime/date를 출력 tz로 변환
• get_current_dt_yyyymmddhhmmss(tz): 현재 시각을 YYYYMMDDHHMMSS 문자열 반환

예시:

from rb_utils.date import parse_date_with_time
from datetime import UTC

dt = parse_date_with_time("start", "2026-02-25", tz=UTC)

6) rb_utils.file

• content_disposition_header(filename) -> str: 다운로드용 Content-Disposition 헤더 생성
• sanitize_filename(name, default="export.csv", max_name_len=100) -> str: OS 금지문자 제거/길이 제한
• get_env_path() -> Path | None: 실행 위치 기준으로 .env 파일 탐색

예시:

from rb_utils.file import sanitize_filename

safe_name = sanitize_filename("../../report?.csv")

7) rb_utils.pagination

• class PageInfo(TypedDict): 페이지 응답 메타 타입 (mode/page/pages/limit/total/sort/order)
• class LogsResponse(TypedDict): 로그 목록 응답 타입 (items, pageInfo)

8) rb_utils.helper

• ManipulateZenohResHelper(obj): Zenoh 응답 객체 보정 유틸 의도 함수

주의:

• 현재 구현은 dict 처리가 아닌 .hasAttr/.set 접근과 즉시 ValueError를 포함하므로, 사용 전 동작 검증이 필요합니다.

실무 사용 패턴

• FlatBuffer/SDK 응답 평탄화: t_to_dict
• Flow step 동적 인자 평가: eval_value, safe_eval_expr
• 비동기 fire-and-forget + 예외 로깅: fire_and_log
• 서비스 레이어 공통 예외: ServiceException
• 파일 다운로드/이름 정리: content_disposition_header, sanitize_filename

공통 주의사항

• flow_manager 평가기는 허용된 연산/함수만 실행합니다.
• timezone 변환 함수는 Asia/Seoul 같은 IANA tz 문자열을 명시해야 합니다.