nebius.aio.service_descriptor.ExtractorChannel

class documentation

class ExtractorChannel(GRPCChannel): (source)

A mock gRPC channel for extracting service names from stub classes.

This channel implementation records the last method called on it and can extract the service name from the method name. It is used to introspect gRPC stub classes without making actual network calls.

Async Method	`__aenter__`	Enter async context.
Async Method	`__aexit__`	Exit async context.
Method	`__init__`	Undocumented
Async Method	`channel_ready`	Wait for channel to be ready (no-op for this mock).
Async Method	`close`	Close the channel (no-op for this mock implementation).
Method	`get_service_name`	Extract the service name from the last recorded method.
Method	`get_state`	Get the channel state (always READY for this mock).
Method	`stream_stream`	Record a stream-stream method call and return a stub.
Method	`stream_unary`	Record a stream-unary method call and return a stub.
Method	`unary_stream`	Record a unary-stream method call and return a stub.
Method	`unary_unary`	Record a unary-unary method call and return a stub.
Async Method	`wait_for_state_change`	Wait for state change (not implemented for this mock).
Instance Variable	`_last_method`	Undocumented

async def __aenter__(self) -> ExtractorChannel: (source) ¶

Enter async context.

Returns
`ExtractorChannel`	Self.

async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any): (source) ¶

Exit async context.

Parameters
exc_type:Any	Exception type.
exc_val:Any	Exception value.
exc_tb:Any	Exception traceback.

def __init__(self): (source) ¶

Undocumented

async def channel_ready(self): (source) ¶

Wait for channel to be ready (no-op for this mock).

async def close(self, grace: float | None = None): (source) ¶

Close the channel (no-op for this mock implementation).

Parameters
grace:float or None	Optional grace period.

def get_service_name(self) -> str: (source) ¶

Extract the service name from the last recorded method.

Returns
str	The service name.
Raises
`NoMethodsInServiceError`	If no methods have been recorded.

def get_state(self, try_to_connect: bool = False) -> ChannelConnectivity: (source) ¶

Get the channel state (always READY for this mock).

Parameters
try_to_connect:bool	Whether to attempt connection.
Returns
`ChannelConnectivity`	The connectivity state.

def stream_stream(self, method: str, request_serializer: SerializingFunction | None = None, response_deserializer: DeserializingFunction | None = None, _registered_method: bool | None = None) -> StreamStreamMultiCallable: (source) ¶

Record a stream-stream method call and return a stub.

Parameters
method:str	The method name.
request_serializer:`SerializingFunction` or None	Optional request serializer.
response_deserializer:`DeserializingFunction` or None	Optional response deserializer.
_registered_method:bool or None	Whether the method is registered.
Returns
`StreamStreamMultiCallable`	A stub callable.

def stream_unary(self, method: str, request_serializer: SerializingFunction | None = None, response_deserializer: DeserializingFunction | None = None, _registered_method: bool | None = None) -> StreamUnaryMultiCallable: (source) ¶

Record a stream-unary method call and return a stub.

Parameters
method:str	The method name.
request_serializer:`SerializingFunction` or None	Optional request serializer.
response_deserializer:`DeserializingFunction` or None	Optional response deserializer.
_registered_method:bool or None	Whether the method is registered.
Returns
`StreamUnaryMultiCallable`	A stub callable.

def unary_stream(self, method: str, request_serializer: SerializingFunction | None = None, response_deserializer: DeserializingFunction | None = None, _registered_method: bool | None = None) -> UnaryStreamMultiCallable[Req, Res]: (source) ¶

Record a unary-stream method call and return a stub.

Parameters
method:str	The method name.
request_serializer:`SerializingFunction` or None	Optional request serializer.
response_deserializer:`DeserializingFunction` or None	Optional response deserializer.
_registered_method:bool or None	Whether the method is registered.
Returns
`UnaryStreamMultiCallable`	A stub callable.

def unary_unary(self, method: str, request_serializer: SerializingFunction | None = None, response_deserializer: DeserializingFunction | None = None, _registered_method: bool | None = False) -> UnaryUnaryMultiCallable[Req, Res]: (source) ¶

Record a unary-unary method call and return a stub.

Parameters
method:str	The method name.
request_serializer:`SerializingFunction` or None	Optional request serializer.
response_deserializer:`DeserializingFunction` or None	Optional response deserializer.
_registered_method:bool or None	Whether the method is registered.
Returns
`UnaryUnaryMultiCallable`	A stub callable.

async def wait_for_state_change(self, last_observed_state: ChannelConnectivity): (source) ¶

Wait for state change (not implemented for this mock).

Parameters
last_observed_state:`ChannelConnectivity`	The last observed state.
Raises
`NotImplementedError`	Always raised.

_last_method = (source) ¶

Undocumented