rb_utils

rb_utils is a collection of utilities reused across the backend.

Modules / Functions

1) rb_utils.parser

• snake_to_camel(s: str) -> str: Converts a snake_case string to CamelCase
• camel_to_snake(s: str) -> str: Converts a CamelCase string to snake_case
• t_to_dict(obj: Any) -> Any: Recursively converts nested objects (FlatBuffer T, dict/list/tuple/set, bytes, numpy arrays) into Python built-in types
• _normalize_filter(filter_: Any) -> dict: Normalizes None/dict/JSON string into a common dict
• to_json(obj: Any) -> str: Serializes the result of t_to_dict to a JSON string
• to_iso(val: str | datetime) -> str: Returns an ISO string for a datetime, or the string unchanged

Example:

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: Safely evaluates an expression under restricted AST rules
• _pick_target_loop(func, provided) -> asyncio.AbstractEventLoop | None: Selects the loop on which to run the callable
• call_with_matching_args(func, **provided) -> Any: Calls a function with only the arguments matching its signature (works for both sync and async)
• eval_value(value, variables=None, get_global_variable=None) -> Any: Recursively evaluates strings/lists/tuples/dicts
• make_builtins_allow_most() -> dict: Builds a builtins dict excluding dangerous builtins

Example:

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): Creates a task on the current loop and logs its completion/exceptions
• _ensure_private_loop() -> asyncio.AbstractEventLoop: Ensures a dedicated background loop exists
• fire_and_log(coro, *, name=None, loop=None): Safely runs on the current loop, an external loop, or the dedicated loop
• _attach_done_logging(fut_or_task, *, name=None): Attaches a callback that logs exceptions when a Future/Task completes

Example:

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):
  • Fields: message: str, status_code: int | None
  • __str__(): Outputs in the form message (status=...)
  • to_dict(): Returns {message, status_code}

Example:

from rb_utils.service_exception import ServiceException

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

5) rb_utils.date

• is_valid_date(d): Validates a date/str/int(timestamp)
• timestamp_ms_to_seconds(timestamp, tz=UTC): Normalizes a ms/seconds timestamp to seconds
• is_has_time(s: str): Determines whether a string contains time information
• parse_date_with_time(mode, d, tz=UTC): Converts a date/string/timestamp to a timezone-aware datetime
• ensure_datetime(dt, tz): Coerces a date | datetime into a datetime in the given tz
• _tz(tz_name): Looks up a timezone from the cache
• convert_dt(local, in_tz, out_tz): Converts a datetime/date from the input tz to the output tz
• get_current_dt_yyyymmddhhmmss(tz): Returns the current time as a YYYYMMDDHHMMSS string

Example:

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: Builds a Content-Disposition header for downloads
• sanitize_filename(name, default="export.csv", max_name_len=100) -> str: Removes OS-forbidden characters and limits length
• get_env_path() -> Path | None: Searches for a .env file relative to the working location

Example:

from rb_utils.file import sanitize_filename

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

7) rb_utils.env

• env_get(key, default=None, start=None): Looks up a value from os.environ first, then the nearest .env.development → .env, in that order
• load_env(start=None, override=False): Loads the nearest env files into os.environ
• iter_env_paths(start=None): Returns the env files that will actually be read, in priority order
• parse_env_file(path): Parses an env file into a dict

Example:

In new code, you don't need load_env() to read a single value directly.

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")

If existing code uses os.getenv(...) heavily, load once at process startup.

import os

from rb_utils.env import load_env

load_env()

username = os.getenv("MONGO_USERNAME")
password = os.getenv("MONGO_PASSWORD")

Default priority:

1. The .env.development closest to the current location
2. .env in the same directory
3. .env.development in the parent directory
4. .env in the parent directory

In other words, in a development environment .env.development overrides values in .env. Production builds do not copy .env.development, so only .env applies.

If you need a different suffix, specify it explicitly:

env_get("KEY", env_files=(".env.local", ".env.development", ".env"))

8) rb_utils.pagination

• class PageInfo(TypedDict): Page response metadata type (mode/page/pages/limit/total/sort/order)
• class LogsResponse(TypedDict): Log list response type (items, pageInfo)

9) rb_utils.helper

• ManipulateZenohResHelper(obj): Function intended as a utility for fixing up Zenoh response objects

Caution:

• The current implementation uses .hasAttr/.set access instead of dict handling and raises ValueError immediately, so verify its behavior before use.

Common Usage Patterns

• Flattening FlatBuffer/SDK responses: t_to_dict
• Evaluating dynamic Flow step arguments: eval_value, safe_eval_expr
• Async fire-and-forget + exception logging: fire_and_log
• Common service-layer exception: ServiceException
• File downloads / filename sanitization: content_disposition_header, sanitize_filename

General Notes

• The flow_manager evaluator only executes allowed operations/functions.
• Timezone conversion functions require an explicit IANA tz string such as Asia/Seoul.