aljaz/bitfinex-api-py
1
0
Fork 0
mirror of https://github.com/aljazceru/bitfinex-api-py.git synced 2025-12-19 14:54:21 +01:00
Code Issues Packages Projects Releases Wiki Activity

Apply pylint's linting rules to bfxapi/websocket/client/*.py.

Browse Source
This commit is contained in:
Davide Casale
2023-03-06 18:46:04 +01:00
parent 7e627dd239
commit 5c707d7929
10 changed files with 173 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files

133
bfxapi/websocket/client/bfx_websocket_client.py
View File

@@ -1,18 +1,19 @@
import traceback, json, asyncio, hmac, hashlib, time, websockets, socket, random
from typing import cast
from collections import namedtuple
from datetime import datetime
import traceback, json, asyncio, hmac, hashlib, time, socket, random, websockets
from pyee.asyncio import AsyncIOEventEmitter
from .bfx_websocket_bucket import _HEARTBEAT, F, _require_websocket_connection, BfxWebsocketBucket
from .bfx_websocket_inputs import BfxWebsocketInputs
from ..handlers import PublicChannelsHandler, AuthenticatedChannelsHandler
from ..exceptions import WebsocketAuthenticationRequired, InvalidAuthenticationCredentials, EventNotSupported, OutdatedClientVersion
from ..exceptions import WebsocketAuthenticationRequired, InvalidAuthenticationCredentials, EventNotSupported, \
OutdatedClientVersion
from ...utils.json_encoder import JSONEncoder
@@ -20,14 +21,15 @@ from ...utils.logger import ColorLogger, FileLogger
def _require_websocket_authentication(function: F) -> F:
async def wrapper(self, *args, **kwargs):
if hasattr(self, "authentication") and self.authentication == False:
raise WebsocketAuthenticationRequired("To perform this action you need to authenticate using your API_KEY and API_SECRET.")
if hasattr(self, "authentication") and not self.authentication:
raise WebsocketAuthenticationRequired("To perform this action you need to authenticate " +
"using your API_KEY and API_SECRET.")
await _require_websocket_connection(function)(self, *args, **kwargs)
return cast(F, wrapper)
class BfxWebsocketClient(object):
class BfxWebsocketClient:
VERSION = BfxWebsocketBucket.VERSION
MAXIMUM_CONNECTIONS_AMOUNT = 20
@@ -43,16 +45,18 @@ class BfxWebsocketClient(object):
self.host, self.credentials, self.event_emitter = host, credentials, AsyncIOEventEmitter()
self.on_open_events, self.buckets, self.authentication = [], [], False
self.inputs = BfxWebsocketInputs(handle_websocket_input=self.__handle_websocket_input)
self.handler = AuthenticatedChannelsHandler(event_emitter=self.event_emitter)
if log_filename == None:
if log_filename is None:
self.logger = ColorLogger("BfxWebsocketClient", level=log_level)
else: self.logger = FileLogger("BfxWebsocketClient", level=log_level, filename=log_filename)
self.event_emitter.add_listener("error",
lambda exception: self.logger.error(f"{type(exception).__name__}: {str(exception)}" + "\n" +
self.event_emitter.add_listener("error",
lambda exception: self.logger.error(f"{type(exception).__name__}: {str(exception)}" + "\n" +
str().join(traceback.format_exception(type(exception), exception, exception.__traceback__))[:-1])
)
@@ -61,23 +65,24 @@ class BfxWebsocketClient(object):
async def start(self, connections = 5):
if connections > BfxWebsocketClient.MAXIMUM_CONNECTIONS_AMOUNT:
self.logger.warning(f"It is not safe to use more than {BfxWebsocketClient.MAXIMUM_CONNECTIONS_AMOUNT} buckets from the same " +
f"connection ({connections} in use), the server could momentarily block the client with <429 Too Many Requests>.")
self.logger.warning(f"It is not safe to use more than {BfxWebsocketClient.MAXIMUM_CONNECTIONS_AMOUNT} "
+ f"buckets from the same connection ({connections} in use), the server could momentarily block the "
+ "client with <429 Too Many Requests>.")
self.on_open_events = [ asyncio.Event() for _ in range(connections) ]
for _ in range(connections):
self.on_open_events.append(asyncio.Event())
self.buckets = [
BfxWebsocketBucket(self.host, self.event_emitter, self.on_open_events[index])
for index in range(connections)
]
for index in range(connections):
self.buckets += [BfxWebsocketBucket(self.host, self.event_emitter, self.on_open_events[index])]
tasks = [ bucket._connect(index) for index, bucket in enumerate(self.buckets) ]
tasks.append(self.__connect(self.credentials))
tasks = [ bucket.connect(index) for index, bucket in enumerate(self.buckets) ]
tasks.append(self.__connect())
await asyncio.gather(*tasks)
async def __connect(self, credentials = None):
#pylint: disable-next=too-many-statements
async def __connect(self):
Reconnection = namedtuple("Reconnection", ["status", "attempts", "timestamp"])
reconnection, delay = Reconnection(status=False, attempts=0, timestamp=None), None
@@ -86,10 +91,10 @@ class BfxWebsocketClient(object):
nonlocal reconnection
async with websockets.connect(self.host) as websocket:
if reconnection.status == True:
self.logger.info(f"Reconnect attempt successful (attempt no.{reconnection.attempts}): The " +
f"client has been offline for a total of {datetime.now() - reconnection.timestamp} " +
f"(connection lost at: {reconnection.timestamp:%d-%m-%Y at %H:%M:%S}).")
if reconnection.status:
self.logger.info(f"Reconnect attempt successful (attempt no.{reconnection.attempts}): The "
+ f"client has been offline for a total of {datetime.now() - reconnection.timestamp} "
+ f"(connection lost at: {reconnection.timestamp:%d-%m-%Y at %H:%M:%S}).")
reconnection = Reconnection(status=False, attempts=0, timestamp=None)
@@ -106,20 +111,22 @@ class BfxWebsocketClient(object):
if isinstance(message, dict) and message["event"] == "info" and "version" in message:
if BfxWebsocketClient.VERSION != message["version"]:
raise OutdatedClientVersion(f"Mismatch between the client version and the server version. " +
f"Update the library to the latest version to continue (client version: {BfxWebsocketClient.VERSION}, " +
f"server version: {message['version']}).")
raise OutdatedClientVersion("Mismatch between the client version and the server version. "
+ "Update the library to the latest version to continue (client version: "
+ f"{BfxWebsocketClient.VERSION}, server version: {message['version']}).")
elif isinstance(message, dict) and message["event"] == "info" and message["code"] == 20051:
rcvd = websockets.frames.Close(code=1012, reason="Stop/Restart Websocket Server (please reconnect).")
raise websockets.ConnectionClosedError(rcvd=rcvd, sent=None)
elif isinstance(message, dict) and message["event"] == "auth":
if message["status"] == "OK":
self.event_emitter.emit("authenticated", message); self.authentication = True
self.event_emitter.emit("authenticated", message)
self.authentication = True
else: raise InvalidAuthenticationCredentials("Cannot authenticate with given API-KEY and API-SECRET.")
elif isinstance(message, dict) and message["event"] == "error":
self.event_emitter.emit("wss-error", message["code"], message["msg"])
elif isinstance(message, list) and (chanId := message[0]) == 0 and message[1] != _HEARTBEAT:
elif isinstance(message, list) and message[0] == 0 and message[1] != _HEARTBEAT:
self.handler.handle(message[1], message[2])
class _Delay:
@@ -138,52 +145,52 @@ class BfxWebsocketClient(object):
self.__backoff_delay = min(__backoff_delay, _Delay.BACKOFF_MAX)
return backoff_delay
def peek(self):
return (self.__backoff_delay == _Delay.BACKOFF_MIN) \
and self.__initial_delay or self.__backoff_delay
while True:
if reconnection.status == True:
if reconnection.status:
await asyncio.sleep(delay.next())
try:
await _connection()
except (websockets.ConnectionClosedError, socket.gaierror) as error:
if isinstance(error, websockets.ConnectionClosedError) and (error.code == 1006 or error.code == 1012):
if isinstance(error, websockets.ConnectionClosedError) and error.code in (1006, 1012):
if error.code == 1006:
self.logger.error("Connection lost: no close frame received "
self.logger.error("Connection lost: no close frame received "
+ "or sent (1006). Attempting to reconnect...")
if error.code == 1012:
self.logger.info("WSS server is about to restart, reconnection "
+ "required (client received 20051). Attempt in progress...")
reconnection = Reconnection(status=True, attempts=1, timestamp=datetime.now());
reconnection = Reconnection(status=True, attempts=1, timestamp=datetime.now())
delay = _Delay(backoff_factor=1.618)
elif isinstance(error, socket.gaierror) and reconnection.status == True:
elif isinstance(error, socket.gaierror) and reconnection.status:
self.logger.warning(f"Reconnection attempt no.{reconnection.attempts} has failed. "
+ f"Next reconnection attempt in ~{round(delay.peek()):.1f} seconds."
+ f"Next reconnection attempt in ~{round(delay.peek()):.1f} seconds."
+ f"(at the moment the client has been offline for {datetime.now() - reconnection.timestamp})")
reconnection = reconnection._replace(attempts=reconnection.attempts + 1)
else: raise error
if reconnection.status == False:
if not reconnection.status:
break
async def __authenticate(self, API_KEY, API_SECRET, filters=None):
data = { "event": "auth", "filter": filters, "apiKey": API_KEY }
async def __authenticate(self, api_key, api_secret, filters=None):
data = { "event": "auth", "filter": filters, "apiKey": api_key }
data["authNonce"] = str(round(time.time() * 1_000_000))
data["authPayload"] = "AUTH" + data["authNonce"]
data["authSig"] = hmac.new(
API_SECRET.encode("utf8"),
api_secret.encode("utf8"),
data["authPayload"].encode("utf8"),
hashlib.sha384
hashlib.sha384
).hexdigest()
await self.websocket.send(json.dumps(data))
@@ -193,56 +200,58 @@ class BfxWebsocketClient(object):
index = counters.index(min(counters))
await self.buckets[index]._subscribe(channel, **kwargs)
await self.buckets[index].subscribe(channel, **kwargs)
async def unsubscribe(self, subId):
async def unsubscribe(self, sub_id):
for bucket in self.buckets:
if (chanId := bucket._get_chan_id(subId)):
await bucket._unsubscribe(chanId=chanId)
if (chan_id := bucket.get_chan_id(sub_id)):
await bucket.unsubscribe(chan_id=chan_id)
async def close(self, code=1000, reason=str()):
if self.websocket != None and self.websocket.open == True:
if self.websocket is not None and self.websocket.open:
await self.websocket.close(code=code, reason=reason)
for bucket in self.buckets:
await bucket._close(code=code, reason=reason)
await bucket.close(code=code, reason=reason)
@_require_websocket_authentication
async def notify(self, info, MESSAGE_ID=None, **kwargs):
await self.websocket.send(json.dumps([ 0, "n", MESSAGE_ID, { "type": "ucm-test", "info": info, **kwargs } ]))
async def notify(self, info, message_id=None, **kwargs):
await self.websocket.send(json.dumps([ 0, "n", message_id, { "type": "ucm-test", "info": info, **kwargs } ]))
@_require_websocket_authentication
async def __handle_websocket_input(self, input, data):
await self.websocket.send(json.dumps([ 0, input, None, data], cls=JSONEncoder))
async def __handle_websocket_input(self, event, data):
await self.websocket.send(json.dumps([ 0, event, None, data], cls=JSONEncoder))
def on(self, *events, callback = None):
for event in events:
if event not in BfxWebsocketClient.EVENTS:
raise EventNotSupported(f"Event <{event}> is not supported. To get a list of available events print BfxWebsocketClient.EVENTS")
raise EventNotSupported(f"Event <{event}> is not supported. To get a list "
+ "of available events print BfxWebsocketClient.EVENTS")
if callback != None:
if callback is not None:
for event in events:
self.event_emitter.on(event, callback)
if callback == None:
if callback is None:
def handler(function):
for event in events:
self.event_emitter.on(event, function)
return handler
return handler
def once(self, *events, callback = None):
for event in events:
if event not in BfxWebsocketClient.EVENTS:
raise EventNotSupported(f"Event <{event}> is not supported. To get a list of available events print BfxWebsocketClient.EVENTS")
raise EventNotSupported(f"Event <{event}> is not supported. To get a list "
+ "of available events print BfxWebsocketClient.EVENTS")
if callback != None:
if callback is not None:
for event in events:
self.event_emitter.once(event, callback)
if callback == None:
if callback is None:
def handler(function):
for event in events:
self.event_emitter.once(event, function)
return handler
return handler