Handlers are where you implement the business logic for your APIs, events, and cron jobs.
Handler files must be named exactly as the API/Event/Cron name in the schema:
Health → src/handlers/api/Health.ts, Health.py, or health.rsUserCreated → Handler defined in event schemaDailyCleanup → src/handlers/cron/DailyCleanup.ts, DailyCleanup.py, or daily_cleanup.rsAPI handlers receive a request object and optionally a State object:
from generated.api.GetUser import GetUserRequest, GetUserResponse
from generated.state import State
async def handle_get_user(req: GetUserRequest, state: State) -> GetUserResponse:
# Access path parameters
user_id = req.id # From path: /users/:id
# Access query parameters
include_posts = req.query_params.get("include_posts", "false")
# Access request body (if present)
if hasattr(req, 'body'):
body_data = req.body
# Use state for logging and events
state.logger.info(f"Fetching user {user_id}")
return GetUserResponse(data=user)Event handlers receive an event object and optionally a State object:
from generated.events.UserCreated import UserCreated
from generated.state import State
async def handle_user_created(event: UserCreated, state: State) -> None:
# Access event payload
user_id = event.payload.id
email = event.payload.email
# Log and trigger additional events
state.logger.info(f"Processing user created: {user_id}")
state.trigger_event("WelcomeEmailSent", {"email": email})Cron handlers receive a request object (usually empty) and optionally a State object:
from generated.state import State
async def handle_daily_cleanup(req, state: State) -> None:
state.logger.info("Starting daily cleanup")
# Cleanup logic
state.trigger_event("CleanupCompleted", {"timestamp": datetime.now().isoformat()})The request object (*Request) contains:
Path Parameters: Extracted from route patterns (e.g., :id in /users/:id)
req.id, req.userId, etc.Query Parameters: URL query string parameters
req.query_params (Dict[str, str])req.queryParams (Record of string to string)req.query_params (HashMap<String, String>)Body: Request body (if specified in schema)
req.body (typed object)req.body (typed object)req.body (typed struct)Example with path and query parameters:
from generated.api.GetUser import GetUserRequest, GetUserResponse
async def handle_get_user(req: GetUserRequest, state: State) -> GetUserResponse:
# Path parameter from /users/:id
user_id = req.id # e.g., "123" from /users/123
# Query parameters from ?include_posts=true&format=json
include_posts = req.query_params.get("include_posts", "false")
format_type = req.query_params.get("format", "json")
# Use parameters
user = fetch_user(user_id, include_posts=include_posts == "true")
return GetUserResponse(data=user)from generated.state import State
async def handler(req, state: State):
# Log levels
state.logger.info("Information message")
state.logger.error("Error message")
state.logger.warning("Warning message")
state.logger.warn("Warning message (alias)")
state.logger.debug("Debug message")
state.logger.trace("Trace message")
# With additional fields
state.logger.info("User action", user_id=123, action="login")from generated.state import State
async def handler(req, state: State):
# Trigger an event manually
state.trigger_event("CustomEvent", {
"data": "value",
"timestamp": datetime.now().isoformat()
})from generated.api.CreateUser import CreateUserRequest, CreateUserResponse
from generated.state import State
async def handle_create_user(req: CreateUserRequest, state: State) -> CreateUserResponse:
user = create_user(req.body)
# Set payload for auto-triggered events (defined in schema)
state.set_payload("UserCreated", {
"id": user.id,
"name": user.name,
"email": user.email
})
# Can also manually trigger other events
state.trigger_event("AnalyticsEvent", {"action": "user_created"})
return CreateUserResponse(data=user)from generated.api.GetUserPosts import GetUserPostsRequest, GetUserPostsResponse
from generated.state import State
async def handle_get_user_posts(
req: GetUserPostsRequest,
state: State
) -> GetUserPostsResponse:
# Path parameter: /users/:userId/posts
user_id = req.userId
# Query parameters: ?limit=10&offset=0&sort=created_at
limit = int(req.query_params.get("limit", "10"))
offset = int(req.query_params.get("offset", "0"))
sort_by = req.query_params.get("sort", "created_at")
# Logging
state.logger.info(f"Fetching posts for user {user_id}", {
"limit": limit,
"offset": offset
})
# Business logic
posts = fetch_user_posts(user_id, limit=limit, offset=offset, sort_by=sort_by)
# Trigger analytics event
state.trigger_event("PostsViewed", {
"user_id": user_id,
"count": len(posts)
})
return GetUserPostsResponse(data=posts)use crate::generated::api::GetUserPosts::{GetUserPostsRequest, GetUserPostsResponse};
use crate::generated::state::State;
use rohas_runtime::Result;
pub async fn handle_get_user_posts(
req: GetUserPostsRequest,
state: &mut State,
) -> Result<GetUserPostsResponse> {
// Path parameter: /users/:userId/posts
let user_id = req.user_id.clone();
// Query parameters: ?limit=10&offset=0&sort=created_at
let limit = req.query_params
.get("limit")
.and_then(|s| s.parse::<i32>().ok())
.unwrap_or(10);
let offset = req.query_params
.get("offset")
.and_then(|s| s.parse::<i32>().ok())
.unwrap_or(0);
let sort_by = req.query_params
.get("sort")
.map(|s| s.as_str())
.unwrap_or("created_at");
// Logging
state.logger().info(&format!(
"Fetching posts for user {}: limit={}, offset={}",
user_id, limit, offset
));
// Business logic
let posts = fetch_user_posts(&user_id, limit, offset, sort_by)?;
// Trigger analytics event
state.trigger_event("PostsViewed", serde_json::json!({
"user_id": user_id,
"count": posts.len()
}))?;
Ok(GetUserPostsResponse { data: posts })
}All types are auto-generated in src/generated/ when you run rohas codegen. Do not edit these files manually.
The generated types include: