I was disgusted of asyncio.Queue() so i made my own queue class

Question

I'm currently making a discord bot playing music, I've been having a lot of problems these days with the song queue so I decided to make my own. I used to read music bot repos and it seems they use some kind of lists optimized specifically for append() and pop(). Could someone please enlighten me on this as I think it would be a great improvement in my code.

If you see any other improvements, don't hesitate, I'm interested

class Song_Queue():
    def __init__(self):
        self.queue = []
    
    async def add_song(self, song, song_name, message_author, song_duration):
        queue = self.queue
        queue.append([song, song_name, message_author, song_duration])
    
    #self.queue[x] = ...
    # x = 0 : fichier son à diffuser
    # x = 1 : nom du son
    # x = 2 : la durée de la musique
    # x = 3 : la personne qui a ajouté la musique au bot
    
    async def get_next_song(self):
        queue = self.queue
        print(f'{queue}')
        return queue[0]
    
    async def remove_song(self):
        self.queue.pop(0)
        return
    
    async def retrieve_all_data(self):
        return self.queue

Answer 1

I disagree with your implementation of a queue:

1. Why are are you doing it? You do not offer anything new on top of the underlying list. add_song() just adds all of the parameters to the list, get_next_song() simply returns the first item, remove_song() pops it and retrieve_all_data() just returns the list. A list is a simple construct, why complicate?
2. If you are trying to encapsulate, you can just use simple functions, on top of a list like so:

   def get_next_song(queue):
       return queue[0]
   

   No need to create your own class. Using regular functions is simple and great for this case. You can create your own class, if you wish, but this is a much simpler solution (See heapq in the standard library for an example).
3. The underlying data structure should actually be a queue and not a list. You're building a queue, so a deque is considerably more appropriate, and is O(1) on popping from the start of the queue unlike the O(n) of a list.
4. You're using async def on synchronous functions, which is unnecessary.
5. I personally would make a song an object (a dataclass) which will have a name, duration, artist etc., and not many different objects in a list.
6. Don't print inside internal class functions. You will not be able to use them if you wish to suppress the print. You should almost always print on the outside.

---

Here is my suggestion:

from collections import deque
from dataclasses import dataclass

@dataclass
class Song:
    name: str
    duration: float
    author: str

SongQueue = deque[Song]
song_queue: SongQueue = deque()

# These functions are unnecessary, as you can simply use the deque.
def add_song(queue: SongQueue, song: Song) -> None:
    queue.append(song)

def pop_first_song(queue: SongQueue) -> Song:
    return queue.popleft()

def pop_last_song(queue: SongQueue) -> Song:
    return queue.pop()

def get_all_songs(queue: SongQueue) -> SongQueue:
    return queue

def get_next_song(queue: SongQueue) -> Song:
    return queue[0]

---

Alternative class implementation (which I don't think is necessary):

from collections import deque
from dataclasses import dataclass
from typing import Self

@dataclass
class Song:
    name: str
    duration: float
    author: str

class SongQueue(deque[Song]):
    def add_song(self, song: Song) -> None:
        self.append(song)

    def pop_first_song(self) -> Song:
        return self.popleft()

    def pop_last_song(self) -> Song:
        return self.pop()

    def get_all_songs(self) -> Self:
        return self

    def get_next_song(self) -> Song:
        return self[0]

Answer 2

There's quite a fundamental issue here, and that's that this solution doesn't work. Asyncio Queues are designed to coordinate adding and removing elements between multiple coroutines, which they do using locking mechanisms, in this case you have no locks, so things will fail.

For example:

from asyncio import gather, Queue, sleep

async def add_items(queue: Queue):
    for item in range(5):
        await sleep(1)
        print(f"Adding item {item} to queue")
        await queue.put(f"Item {item}")

async def receive_and_print_items(queue: Queue):
    for _ in range(5):
        item = await queue.get()
        print(f"Received: {item}")

queue = Queue()

await gather(add_items(queue), receive_and_print_items(queue))

^This gives us the following, as expected:

Adding item 0 to queue
Received: Item 0
Adding item 1 to queue
Received: Item 1
Adding item 2 to queue
Received: Item 2
Adding item 3 to queue
Received: Item 3
Adding item 4 to queue
Received: Item 4

If we try something similar with your code:

async def add_songs(song_queue):
    for index in range(5):
        await sleep(0.5)
        await song_queue.add_song(f"Song {index}", "Song name", "author", "duration")

async def play_songs(song_queue):
    for index in range(5):
        song = await song_queue.get_next_song()
        print(song)
        await song_queue.remove_song()


song_queue = Song_Queue()

await gather(add_songs(song_queue), play_songs(song_queue))

We immediately error out with an index error, this is because the coroutine that tries to get songs from the queue is trying to get songs before anything has been added, so 'queue[0]' doesn't exist.

In the real world, this would mean as soon as the playlist was empty and a bot tried to get the next item it would crash, compared to the native async version which would wait patiently until someone added something to the playlist, at which point it would play it.

You could have your code use an asyncio queue instead of a list, but we'd need to see the code that is using this to be able to advise on what you should do here, do you have any examples?