38 lines
1.1 KiB
Python
38 lines
1.1 KiB
Python
from dataclasses import dataclass, field, _MISSING_TYPE as mt
|
|
from asyncio import iscoroutinefunction
|
|
|
|
@dataclass
|
|
class Event():
|
|
event_type:str
|
|
data:Object=None
|
|
raw_data:dict=None
|
|
outbound:bool=False
|
|
|
|
class Listener():
|
|
def __init__(self,name,match,function):
|
|
self.name=name or function.__name__
|
|
self.match=match
|
|
self._matchstr=isinstance(match,str)
|
|
self.function=function
|
|
self._async=iscoroutinefunction(function)
|
|
|
|
async def __call__(self,*args,**kwargs):
|
|
if not self._async: return self.function(*args,**kwargs)
|
|
return await self.function(*args,**kwargs)
|
|
|
|
def __eq__(self,other):
|
|
if isinstance(other,Event):
|
|
if self._matchstr: return self.match==other.event_type
|
|
return self.match(other) # If it's not a string, assume it's a callable.
|
|
else: return super.__eq__(self,other)
|
|
def __str__(self): return self.name
|
|
|
|
def get_or_create(needle,haystack,default):
|
|
"""
|
|
This is a wrapper for filter that can add stuff. Nothing special. Needle is a function, default isn't. Haystack is a list. I might fix that later.
|
|
"""
|
|
f=filter(needle,haystack)
|
|
try: return next(f)
|
|
except StopIteration: pass
|
|
haystack.append(default)
|
|
return default |