Skip to content

Commit 19666d3

Browse files
Add a timeout in the gRPC function calls
Signed-off-by: camille-bouvy-frequenz <[email protected]>
1 parent ed6c425 commit 19666d3

File tree

1 file changed

+44
-11
lines changed

1 file changed

+44
-11
lines changed

src/frequenz/client/electricity_trading/_client.py

Lines changed: 44 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -5,10 +5,11 @@
55

66
from __future__ import annotations
77

8+
import asyncio
89
import logging
910
from datetime import datetime, timezone
1011
from decimal import Decimal, InvalidOperation
11-
from typing import TYPE_CHECKING, Any, Awaitable, cast
12+
from typing import TYPE_CHECKING, Any, Awaitable, Callable, cast
1213

1314
import grpc
1415

@@ -91,6 +92,31 @@ def validate_decimal_places(value: Decimal, decimal_places: int, name: str) -> N
9192
) from exc
9293

9394

95+
async def grpc_call_with_timeout(
96+
call: Callable[..., Awaitable[Any]], *args: Any, timeout: float = 300, **kwargs: Any
97+
) -> Any:
98+
"""
99+
Call a gRPC function with a timeout (in seconds).
100+
101+
Args:
102+
call: The gRPC method to be called.
103+
*args: Positional arguments for the gRPC call.
104+
timeout: Timeout duration in seconds. Defaults to 300.
105+
**kwargs: Keyword arguments for the gRPC call.
106+
107+
Returns:
108+
The result of the gRPC call.
109+
110+
Raises:
111+
asyncio.TimeoutError: If the call exceeds the timeout.
112+
"""
113+
try:
114+
return await asyncio.wait_for(call(*args, **kwargs), timeout=timeout)
115+
except asyncio.TimeoutError:
116+
_logger.error("Timeout while calling %s", call)
117+
raise
118+
119+
94120
class Client(BaseApiClient):
95121
"""Electricity trading client."""
96122

@@ -512,10 +538,10 @@ async def create_gridpool_order(
512538
try:
513539
response = await cast(
514540
Awaitable[electricity_trading_pb2.CreateGridpoolOrderResponse],
515-
self.stub.CreateGridpoolOrder(
541+
grpc_call_with_timeout(
542+
self.stub.CreateGridpoolOrder,
516543
electricity_trading_pb2.CreateGridpoolOrderRequest(
517-
gridpool_id=gridpool_id,
518-
order=order.to_pb(),
544+
gridpool_id=gridpool_id, order=order.to_pb()
519545
),
520546
metadata=self._metadata,
521547
),
@@ -624,7 +650,8 @@ async def update_gridpool_order(
624650
try:
625651
response = await cast(
626652
Awaitable[electricity_trading_pb2.UpdateGridpoolOrderResponse],
627-
self.stub.UpdateGridpoolOrder(
653+
grpc_call_with_timeout(
654+
self.stub.UpdateGridpoolOrder,
628655
electricity_trading_pb2.UpdateGridpoolOrderRequest(
629656
gridpool_id=gridpool_id,
630657
order_id=order_id,
@@ -659,7 +686,8 @@ async def cancel_gridpool_order(
659686
try:
660687
response = await cast(
661688
Awaitable[electricity_trading_pb2.CancelGridpoolOrderResponse],
662-
self.stub.CancelGridpoolOrder(
689+
grpc_call_with_timeout(
690+
self.stub.CancelGridpoolOrder,
663691
electricity_trading_pb2.CancelGridpoolOrderRequest(
664692
gridpool_id=gridpool_id, order_id=order_id
665693
),
@@ -687,7 +715,8 @@ async def cancel_all_gridpool_orders(self, gridpool_id: int) -> int:
687715
try:
688716
response = await cast(
689717
Awaitable[electricity_trading_pb2.CancelAllGridpoolOrdersResponse],
690-
self.stub.CancelAllGridpoolOrders(
718+
grpc_call_with_timeout(
719+
self.stub.CancelAllGridpoolOrders,
691720
electricity_trading_pb2.CancelAllGridpoolOrdersRequest(
692721
gridpool_id=gridpool_id
693722
),
@@ -719,7 +748,8 @@ async def get_gridpool_order(self, gridpool_id: int, order_id: int) -> OrderDeta
719748
try:
720749
response = await cast(
721750
Awaitable[electricity_trading_pb2.GetGridpoolOrderResponse],
722-
self.stub.GetGridpoolOrder(
751+
grpc_call_with_timeout(
752+
self.stub.GetGridpoolOrder,
723753
electricity_trading_pb2.GetGridpoolOrderRequest(
724754
gridpool_id=gridpool_id, order_id=order_id
725755
),
@@ -779,7 +809,8 @@ async def list_gridpool_orders(
779809
try:
780810
response = await cast(
781811
Awaitable[electricity_trading_pb2.ListGridpoolOrdersResponse],
782-
self.stub.ListGridpoolOrders(
812+
grpc_call_with_timeout(
813+
self.stub.ListGridpoolOrders,
783814
electricity_trading_pb2.ListGridpoolOrdersRequest(
784815
gridpool_id=gridpool_id,
785816
filter=gridpool_order_filer.to_pb(),
@@ -851,7 +882,8 @@ async def list_gridpool_trades(
851882
try:
852883
response = await cast(
853884
Awaitable[electricity_trading_pb2.ListGridpoolTradesResponse],
854-
self.stub.ListGridpoolTrades(
885+
grpc_call_with_timeout(
886+
self.stub.ListGridpoolTrades,
855887
electricity_trading_pb2.ListGridpoolTradesRequest(
856888
gridpool_id=gridpool_id,
857889
filter=gridpool_trade_filter.to_pb(),
@@ -908,7 +940,8 @@ async def list_public_trades(
908940
try:
909941
response = await cast(
910942
Awaitable[electricity_trading_pb2.ListPublicTradesResponse],
911-
self.stub.ListPublicTrades(
943+
grpc_call_with_timeout(
944+
self.stub.ListPublicTrades,
912945
electricity_trading_pb2.ListPublicTradesRequest(
913946
filter=public_trade_filter.to_pb(),
914947
pagination_params=pagination_params.to_proto(),

0 commit comments

Comments
 (0)