Psion-fanatic/platypush
1
0
Fork 0
forked from platypush/platypush
Code Pull requests Activity

Added get_remaining_timeout utility function.

Browse source
This commit is contained in:
Fabio Manganiello 2023-08-14 23:05:32 +02:00
parent dc1a152433
commit 7157936b87
Signed by untrusted user: blacklight
GPG key ID: D90FBA7F76362774
1 changed files with 14 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
platypush/utils/__init__.py
View file

@ -15,7 +15,8 @@ import ssl
import urllib.request import urllib.request
from threading import Lock as TLock from threading import Lock as TLock
from tempfile import gettempdir from tempfile import gettempdir
from typing import Generator, Optional, Tuple, Union import time
from typing import Generator, Optional, Tuple, Type, Union
from dateutil import parser, tz from dateutil import parser, tz
from redis import Redis from redis import Redis
@ -633,4 +634,16 @@ def get_default_pid_file() -> str:
return os.path.join(gettempdir(), 'platypush.pid') return os.path.join(gettempdir(), 'platypush.pid')
def get_remaining_timeout(
timeout: Optional[float], start: float, cls: Union[Type[int], Type[float]] = float
) -> Optional[Union[int, float]]:
"""
Get the remaining timeout, given a start time.
"""
if timeout is None:
return None
return cls(max(0, timeout - (time.time() - start)))
# vim:sw=4:ts=4:et: # vim:sw=4:ts=4:et: