ref vs WeakMethod

It's hard to correctly accept a callback function and keep a weakref to it.

Problem

First, what works:

import inspect, weakref

class C:
   def m(self): pass

c = C()
w1 = weakref.ref(c, print)
del c
<weakref at 0x7f41575dcc20; dead>
# good

If the caller sends you a bound method, however, that code is wrong:

c = C()
w2 = weakref.ref(c.m, print)
<weakref at 0x7f4158da6250; dead>
# too soon- the bound method is transient and cleaned up
del c

There's a fix, though! IMHO it should be transparent, but here we are:

c = C()
w3 = weakref.WeakMethod(c.m, print)
del c
<weakref at 0x7f41577c1af0; dead>
# good

Of course, you want your API to accept not-bound-methods and bound-methods, just like anything else would. Can't rely on WeakMethod:

c = C()
w4 = weakref.WeakMethod(c, print)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/usr/lib/python3.11/weakref.py", line 51, in __new__
    raise TypeError("argument should be a bound method, not {}"
TypeError: argument should be a bound method, not <class '__main__.C'>
# oh come on; there's only one thing I could want here

Solution

So the best I've figured out is to branch like this:

def print_when_cb_is_deleted1(cb):
  if inspect.ismethod(cb):
    w = weakref.WeakMethod(cb, print)
  else:
    w = weakref.ref(cb, print)
# ew

Or, barely less readable, factored out like this:

def print_when_cb_is_deleted2(cb):
  w = (weakref.WeakMethod if inspect.ismethod(cb) else weakref.ref)(cb, print)

Real code

FTR, here's my actual context. Deleting the callback shall unsubscribe the mqtt topic (if no other callbacks need it):

    async def subscribe(self, topic: str, cb: Callable[[float, str], None]):

        if topic not in self._subs:
            self._subs[topic] = []

        if inspect.ismethod(cb):
            ref = weakref.WeakMethod(cb, lambda _: self._cbDeleted(topic))
        else:
            ref = weakref.ref(cb, lambda _: self._cbDeleted(topic))

        self._subs[topic].append(ref)
        ...
            await self.mqttSub(topic)


    def _cbDeleted(self, topic: str):
        if topic not in self._subs:
            return
        self._subs[topic] = [v for v in self._subs[topic] if v() is not None]
        if not self._subs[topic]:
            asyncio.create_task(self.mqttUnsub(topic)) # small race condition here
            del self._subs[topic]