Pre-release
This release is a pre-release and may not be stable for production use.
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.env
env_get(key, default=None, start=None):os.environ우선, 이후 가까운.env.development→.env순서로 값 조회load_env(start=None, override=False): 가까운 env 파일들을os.environ에 로드iter_env_paths(start=None): 실제 읽을 env 파일 목록을 우선순위 순서로 반환parse_env_file(path): env 파일을 dict로 파싱
예시:
새 코드에서 값 하나를 바로 읽을 때는 load_env()가 필요 없습니다.
from rb_utils.env import env_get
username = env_get("MONGO_USERNAME")
password = env_get("MONGO_PASSWORD")
host = env_get("MONGO_HOST", "127.0.0.1")
기존 코드가 os.getenv(...)를 많이 쓰고 있으면 프로세스 시작 시 한 번만 로드합니다.
import os
from rb_utils.env import load_env
load_env()
username = os.getenv("MONGO_USERNAME")
password = os.getenv("MONGO_PASSWORD")
기본 우선순위:
- 현재 위치에서 가장 가까운
.env.development - 같은 디렉토리의
.env - 부모 디렉토리의
.env.development - 부모 디렉토리의
.env
즉 개발 환경에서는 .env.development가 .env 값을 덮어씁니다.
운영 빌드에는 .env.development를 copy하지 않으므로 .env만 적용됩니다.
다른 suffix가 필요하면 명시적으로 지정:
env_get("KEY", env_files=(".env.local", ".env.development", ".env"))
8) rb_utils.pagination
class PageInfo(TypedDict): 페이지 응답 메타 타입 (mode/page/pages/limit/total/sort/order)class LogsResponse(TypedDict): 로그 목록 응답 타입 (items,pageInfo)
9) 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 문자열을 명시해야 합니다.
1 maintainer
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file rainbow_rb_utils-0.1.2.dev1.tar.gz.
File metadata
- Download URL: rainbow_rb_utils-0.1.2.dev1.tar.gz
- Upload date:
- Size: 27.5 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via:
twine/7.0.0 CPython/3.12.5
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
255ff883a29bd0d312b086036154bb8af5d9862bc364e3cf8c9846f20a9efb94
|
|
| MD5 |
fcb06e63955dca31113e64e020d29d3a
|
|
| BLAKE2b-256 |
236c2bd2c3571185f6697d30425bd046292f327886ee8b253102fd70714b7ad2
|
File details
Details for the file rainbow_rb_utils-0.1.2.dev1-py3-none-any.whl.
File metadata
- Download URL: rainbow_rb_utils-0.1.2.dev1-py3-none-any.whl
- Upload date:
- Size: 29.1 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via:
twine/7.0.0 CPython/3.12.5
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
4ca2b5c2005a52ecd530a25012761b69a87fb983546d0b6e6cd2c83f9771b499
|
|
| MD5 |
e6602af75b598104244211817212513b
|
|
| BLAKE2b-256 |
658c4b33dba08ca2c953502db3d1e45d77c2c84188a52be040b36dd838583bab
|
