(
self,
host: Optional[str] = None,
port: Optional[int] = None,
node_name: Optional[str] = None,
)
| 1804 | self._event_dispatcher = event_dispatcher |
| 1805 | |
| 1806 | def get_node( |
| 1807 | self, |
| 1808 | host: Optional[str] = None, |
| 1809 | port: Optional[int] = None, |
| 1810 | node_name: Optional[str] = None, |
| 1811 | ) -> Optional["ClusterNode"]: |
| 1812 | if host and port: |
| 1813 | # the user passed host and port |
| 1814 | if host == "localhost": |
| 1815 | host = socket.gethostbyname(host) |
| 1816 | return self.nodes_cache.get(get_node_name(host=host, port=port)) |
| 1817 | elif node_name: |
| 1818 | return self.nodes_cache.get(node_name) |
| 1819 | else: |
| 1820 | raise DataError( |
| 1821 | "get_node requires one of the following: 1. node name 2. host and port" |
| 1822 | ) |
| 1823 | |
| 1824 | def set_nodes( |
| 1825 | self, |
no test coverage detected