A robust and powerful, fully asynchronous Lavalink wrapper built for discord.py in Python.
Comprehensive track searching across multiple platforms (YouTube, SoundCloud, YouTube Music) with rich metadata, playlist support, and advanced search capabilities. The track system provides rich objects for individual tracks and playlists with extensive metadata and search functionality.
Global track searching across supported platforms with automatic source detection and comprehensive result handling.
# Global search function through Pool
async def fetch_tracks(
query: str,
*,
node: Node | None = None
) -> list[Playable] | Playlist:
"""
Search for tracks using the node pool.
Parameters:
- query: Search query (URL, search terms, or platform-specific query)
- node: Specific node to use (None for automatic selection)
Returns:
list[Playable] | Playlist: Search results as tracks or playlist
Raises:
LavalinkLoadException: If search fails or no results found
"""
# Track-specific search method
class Playable:
@classmethod
async def search(
cls,
query: str,
*,
source: TrackSource | str | None = TrackSource.YouTubeMusic,
node: Node | None = None
) -> Search:
"""
Search for tracks with optional source filtering.
Parameters:
- query: Search query string
- source: Specific source to search (default: YouTubeMusic)
- node: Specific node to use for search
Returns:
Search: List of tracks or playlist (type alias for list[Playable] | Playlist)
Note:
This method applies relevant search prefixes automatically when URLs are not provided.
"""
# Search result type alias
Search = list[Playable] | PlaylistIndividual track representation with comprehensive metadata and playback information.
class Playable:
@property
def encoded(self) -> str:
"""Lavalink encoded track data for internal use."""
@property
def identifier(self) -> str:
"""Unique track identifier from the source platform."""
@property
def is_seekable(self) -> bool:
"""Whether the track supports seeking to specific positions."""
@property
def author(self) -> str:
"""Track author, artist, or uploader name."""
@property
def length(self) -> int:
"""Track duration in milliseconds."""
@property
def is_stream(self) -> bool:
"""Whether the track is a live stream."""
@property
def position(self) -> int:
"""Current playback position in milliseconds."""
@property
def title(self) -> str:
"""Track title or name."""
@property
def uri(self) -> str:
"""Direct URI to the track source."""
@property
def artwork(self) -> str | None:
"""URL to track artwork/thumbnail image."""
@property
def isrc(self) -> str | None:
"""International Standard Recording Code."""
@property
def source(self) -> TrackSource:
"""Platform source of the track."""
@property
def album(self) -> Album | None:
"""Album information if available."""
@property
def artist(self) -> Artist | None:
"""Artist information if available."""
@property
def preview_url(self) -> str | None:
"""URL to a preview/snippet of the track."""
@property
def is_preview(self) -> bool:
"""Whether this track is a preview/snippet."""
@property
def playlist(self) -> PlaylistInfo | None:
"""Information about the source playlist."""
@property
def recommended(self) -> bool:
"""Whether this track was recommended by AutoPlay."""
@property
def extras(self) -> ExtrasNamespace:
"""Additional track metadata and custom attributes."""
@property
def raw_data(self) -> dict[str, Any]:
"""Raw track data from Lavalink server."""Collection of tracks with metadata for handling playlists from various sources.
class Playlist:
def __init__(self, data: dict, tracks: list[Playable]):
"""
Initialize a playlist with tracks and metadata.
Parameters:
- data: Playlist metadata from Lavalink
- tracks: List of tracks in the playlist
"""
@property
def name(self) -> str:
"""Playlist name or title."""
@property
def tracks(self) -> list[Playable]:
"""List of tracks in the playlist."""
@property
def track_count(self) -> int:
"""Number of tracks in the playlist."""
@property
def extras(self) -> ExtrasNamespace:
"""Additional playlist metadata."""
def pop(self, index: int = -1) -> Playable:
"""
Remove and return a track from the playlist.
Parameters:
- index: Index of track to remove (default: last track)
Returns:
Playable: The removed track
Raises:
IndexError: If index is out of range
"""
def track_extras(self, **attrs) -> None:
"""
Set extra attributes on all tracks in the playlist.
Parameters:
- **attrs: Attributes to set on all tracks
"""
# Container protocol methods
def __len__(self) -> int:
"""Return the number of tracks in the playlist."""
def __getitem__(self, index: int | slice) -> Playable | list[Playable]:
"""Get track(s) by index or slice."""
def __iter__(self) -> Iterator[Playable]:
"""Iterate over tracks in the playlist."""
def __bool__(self) -> bool:
"""Return True if playlist has tracks."""Supporting objects for rich track and playlist metadata.
class Album:
"""Album metadata container."""
name: str # Album name
url: str # Album URL or identifier
class Artist:
"""Artist metadata container."""
name: str # Artist name
url: str # Artist URL or identifier
class PlaylistInfo:
"""Playlist metadata container."""
name: str # Playlist name
selected_track: int # Index of selected track in playlistEnumeration of supported track sources for platform-specific searching.
class TrackSource(enum.Enum):
"""Enumeration of supported track sources."""
YouTube = 0 # YouTube videos
YouTubeMusic = 1 # YouTube Music tracks
SoundCloud = 2 # SoundCloud tracksimport wavelink
# Search for tracks using various query types
async def search_examples():
# Search by title and artist
tracks = await wavelink.Pool.fetch_tracks("Never Gonna Give You Up Rick Astley")
# Search with YouTube URL
tracks = await wavelink.Pool.fetch_tracks("https://www.youtube.com/watch?v=dQw4w9WgXcQ")
# Search SoundCloud URL
tracks = await wavelink.Pool.fetch_tracks("https://soundcloud.com/artist/track")
# Search playlist URL
result = await wavelink.Pool.fetch_tracks("https://www.youtube.com/playlist?list=...")
if isinstance(result, wavelink.Playlist):
print(f"Found playlist: {result.name} with {len(result)} tracks")
for track in result:
print(f"- {track.title} by {track.author}")
else:
print(f"Found {len(result)} tracks")
for track in result:
print(f"- {track.title} by {track.author} ({track.length // 1000}s)")# Search specific platforms
async def platform_search():
# Search only YouTube
youtube_tracks = await wavelink.Playable.search(
"rock music",
source=wavelink.TrackSource.YouTube
)
# Search only SoundCloud
soundcloud_tracks = await wavelink.Playable.search(
"electronic music",
source=wavelink.TrackSource.SoundCloud
)
# Search YouTube Music specifically
ytmusic_tracks = await wavelink.Playable.search(
"classical music",
source=wavelink.TrackSource.YouTubeMusic
)async def display_track_info(track: wavelink.Playable):
"""Display comprehensive track information."""
print(f"Title: {track.title}")
print(f"Artist: {track.author}")
print(f"Duration: {track.length // 60000}:{(track.length // 1000) % 60:02d}")
print(f"Source: {track.source.name}")
print(f"Seekable: {track.is_seekable}")
print(f"Stream: {track.is_stream}")
print(f"URI: {track.uri}")
if track.artwork:
print(f"Artwork: {track.artwork}")
if track.album:
print(f"Album: {track.album.name}")
if track.artist:
print(f"Artist Info: {track.artist.name}")
if track.isrc:
print(f"ISRC: {track.isrc}")
# Display any extra metadata
if track.extras:
print("Extra metadata:", dict(track.extras))async def handle_playlist(query: str):
"""Handle playlist search results."""
result = await wavelink.Pool.fetch_tracks(query)
if isinstance(result, wavelink.Playlist):
playlist = result
print(f"Playlist: {playlist.name}")
print(f"Track count: {len(playlist)}")
# Add custom metadata to all tracks
playlist.track_extras(added_by="AutoSearch", priority="high")
# Get first 5 tracks
first_tracks = playlist[:5]
# Remove and return last track
last_track = playlist.pop()
# Iterate through all tracks
for i, track in enumerate(playlist):
print(f"{i+1}. {track.title} - {track.author}")
return playlist
else:
# Handle individual tracks
tracks = result
print(f"Found {len(tracks)} individual tracks")
return tracks@bot.command()
async def search(ctx, *, query: str):
"""Search for tracks and display results."""
try:
result = await wavelink.Pool.fetch_tracks(query)
if isinstance(result, wavelink.Playlist):
# Handle playlist result
embed = discord.Embed(
title="Playlist Found",
description=f"**{result.name}**\n{len(result)} tracks",
color=discord.Color.green()
)
# Show first 10 tracks
track_list = []
for i, track in enumerate(result[:10]):
duration = f"{track.length // 60000}:{(track.length // 1000) % 60:02d}"
track_list.append(f"{i+1}. {track.title} - {track.author} ({duration})")
embed.add_field(
name="Tracks",
value="\n".join(track_list) + ("..." if len(result) > 10 else ""),
inline=False
)
else:
# Handle track list result
if not result:
return await ctx.send("No tracks found!")
embed = discord.Embed(
title="Search Results",
description=f"Found {len(result)} tracks",
color=discord.Color.blue()
)
# Show first 10 results
track_list = []
for i, track in enumerate(result[:10]):
duration = f"{track.length // 60000}:{(track.length // 1000) % 60:02d}"
track_list.append(f"{i+1}. {track.title} - {track.author} ({duration})")
embed.add_field(
name="Results",
value="\n".join(track_list),
inline=False
)
await ctx.send(embed=embed)
except wavelink.LavalinkLoadException as e:
await ctx.send(f"Search failed: {e.error}")@bot.command()
async def choose(ctx, *, query: str):
"""Search and let user choose a track."""
tracks = await wavelink.Pool.fetch_tracks(query)
if isinstance(tracks, wavelink.Playlist):
tracks = tracks.tracks[:10] # Limit to first 10 from playlist
elif isinstance(tracks, list):
tracks = tracks[:10] # Limit to first 10 results
else:
return await ctx.send("No tracks found!")
if not tracks:
return await ctx.send("No tracks found!")
# Display options
description = []
for i, track in enumerate(tracks, 1):
duration = f"{track.length // 60000}:{(track.length // 1000) % 60:02d}"
description.append(f"{i}. {track.title} - {track.author} ({duration})")
embed = discord.Embed(
title="Choose a track",
description="\n".join(description),
color=discord.Color.blue()
)
embed.set_footer(text="Type a number to select (1-{})".format(len(tracks)))
await ctx.send(embed=embed)
# Wait for user choice
def check(message):
return (message.author == ctx.author and
message.channel == ctx.channel and
message.content.isdigit() and
1 <= int(message.content) <= len(tracks))
try:
msg = await bot.wait_for('message', check=check, timeout=30.0)
choice = int(msg.content) - 1
selected_track = tracks[choice]
# Play the selected track
if not ctx.voice_client:
player = await ctx.author.voice.channel.connect(cls=wavelink.Player)
else:
player = ctx.voice_client
if player.playing:
player.queue.put(selected_track)
await ctx.send(f"Added to queue: {selected_track.title}")
else:
await player.play(selected_track)
await ctx.send(f"Now playing: {selected_track.title}")
except asyncio.TimeoutError:
await ctx.send("Selection timed out!")Install with Tessl CLI
npx tessl i tessl/pypi-wavelink