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

Remove type hinting for decorators _require_websocket_connection and _require_websocket_authentication.

Browse Source
This commit is contained in:
Davide Casale
2023-06-12 16:17:45 +02:00
parent d63c2c63c3
commit cc5f9f5b0e
2 changed files with 5 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
bfxapi/websocket/client/bfx_websocket_bucket.py
View File

@@ -1,5 +1,3 @@
from typing import Literal, TypeVar, Callable, cast
import asyncio, json, uuid, websockets import asyncio, json, uuid, websockets
from ..handlers import PublicChannelsHandler from ..handlers import PublicChannelsHandler
@@ -8,16 +6,14 @@ from ..exceptions import ConnectionNotOpen, TooManySubscriptions
_HEARTBEAT = "hb" _HEARTBEAT = "hb"
F = TypeVar("F", bound=Callable[..., Literal[None]]) def _require_websocket_connection(function):
def _require_websocket_connection(function: F) -> F:
async def wrapper(self, *args, **kwargs): async def wrapper(self, *args, **kwargs):
if self.websocket is None or not self.websocket.open: if self.websocket is None or not self.websocket.open:
raise ConnectionNotOpen("No open connection with the server.") raise ConnectionNotOpen("No open connection with the server.")
await function(self, *args, **kwargs) await function(self, *args, **kwargs)
return cast(F, wrapper) return wrapper
class BfxWebSocketBucket: class BfxWebSocketBucket:
VERSION = 2 VERSION = 2

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

@@ -1,5 +1,3 @@
from typing import cast
from collections import namedtuple from collections import namedtuple
from datetime import datetime from datetime import datetime
@@ -8,7 +6,7 @@ import traceback, json, asyncio, hmac, hashlib, time, socket, random, websockets
from pyee.asyncio import AsyncIOEventEmitter from pyee.asyncio import AsyncIOEventEmitter
from .bfx_websocket_bucket import _HEARTBEAT, F, _require_websocket_connection, BfxWebSocketBucket from .bfx_websocket_bucket import _HEARTBEAT, _require_websocket_connection, BfxWebSocketBucket
from .bfx_websocket_inputs import BfxWebSocketInputs from .bfx_websocket_inputs import BfxWebSocketInputs
from ..handlers import PublicChannelsHandler, AuthEventsHandler from ..handlers import PublicChannelsHandler, AuthEventsHandler
@@ -19,7 +17,7 @@ from ...utils.json_encoder import JSONEncoder
from ...utils.logger import ColorLogger, FileLogger from ...utils.logger import ColorLogger, FileLogger
def _require_websocket_authentication(function: F) -> F: def _require_websocket_authentication(function):
async def wrapper(self, *args, **kwargs): async def wrapper(self, *args, **kwargs):
if hasattr(self, "authentication") and not self.authentication: if hasattr(self, "authentication") and not self.authentication:
raise WebSocketAuthenticationRequired("To perform this action you need to " \ raise WebSocketAuthenticationRequired("To perform this action you need to " \
@@ -27,7 +25,7 @@ def _require_websocket_authentication(function: F) -> F:
await _require_websocket_connection(function)(self, *args, **kwargs) await _require_websocket_connection(function)(self, *args, **kwargs)
return cast(F, wrapper) return wrapper
class _Delay: class _Delay:
BACKOFF_MIN, BACKOFF_MAX = 1.92, 60.0 BACKOFF_MIN, BACKOFF_MAX = 1.92, 60.0