Source code for copra.websocket.channel

# -*- coding: utf-8 -*-
"""WebSocket channel class for use with the copra WebSocket client.

"""

[docs]class Channel: """A WebSocket channel. A Channel object encapsulates the Coinbase Pro WebSocket channel name *and* one or more Coinbase Pro product ids. To read about Coinbase Pro channels and the data they return, visit: https://docs.gdax.com/#channels :ivar str name: The name of the WebSocket channel. :ivar product_ids: Product ids for the channel. :vartype product_ids: set of str """
[docs] def __init__(self, name, product_ids): """ :param str name: The name of the WebSocket channel. Possible values are heatbeat, ticker, level2, full, matches, or user :param product_ids: A single product id (eg., 'BTC-USD') or list of product ids (eg., ['BTC-USD', 'ETH-EUR', 'LTC-BTC']) :type product_ids: str or list of str :raises ValueError: If name not valid or product ids is empty. """ self.name = name.lower() if self.name not in ('heartbeat', 'ticker', 'level2', 'full', 'matches', 'user'): raise ValueError("invalid name {}".format(name)) if not product_ids: raise ValueError("must include at least one product id") if not isinstance(product_ids, list): product_ids = [product_ids] self.product_ids = set(product_ids)
def __repr__(self): return str(self._as_dict()) def _as_dict(self): """Returns the Channel as a dictionary. :returns dict: The Channel as a dict with keys name & value list of product_ids. """ return {'name': self.name, 'product_ids': list(self.product_ids)} def __eq__(self, other): if self.name != other.name: raise TypeError('Channels need the same name to be compared.') return (self.name == other.name and self.product_ids == other.product_ids) def __add__(self, other): if self.name != other.name: raise TypeError('Channels need the same name to be added.') return Channel(self.name, list(self.product_ids | other.product_ids)) def __sub__(self, other): if self.name != other.name: raise TypeError('Channels need the same name to be subtracted.') product_ids = self.product_ids - other.product_ids if not product_ids: return None return Channel(self.name, list(product_ids))