nebius.aio.stream.StreamRequest

class documentation

class StreamRequest(Generic[Req, Res]): (source)

Constructor: StreamRequest(channel, route, request, result_class, ...)

Lazy native async call for one of the three streaming RPC shapes.

Use server streams as async context managers when iteration may stop early:

async with client.watch(request) as stream:
    async for item in stream:
        if done(item):
            break

The context exit calls aclose. This call cancels the native stream and releases its address channel.

A caller-supplied asynchronous request iterator is consumed on the SDK loop. It must not contain state bound to a different event loop. The SDK cannot detect hidden loop ownership in an arbitrary iterator.

A unary request message and authentication options are copied when this wrapper is created. Each explicit write copies a supported message before dispatch to the SDK loop. Unknown custom values keep their previous pass-through behavior and must be safe to share between threads.

Timeout budgets start with the first stream operation. They include time waiting for SDK-loop dispatch, authentication, and the native RPC rather than starting only after a queued operation reaches the internal loop. The authorization budget applies only when a provider is active and authorization is not explicitly disabled.

Raises
`ValueError`	If `timeout` or `auth_timeout` is NaN or infinite. Use `None` for an unlimited timeout.

Async Method	`__aenter__`	Undocumented
Async Method	`__aexit__`	Undocumented
Method	`__aiter__`	Undocumented
Method	`__await__`	Undocumented
Method	`__init__`	Undocumented
Async Method	`aclose`	Cancel the native call and discard its address channel.
Method	`cancel`	Request cancellation without accessing loop-owned state off-loop.
Async Method	`done_writing`	Undocumented
Async Method	`write`	Snapshot and write one request on the SDK event loop.
Static Method	`_serialize`	Undocumented
Method	`_abort`	Undocumented
Async Method	`_aclose`	Abort the native call and release its leased channel.
Async Method	`_authenticate`	Undocumented
Method	`_cancel_submission_finished`	Observe asynchronous cancellation cleanup and restore retryability.
Method	`_check_process`	Reject a stream inherited across `fork` before taking its locks.
Async Method	`_done_writing`	Finish explicit request writes on the call owner loop.
Method	`_is_cancelled`	Return the cancellation state under the state lock.
Method	`_is_released`	Return the channel-release state under the state lock.
Method	`_mark_native_terminal`	Publish native completion before the stream wrapper resumes.
Async Method	`_next_response`	Read one response while serializing access to the iterator.
Async Method	`_on_sdk_loop`	Run an awaitable on the SDK loop when the channel supports dispatch.
Method	`_pause_request_deadline_for_authorization`	Pause the request-only stream clock during authentication.
Method	`_release`	Undocumented
Method	`_release_soon`	Release transport state without blocking an active caller loop.
Method	`_remaining_deadline`	Return the applicable stream deadline remaining in seconds.
Method	`_remaining_dispatch_deadline`	Return the request/authentication budget before SDK-loop dispatch.
Async Method	`_responses`	Undocumented
Async Method	`_responses_internal`	Yield streaming responses directly on the call owner loop.
Async Method	`_result`	Undocumented
Async Method	`_result_internal`	Read a unary response and release the leased channel.
Method	`_resume_request_deadline_after_authorization`	Resume request timeout and publish completed authentication.
Async Method	`_start`	Undocumented
Async Method	`_write`	Write one request on the call owner loop.
Instance Variable	`_address_channel`	Undocumented
Instance Variable	`_auth_options`	Undocumented
Instance Variable	`_auth_timeout`	Undocumented
Instance Variable	`_authentication_finished`	Undocumented
Instance Variable	`_authorization_deadline`	Undocumented
Instance Variable	`_authorization_deadline_enabled`	Undocumented
Instance Variable	`_call`	Undocumented
Instance Variable	`_cancel_event`	Undocumented
Instance Variable	`_cancel_requested`	Undocumented
Instance Variable	`_cancelled`	Undocumented
Instance Variable	`_channel`	Undocumented
Instance Variable	`_client_streaming`	Undocumented
Instance Variable	`_compression`	Undocumented
Instance Variable	`_credentials`	Undocumented
Instance Variable	`_deadlines_started`	Undocumented
Instance Variable	`_metadata`	Undocumented
Instance Variable	`_native_terminal`	Undocumented
Instance Variable	`_owner_loop`	Undocumented
Instance Variable	`_process_id`	Undocumented
Instance Variable	`_read_lock`	Undocumented
Instance Variable	`_released`	Undocumented
Instance Variable	`_request`	Undocumented
Instance Variable	`_request_deadline`	Undocumented
Instance Variable	`_request_deadline_paused`	Undocumented
Instance Variable	`_request_timeout_remaining`	Undocumented
Instance Variable	`_response_iterator`	Undocumented
Instance Variable	`_result_class`	Undocumented
Instance Variable	`_route`	Undocumented
Instance Variable	`_server_streaming`	Undocumented
Instance Variable	`_start_entered`	Undocumented
Instance Variable	`_start_error`	Undocumented
Instance Variable	`_start_lock`	Undocumented
Instance Variable	`_state_lock`	Undocumented
Instance Variable	`_timeout`	Undocumented
Instance Variable	`_wait_for_ready`	Undocumented
Instance Variable	`_write_lock`	Undocumented

async def __aenter__(self) -> StreamRequest[Req, Res]: (source) ¶

Undocumented

async def __aexit__(self, exc_type: type[BaseException] | None, exc: BaseException | None, traceback: object | None): (source) ¶

Undocumented

def __aiter__(self) -> AsyncIterator[Res]: (source) ¶

Undocumented

def __await__(self) -> Generator[Any, None, Res]: (source) ¶

Undocumented

def __init__(self, *, channel: Any, route: Route, request: Any, result_class: type[Any], client_streaming: bool, server_streaming: bool, metadata: Metadata | list[tuple[str, str]] | None = None, timeout: float | None = None, auth_timeout: float | None = None, auth_options: dict[str, str] | None = None, credentials: CallCredentials | None = None, compression: Compression | None = None, wait_for_ready: bool | None = True, grpc_channel_override: AddressChannel | None = None, **unsupported: object): (source) ¶

Undocumented

async def aclose(self): (source) ¶

Cancel the native call and discard its address channel.

def cancel(self) -> bool: (source) ¶

Request cancellation without accessing loop-owned state off-loop.

SDK channels dispatch cancellation to the SDK loop. A compatibility channel without immediate scheduling support—including an adapter whose run_async returns the original one-shot awaitable—uses the loop that started the stream instead. A foreign caller receives False if that owner loop is not running or closes during dispatch; it may retry after restoring the loop. As with every accepted event-loop callback, the owner must remain running long enough to execute it. A later asynchronous cleanup failure is logged and makes cancellation retryable when the transport was not released.

Returns
`bool`	`True` if native cancellation was applied or accepted for dispatch; otherwise `False`. A terminal server stream returns `False` even when its remaining lease cleanup was scheduled.

async def done_writing(self): (source) ¶

Undocumented

async def write(self, request: Req): (source) ¶

Snapshot and write one request on the SDK event loop.

Parameters
request:`Req`	Request message to write. Supported mutable protobuf messages are copied before dispatch. Unknown custom values retain their historical pass-through behavior and must be thread-safe.

@staticmethod
def _serialize(message: object) -> bytes: (source) ¶

Undocumented

def _abort(self): (source) ¶

Undocumented

async def _aclose(self): (source) ¶

Abort the native call and release its leased channel.

async def _authenticate(self): (source) ¶

Undocumented

def _cancel_submission_finished(self, completed: Any): (source) ¶

Observe asynchronous cancellation cleanup and restore retryability.

Parameters
completed:`Any`	Future-like cancellation submission.

def _check_process(self): (source) ¶

Reject a stream inherited across fork before taking its locks.

async def _done_writing(self): (source) ¶

Finish explicit request writes on the call owner loop.

Raises
`TypeError`	If the RPC does not accept explicit writes.

def _is_cancelled(self) -> bool: (source) ¶

Return the cancellation state under the state lock.

def _is_released(self) -> bool: (source) ¶

Return the channel-release state under the state lock.

def _mark_native_terminal(self, _: object): (source) ¶

Publish native completion before the stream wrapper resumes.

async def _next_response(self) -> Res: (source) ¶

Read one response while serializing access to the iterator.

Returns
`Res`	Next streaming response.
Raises
`StopAsyncIteration`	If the response stream is complete.

async def _on_sdk_loop(self, awaitable: Awaitable[T], *, enforce_deadline: bool = True, terminate_on_rejection: bool = False) -> T: (source) ¶

Run an awaitable on the SDK loop when the channel supports dispatch.

Parameters
awaitable:`Awaitable[T]`	Stream work to run.
enforce_deadline:`bool`	Apply the stream's caller-side deadline. Close cleanup disables this limit so an expired stream can release its transport.
terminate_on_rejection:`bool`	Mark an active stream terminal and discard its lease when close cleanup cannot start. Ordinary operation rejection leaves an existing stream active.
Returns
`T`	Result of the stream work.

def _pause_request_deadline_for_authorization(self): (source) ¶

Pause the request-only stream clock during authentication.

Queueing before stream startup is already charged. The retained request budget resumes only after authentication succeeds, while the authorization deadline continues to bound authentication plus the native stream lifetime.

def _release(self, *, discard: bool = False): (source) ¶

Undocumented

def _release_soon(self, *, discard: bool = False): (source) ¶

Release transport state without blocking an active caller loop.

def _remaining_deadline(self, *, initialize: bool = False) -> float | None: (source) ¶

Return the applicable stream deadline remaining in seconds.

The first caller-side operation fixes monotonic request and applicable authorization deadlines under the state lock. Later operations reuse those deadlines. Concurrent reads and writes cannot each obtain a new budget. SDK-loop queueing uses the same native RPC lifetime. Before authentication completes, only its overall authorization deadline is exposed to caller-side waiting; request timeout is paused during authentication and resumes for native stream work.

Parameters
initialize:`bool`	Start the deadlines when no prior stream operation has done so. Cleanup calls leave this false because they must be able to release transport state after a deadline expires.
Returns
`float \| None`	The smaller remaining request/authorization budget, or `None` when both configured limits are infinite.