mirror of
https://github.com/aljazceru/bitfinex-api-py.git
synced 2025-12-18 22:34:21 +01:00
Apply pylint's linting rules to bfxapi/__init__.py, bfxapi/enums.py, bfxapi/exceptions.py, bfxapi/labeler.py, bfxapi/notification.py and bfxapi/urls.py.
This commit is contained in:
@@ -1,6 +1,6 @@
|
||||
[MAIN]
|
||||
|
||||
py-version = 3.8.0
|
||||
py-version=3.8.0
|
||||
|
||||
ignore=examples
|
||||
|
||||
@@ -8,4 +8,9 @@ ignore=examples
|
||||
|
||||
disable=
|
||||
missing-docstring,
|
||||
too-few-public-methods
|
||||
too-few-public-methods,
|
||||
dangerous-default-value
|
||||
|
||||
[FORMAT]
|
||||
|
||||
max-line-length=120
|
||||
@@ -3,4 +3,4 @@ from .client import Client
|
||||
from .urls import REST_HOST, PUB_REST_HOST, \
|
||||
WSS_HOST, PUB_WSS_HOST
|
||||
|
||||
NAME = "bfxapi"
|
||||
NAME = "bfxapi"
|
||||
|
||||
@@ -47,4 +47,4 @@ class Error(int, Enum):
|
||||
ERR_SUB_LIMIT = 10305
|
||||
ERR_UNSUB_FAIL = 10400
|
||||
ERR_UNSUB_NOT = 10401
|
||||
ERR_READY = 11000
|
||||
ERR_READY = 11000
|
||||
|
||||
@@ -9,11 +9,8 @@ class BfxBaseException(Exception):
|
||||
Base class for every custom exception in bfxapi/rest/exceptions.py and bfxapi/websocket/exceptions.py.
|
||||
"""
|
||||
|
||||
pass
|
||||
|
||||
class LabelerSerializerException(BfxBaseException):
|
||||
"""
|
||||
This exception indicates an error thrown by the _Serializer class in bfxapi/labeler.py.
|
||||
"""
|
||||
|
||||
pass
|
||||
|
||||
@@ -1,12 +1,12 @@
|
||||
from .exceptions import LabelerSerializerException
|
||||
|
||||
from typing import Type, Generic, TypeVar, Iterable, Optional, Dict, List, Tuple, Any, cast
|
||||
|
||||
from .exceptions import LabelerSerializerException
|
||||
|
||||
T = TypeVar("T", bound="_Type")
|
||||
|
||||
def compose(*decorators):
|
||||
def wrapper(function):
|
||||
for decorator in reversed(decorators):
|
||||
for decorator in reversed(decorators):
|
||||
function = decorator(function)
|
||||
return function
|
||||
|
||||
@@ -28,36 +28,37 @@ def partial(cls):
|
||||
|
||||
return cls
|
||||
|
||||
class _Type(object):
|
||||
class _Type:
|
||||
"""
|
||||
Base class for any dataclass serializable by the _Serializer generic class.
|
||||
"""
|
||||
|
||||
pass
|
||||
|
||||
class _Serializer(Generic[T]):
|
||||
def __init__(self, name: str, klass: Type[_Type], labels: List[str], IGNORE: List[str] = [ "_PLACEHOLDER" ]):
|
||||
self.name, self.klass, self.__labels, self.__IGNORE = name, klass, labels, IGNORE
|
||||
def __init__(self, name: str, klass: Type[_Type], labels: List[str],
|
||||
*, ignore: List[str] = [ "_PLACEHOLDER" ]):
|
||||
self.name, self.klass, self.__labels, self.__ignore = name, klass, labels, ignore
|
||||
|
||||
def _serialize(self, *args: Any, skip: Optional[List[str]] = None) -> Iterable[Tuple[str, Any]]:
|
||||
labels = list(filter(lambda label: label not in (skip or list()), self.__labels))
|
||||
labels = list(filter(lambda label: label not in (skip or []), self.__labels))
|
||||
|
||||
if len(labels) > len(args):
|
||||
raise LabelerSerializerException(f"{self.name} -> <labels> and <*args> arguments should contain the same amount of elements.")
|
||||
raise LabelerSerializerException(f"{self.name} -> <labels> and <*args> " +
|
||||
"arguments should contain the same amount of elements.")
|
||||
|
||||
for index, label in enumerate(labels):
|
||||
if label not in self.__IGNORE:
|
||||
if label not in self.__ignore:
|
||||
yield label, args[index]
|
||||
|
||||
def parse(self, *values: Any, skip: Optional[List[str]] = None) -> T:
|
||||
return cast(T, self.klass(**dict(self._serialize(*values, skip=skip))))
|
||||
|
||||
def get_labels(self) -> List[str]:
|
||||
return [ label for label in self.__labels if label not in self.__IGNORE ]
|
||||
return [ label for label in self.__labels if label not in self.__ignore ]
|
||||
|
||||
class _RecursiveSerializer(_Serializer, Generic[T]):
|
||||
def __init__(self, name: str, klass: Type[_Type], labels: List[str], serializers: Dict[str, _Serializer[Any]], IGNORE: List[str] = ["_PLACEHOLDER"]):
|
||||
super().__init__(name, klass, labels, IGNORE)
|
||||
def __init__(self, name: str, klass: Type[_Type], labels: List[str],
|
||||
*, serializers: Dict[str, _Serializer[Any]], ignore: List[str] = ["_PLACEHOLDER"]):
|
||||
super().__init__(name, klass, labels, ignore = ignore)
|
||||
|
||||
self.serializers = serializers
|
||||
|
||||
@@ -70,8 +71,11 @@ class _RecursiveSerializer(_Serializer, Generic[T]):
|
||||
|
||||
return cast(T, self.klass(**serialization))
|
||||
|
||||
def generate_labeler_serializer(name: str, klass: Type[T], labels: List[str], IGNORE: List[str] = [ "_PLACEHOLDER" ]) -> _Serializer[T]:
|
||||
return _Serializer[T](name, klass, labels, IGNORE)
|
||||
def generate_labeler_serializer(name: str, klass: Type[T], labels: List[str],
|
||||
*, ignore: List[str] = [ "_PLACEHOLDER" ]) -> _Serializer[T]:
|
||||
return _Serializer[T](name, klass, labels, ignore=ignore)
|
||||
|
||||
def generate_recursive_serializer(name: str, klass: Type[T], labels: List[str], serializers: Dict[str, _Serializer[Any]], IGNORE: List[str] = [ "_PLACEHOLDER" ]) -> _RecursiveSerializer[T]:
|
||||
return _RecursiveSerializer[T](name, klass, labels, serializers, IGNORE)
|
||||
def generate_recursive_serializer(name: str, klass: Type[T], labels: List[str],
|
||||
*, serializers: Dict[str, _Serializer[Any]], ignore: List[str] = [ "_PLACEHOLDER" ]
|
||||
) -> _RecursiveSerializer[T]:
|
||||
return _RecursiveSerializer[T](name, klass, labels, serializers=serializers, ignore=ignore)
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
from typing import List, Dict, Union, Optional, Any, TypedDict, Generic, TypeVar, cast
|
||||
from typing import List, Optional, Any, Generic, TypeVar, cast
|
||||
from dataclasses import dataclass
|
||||
from .labeler import _Type, _Serializer
|
||||
|
||||
@@ -7,7 +7,7 @@ T = TypeVar("T")
|
||||
@dataclass
|
||||
class Notification(_Type, Generic[T]):
|
||||
mts: int
|
||||
type: str
|
||||
type: str
|
||||
message_id: Optional[int]
|
||||
data: T
|
||||
code: Optional[int]
|
||||
@@ -18,7 +18,7 @@ class _Notification(_Serializer, Generic[T]):
|
||||
__LABELS = [ "mts", "type", "message_id", "_PLACEHOLDER", "data", "code", "status", "text" ]
|
||||
|
||||
def __init__(self, serializer: Optional[_Serializer] = None, is_iterable: bool = False):
|
||||
super().__init__("Notification", Notification, _Notification.__LABELS, IGNORE = [ "_PLACEHOLDER" ])
|
||||
super().__init__("Notification", Notification, _Notification.__LABELS, ignore = [ "_PLACEHOLDER" ])
|
||||
|
||||
self.serializer, self.is_iterable = serializer, is_iterable
|
||||
|
||||
@@ -28,11 +28,11 @@ class _Notification(_Serializer, Generic[T]):
|
||||
if isinstance(self.serializer, _Serializer):
|
||||
data = cast(List[Any], notification.data)
|
||||
|
||||
if self.is_iterable == False:
|
||||
if not self.is_iterable:
|
||||
if len(data) == 1 and isinstance(data[0], list):
|
||||
data = data[0]
|
||||
|
||||
notification.data = cast(T, self.serializer.klass(**dict(self.serializer._serialize(*data, skip=skip))))
|
||||
else: notification.data = cast(T, [ self.serializer.klass(**dict(self.serializer._serialize(*sub_data, skip=skip))) for sub_data in data ])
|
||||
notification.data = self.serializer.parse(*data, skip=skip)
|
||||
else: notification.data = cast(T, [ self.serializer.parse(*sub_data, skip=skip) for sub_data in data ])
|
||||
|
||||
return notification
|
||||
return notification
|
||||
|
||||
@@ -2,4 +2,4 @@ REST_HOST = "https://api.bitfinex.com/v2"
|
||||
PUB_REST_HOST = "https://api-pub.bitfinex.com/v2"
|
||||
|
||||
WSS_HOST = "wss://api.bitfinex.com/ws/2"
|
||||
PUB_WSS_HOST = "wss://api-pub.bitfinex.com/ws/2"
|
||||
PUB_WSS_HOST = "wss://api-pub.bitfinex.com/ws/2"
|
||||
|
||||
Reference in New Issue
Block a user