mirror of
https://github.com/onyx-and-iris/obsws-python.git
synced 2024-11-25 06:10:52 +00:00
27fd86efa5
only pass auth token if auth enabled add context manager methods to reqclient. added logging
54 lines
1.5 KiB
Python
54 lines
1.5 KiB
Python
from typing import Callable, Iterable, Union
|
|
|
|
from .util import as_dataclass, to_camel_case, to_snake_case
|
|
|
|
|
|
class Callback:
|
|
"""Adds support for callbacks"""
|
|
|
|
def __init__(self):
|
|
"""list of current callbacks"""
|
|
|
|
self._callbacks = list()
|
|
|
|
def get(self) -> list:
|
|
"""returns a list of registered events"""
|
|
|
|
return [to_camel_case(fn.__name__[2:]) for fn in self._callbacks]
|
|
|
|
def trigger(self, event, data):
|
|
"""trigger callback on event"""
|
|
|
|
for fn in self._callbacks:
|
|
if fn.__name__ == f"on_{to_snake_case(event)}":
|
|
fn(as_dataclass(event, data))
|
|
|
|
def register(self, fns: Union[Iterable, Callable]):
|
|
"""registers callback functions"""
|
|
|
|
try:
|
|
iterator = iter(fns)
|
|
for fn in iterator:
|
|
if fn not in self._callbacks:
|
|
self._callbacks.append(fn)
|
|
except TypeError:
|
|
if fns not in self._callbacks:
|
|
self._callbacks.append(fns)
|
|
|
|
def deregister(self, fns: Union[Iterable, Callable]):
|
|
"""deregisters callback functions"""
|
|
|
|
try:
|
|
iterator = iter(fns)
|
|
for fn in iterator:
|
|
if fn in self._callbacks:
|
|
self._callbacks.remove(fn)
|
|
except TypeError:
|
|
if fns in self._callbacks:
|
|
self._callbacks.remove(fns)
|
|
|
|
def clear(self):
|
|
"""clears the _callbacks list"""
|
|
|
|
self._callbacks.clear()
|