tessl/pypi-wavelink
A robust and powerful, fully asynchronous Lavalink wrapper built for discord.py in Python.
node-management.mddocs/
Node Management
Comprehensive connection and management of Lavalink servers including pool management, node discovery, health monitoring, and load balancing across multiple nodes. The node system provides the foundation for all wavelink functionality by managing connections to Lavalink servers.
Capabilities
Pool Management
The Pool class provides global management of Lavalink nodes with automatic load balancing, connection pooling, and node selection for optimal performance.
class Pool:
@classmethod
async def connect(
cls,
uri: str,
password: str,
*,
identifier: str | None = None,
client: discord.Client,
**kwargs
) -> Node:
"""
Connect to a Lavalink node and add it to the pool.
Parameters:
- uri: The URI of the Lavalink server (e.g., "http://localhost:2333")
- password: Authentication password for the Lavalink server
- identifier: Optional unique identifier for the node
- client: The discord.py client instance
- **kwargs: Additional connection parameters
Returns:
Node: The connected Node instance
Raises:
InvalidClientException: If the client is invalid
AuthorizationFailedException: If authentication fails
"""
@classmethod
async def reconnect(cls) -> dict[str, Node]:
"""
Reconnect all nodes in the pool.
Returns:
dict[str, Node]: Dictionary of reconnected nodes by identifier
"""
@classmethod
async def close(cls) -> None:
"""Close all node connections in the pool."""
@classmethod
def get_node(cls, identifier: str | None = None) -> Node:
"""
Get a node from the pool by identifier.
Parameters:
- identifier: Node identifier, if None returns best available node
Returns:
Node: The requested node
Raises:
InvalidNodeException: If node doesn't exist or pool is empty
"""
@classmethod
async def fetch_tracks(
cls,
query: str,
*,
node: Node | None = None
) -> list[Playable] | Playlist:
"""
Search for tracks using the pool's nodes.
Parameters:
- query: Search query (URL, search terms, etc.)
- node: Specific node to use, if None uses best available
Returns:
list[Playable] | Playlist: Search results
Raises:
LavalinkLoadException: If track loading fails
"""
@classmethod
def cache(cls, capacity: int | None | bool = None) -> None:
"""
Configure track caching for the pool.
Parameters:
- capacity: Cache capacity (None for unlimited, False to disable)
"""
@classmethod
def has_cache(cls) -> bool:
"""
Check if track caching is enabled.
Returns:
bool: True if caching is enabled
"""
@classmethod
@property
def nodes(cls) -> dict[str, Node]:
"""
All connected nodes in the pool.
Returns:
dict[str, Node]: Dictionary of nodes by identifier
"""Node Connection
Individual Lavalink server connection management with REST API access, WebSocket communication, and comprehensive server interaction.
class Node:
def __init__(
self,
*,
uri: str,
password: str,
identifier: str | None = None,
client: discord.Client | None = None,
**kwargs
):
"""
Initialize a new Node connection.
Parameters:
- uri: Lavalink server URI
- password: Authentication password
- identifier: Unique node identifier
- client: Discord client instance
"""
@property
def headers(self) -> dict[str, str]:
"""HTTP headers used for REST requests."""
@property
def identifier(self) -> str:
"""Unique identifier for this node."""
@property
def uri(self) -> str:
"""URI of the Lavalink server."""
@property
def status(self) -> NodeStatus:
"""Current connection status of the node."""
@property
def players(self) -> dict[int, Player]:
"""Dictionary of active players on this node by guild ID."""
@property
def client(self) -> discord.Client | None:
"""Connected Discord client instance."""
@property
def password(self) -> str:
"""Authentication password for the node."""
@property
def heartbeat(self) -> float:
"""WebSocket heartbeat interval."""
@property
def session_id(self) -> str | None:
"""Current WebSocket session ID."""
async def close(self, *, eject: bool = False) -> None:
"""
Close the node connection.
Parameters:
- eject: Whether to eject players before closing
"""
async def send(self, **data) -> None:
"""
Send data to the node via WebSocket.
Parameters:
- **data: Data to send to the node
"""
async def fetch_players(self) -> list[PlayerResponsePayload]:
"""
Fetch information about all players on this node.
Returns:
list[PlayerResponsePayload]: List of player information
"""
async def fetch_player_info(self, guild_id: int) -> PlayerResponsePayload | None:
"""
Fetch information about a specific player.
Parameters:
- guild_id: Discord guild ID
Returns:
PlayerResponsePayload | None: Player information or None if not found
"""
async def fetch_info(self) -> InfoResponsePayload:
"""
Fetch node and Lavalink server information.
Returns:
InfoResponsePayload: Node information including version, plugins, etc.
"""
async def fetch_stats(self) -> StatsResponsePayload:
"""
Fetch node performance statistics.
Returns:
StatsResponsePayload: Node statistics including CPU, memory, etc.
"""
async def fetch_version(self) -> str:
"""
Fetch the Lavalink server version.
Returns:
str: Server version string
"""
def get_player(self, guild_id: int) -> Player | None:
"""
Get a player by guild ID.
Parameters:
- guild_id: Discord guild ID
Returns:
Player | None: Player instance or None if not found
"""Node Status
Enumeration of possible node connection states for monitoring and health checking.
class NodeStatus(enum.Enum):
"""
Enum representing the connection status of a Node.
"""
DISCONNECTED = 0 # Never connected or disconnected
CONNECTING = 1 # Currently attempting connection
CONNECTED = 2 # Successfully connectedUsage Examples
Basic Node Connection
import wavelink
import discord
# Connect to a single Lavalink node
bot = commands.Bot(command_prefix='!', intents=discord.Intents.all())
@bot.event
async def on_ready():
# Connect to local Lavalink server
node = await wavelink.Pool.connect(
uri="http://localhost:2333",
password="youshallnotpass",
client=bot,
identifier="main_node"
)
print(f"Connected to node: {node.identifier}")Multiple Node Setup
@bot.event
async def on_ready():
# Connect to multiple nodes for redundancy
nodes = [
("http://localhost:2333", "password1", "node1"),
("http://lavalink2.example.com:2333", "password2", "node2"),
]
for uri, password, identifier in nodes:
try:
await wavelink.Pool.connect(
uri=uri,
password=password,
client=bot,
identifier=identifier
)
print(f"Connected to {identifier}")
except wavelink.AuthorizationFailedException:
print(f"Failed to authenticate with {identifier}")Node Health Monitoring
async def check_node_health():
"""Check the health of all connected nodes."""
for identifier, node in wavelink.Pool.nodes.items():
if node.status == wavelink.NodeStatus.CONNECTED:
try:
stats = await node.fetch_stats()
print(f"Node {identifier}: CPU {stats['cpu']['used']}%, Memory {stats['memory']['used']}MB")
except Exception as e:
print(f"Failed to get stats for {identifier}: {e}")
else:
print(f"Node {identifier} is {node.status.name}")Caching Configuration
# Enable track caching with 1000 track capacity
wavelink.Pool.cache(capacity=1000)
# Check if caching is enabled
if wavelink.Pool.has_cache():
print("Track caching is enabled")
# Disable caching
wavelink.Pool.cache(capacity=False)Install with Tessl CLI
npx tessl i tessl/pypi-wavelink