rb_schemas User Guide

rb_schemas is a package that provides the project's shared Pydantic schemas and Flow Manager integration types.

1. Contents

• Base schemas (base.py)
  • Response_ReturnValuePD
  • Shared models such as NJointfPD, NInputfPD, Vec3fPD
• Flow Manager integration types (sdk.py)
  • FlowManagerArgs
• Schema utilities (utility.py)
  • Pick(base, *include)
  • Omit(base, *omit)
• Models auto-generated from FBS
  • rb_schemas.fbs_models.*
  • Class names follow the ...PD suffix convention

2. Purpose of FlowManagerArgs

FlowManagerArgs is used when an SDK function receives the Flow Manager execution context.

from rb_schemas.sdk import FlowManagerArgs

def my_step_handler(flow_manager_args: FlowManagerArgs | None = None):
    # perform step logic
    if flow_manager_args is not None:
        flow_manager_args.done()

3. Pick/Omit Example

from pydantic import BaseModel
from rb_schemas.utility import Omit, Pick

class UserPD(BaseModel):
    user_id: str
    email: str
    password: str

UserPublicPD = Omit(UserPD, "password")
UserOnlyIdPD = Pick(UserPD, "user_id")

4. FBS Model Example

# Example: import an auto-generated model
from rb_schemas.fbs_models.nexus.v1.deploy_models import Response_Deploy_ProgressPD

payload = Response_Deploy_ProgressPD(
    sw_name="example",
    ip="127.0.0.1",
    mode="dev",
    tag="v1",
    percentage=80,
    service_name="manipulate",
    result="success",
)

5. Notes

• FBS-based models are automatically regenerated when backend.flatc is run.
• Generated models may have both snake_case fields and aliases (the original CamelCase names).