From within your app

By "internal" to a Reboot application, we specifically mean within one of the methods that you've implemented for your servicers. Each of those methods takes a context argument that you can use to make calls:

• ReaderContext - Passed to a reader.
• WriterContext - Passed to a writer.
• TransactionContext - Passed to a transaction.
• WorkflowContext - Passed to a workflow.

The types of these contexts allow Reboot (as well as static type checkers like mypy, Pyright, or tsc) to enforce its safety guarantees throughout the call graph.

Making calls to other durable data types

Within your servicer methods, you can call methods on other durable data types by first getting a reference to it using .ref(), then calling the method while passing along the context from your servicer method:

Python

async def transfer(
    self,
    context: TransactionContext,
    request: Bank.TransferRequest,
) -> None:
    from_account = Account.ref(request.from_account_id)
    to_account = Account.ref(request.to_account_id)

    await from_account.withdraw(context, amount=request.amount)
    await to_account.deposit(context, amount=request.amount)

TypeScript

async transfer(
  context: TransactionContext,
  request: Bank.TransferRequest
): Promise<Bank.PartialTransferResponse> {
  const fromAccount = Account.ref(request.fromAccountId);
  const toAccount = Account.ref(request.toAccountId);

  await fromAccount.withdraw(context, { amount: request.amount });
  await toAccount.deposit(context, { amount: request.amount });

Accessing local state

Within your servicer methods, you can access the state of the current instance via self.state (Python) or this.state (TypeScript class-based).

In TypeScript, if you're using the object literal syntax, state is passed as a parameter to your method.

workflow methods are an exception: they don't get direct access to state, but instead read snapshots of it and modify it explicitly via write() — see Workflows.

Python

async def send(
    self,
    context: WriterContext,
    request: SendRequest,
) -> SendResponse:
    message = request.message
    self.state.messages.extend([message])
    return SendResponse()

TypeScript Class

async send(
  context: WriterContext,
  request: ChatRoom.SendRequest
): Promise<ChatRoom.PartialSendResponse> {
  this.state.messages.push(request.message);
  return {};
}

TypeScript Object Literal

send: async (
  context: WriterContext,
  state: ChatRoom.State,
  request: ChatRoom.SendRequest
): Promise<[ChatRoom.State, ChatRoom.PartialSendResponse]> => {
  state.messages.push(request.message);
  return [state, {}];
}

important

Only Writer, Transaction, and Workflow methods can modify state. Reader methods can only read state.

Accessing your state ID

You can access the ID of the current state instance via context.state_id (Python) or context.stateId (TypeScript):

Python

async def get_id(
    self,
    context: ReaderContext,
    request: GetIdRequest,
) -> GetIdResponse:
    current_id = context.state_id
    return GetIdResponse(id=current_id)

TypeScript

async getId(
  context: ReaderContext,
  request: GetIdRequest
): Promise<GetIdResponse> {
  const currentId = context.stateId;
  return { id: currentId };
}

Making concurrent calls

You can make concurrent calls using asyncio.gather() (Python) or Promise.all() (TypeScript):

Python

balances = await asyncio.gather(
    *[
        Account.ref(account_id).balance(context)
        for account_id in account_ids
    ]
)

TypeScript

return {
  balances: await Promise.all(
    accountIds.map(async (accountId) => {
      const { amount } = await Account.ref(accountId).balance(context);
      return { accountId, balance: amount };
    })
  ),
};

tip

Using concurrent calls can significantly improve performance when gathering data from multiple state instances.

concurrently (Python)

For Python applications, Reboot provides a concurrently helper that improves on asyncio.gather in several ways:

• Adaptive concurrency limits — automatically discovers the optimal number of in-flight operations based on measured latency, rather than launching everything at once.
• Streaming results — use async for to process results as they complete, rather than waiting for all of them.
• Built-in logging — pass log=True to see what the adaptive limiter is doing.

Generator expressions and iterables

The simplest usage passes a generator expression of awaitables:

from reboot.aio.concurrently import concurrently

# Collect all results into a list (like `asyncio.gather`):
balances = await concurrently(
    Account.ref(account_id).balance(context)
    for account_id in account_ids
)

Use async for to process results as they complete:

async for balance in concurrently(
    Account.ref(account_id).balance(context)
    for account_id in account_ids
):
    process(balance)

You can also pass any iterable of awaitables, such as a list:

balances = await concurrently([
    Account.ref(account_id).balance(context)
    for account_id in account_ids
])

Async generators

When the work to produce each awaitable is more complex (e.g., multi-step logic, or the source of awaitables is itself async), pass an async generator directly:

async def fetch_balances():
    for customer in customers:
        account_id = get_account_id(customer)
        yield Account.ref(account_id).balance(context)

balances = await concurrently(fetch_balances())

Awaitables are consumed lazily — only as many as the adaptive concurrency limit allows at a time.

for_each — pairing elements with results

When you need to associate each input element with its result (e.g., building a dict), use the for_each parameter with a callable. Each iteration yields (element, result) tuples, which works well in dictionary comprehensions:

balances_by_account_id = {
    account_id: balance
    async for account_id, balance in concurrently(
        lambda account_id: Account.ref(account_id).balance(context),
        for_each=account_ids,
    )
}

With await, you get a list of the same (element, result) tuples:

balances = await concurrently(
    lambda account_id: Account.ref(account_id).balance(context),
    for_each=account_ids,
)

for account_id, balance in balances:
    process(account_id, balance)

If you just want the results without elements, use a generator expression instead of for_each:

balances = await concurrently(
    Account.ref(account_id).balance(context)
    for account_id in account_ids
)

The for_each parameter also accepts an async generator for lazy streaming sources like database cursors or paginated APIs:

async def account_ids():
    async for page in paginated_api():
        for account_id in page:
            yield account_id

balances_by_id = {
    account_id: balance
    async for account_id, balance in concurrently(
        lambda account_id: Account.ref(account_id).balance(context),
        for_each=account_ids(),
    )
}

Error handling

By default, exceptions are raised immediately (like asyncio.gather). Pass return_exceptions=True to return them as values instead — each item becomes result | BaseException, matching asyncio.gather(return_exceptions=True). This works with both await and async for, and enables tools like mypy to do isinstance narrowing for type-safe error handling:

# With `await`:
balances = await concurrently(
    (Account.ref(account_id).balance(context) for account_id in account_ids),
    return_exceptions=True,
)
for balance in balances:
    if isinstance(balance, BaseException):
        handle_error(balance)
    else:
        process(balance)  # mypy narrows the type correctly!

# With `async for` and `for_each`:
async for account_id, balance in concurrently(
    lambda account_id: Account.ref(account_id).balance(context),
    for_each=account_ids,
    return_exceptions=True,
):
    if isinstance(balance, BaseException):
        handle_error(account_id, balance)
    else:
        process(account_id, balance)  # mypy narrows the type correctly!

Context constraints

Each context type constrains what methods you can call, enforcing Reboot's safety guarantees:

Context Type	Can Call	Can Modify State
ReaderContext	Only reader methods	No
WriterContext	Only reader methods	Yes (own state only)
TransactionContext	reader, writer, and transaction methods	Yes
WorkflowContext	Any method type	Yes

tip

Type checkers like mypy, Pyright, and tsc will catch violations of these constraints at compile time, helping you catch bugs before runtime.

Calling writers from other writers

Notice that WriterContext cannot call other writer methods directly. This is by design to prevent unsafe, partial updates in the case that one of the calls fails.

If you need to call multiple writer methods atomically, use a transaction method instead.

Scheduling writers from writers

The one exception to the "writers can't call writers" rule is through scheduled tasks. A writer can schedule another writer to execute later:

Python

async def open(
    self,
    context: WriterContext,
) -> None:
    self.state.balance = 0.0
    await self.ref().schedule(
        when=timedelta(seconds=1),
    ).interest(context)

TypeScript

async open(
  context: WriterContext,
  request: Account.OpenRequest
): Promise<Account.PartialOpenResponse> {
  // Since this is a constructor, we are setting the initial state of the
  // state machine.
  this.state.customerName = request.customerName;

  // We'd like to send the new customer a welcome email, but that can be
  // done asynchronously, so we schedule it as a task.
  const taskId = await this.ref().schedule().welcomeEmail(context);

  return { welcomeEmailTaskId: taskId };
}

This works because .schedule() creates a separate task that will execute independently, not as part of the current writer's execution. See Tasks for more details.