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
rainbow_rb_utils-0.0.77.tar.gz
(27.1 kB
view details)
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.0.77.tar.gz.
File metadata
- Download URL: rainbow_rb_utils-0.0.77.tar.gz
- Upload date:
- Size: 27.1 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via:
twine/7.0.0 CPython/3.12.5
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
7cac9e9dcaa6b1d939be16a1b705afd680a7becfa1138427f891502492f8d588
|
|
| MD5 |
b7e83f18af8bf22374957f3fb2fb0303
|
|
| BLAKE2b-256 |
9551af65af2d263dab85ef822921c2150a39fec2ebe0e11465d02a0aff9735cb
|
File details
Details for the file rainbow_rb_utils-0.0.77-py3-none-any.whl.
File metadata
- Download URL: rainbow_rb_utils-0.0.77-py3-none-any.whl
- Upload date:
- Size: 28.7 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 |
b03c9514b1edefc3421c9a89630f5d393b18efb7e5044cb733ca3cf9205bee74
|
|
| MD5 |
a7d470f29892c153ca25f4a3d6676e97
|
|
| BLAKE2b-256 |
bdabbfe323449a36e9b77fe5659641dd41e6c87e0c569d758fde61ac8216725c
|
