Coverage for adrs/io/event.py: 64%

1import types 

2from enum import Enum 

3from abc import ABC, abstractmethod 

4from typing import Any 

5 

6 

7class EventType(str, Enum): 

8 Unknown = "unknown" 

9 Authenticated = "authenticated" 

10 Subscribed = "subscribed" 

11 OrderUpdate = "order_update" 

12 DatasourceSubscribed = "datasource_subscribed" 

13 DatasourceUpdate = "datasource_update" 

14 Error = "error" 

15 

16 

17class Event: 

18 event_type: EventType 

19 data: Any 

20 orig: Any 

21 

22 def __init__(self, event_type: EventType, orig: Any, data: Any | None = None): 

23 self.event_type = event_type 

24 self.orig = orig 

25 self.data = data 

26 

27 

28class EventHandler(ABC): 

29 """Base class for event handling.""" 

30 

31 def __setattr__(self, name, value): 

32 """ 

33 This is to allow more advanced use case where the user does not want to be 

34 tied to the provided starter class `BaseStrategy`. 

35 """ 

36 if callable(value) and hasattr(value, "__code__"): 

37 if value.__class__ == types.MethodType: 

38 pass 

39 elif value.__class__ == types.FunctionType and ( 

40 value.__code__.co_argcount > 0 

41 and value.__code__.co_varnames[0] == "self" 

42 ): 

43 value = types.MethodType(value, self) 

44 

45 super().__setattr__(name, value) 

46 

47 @abstractmethod 

48 async def on_event(self, event: Event) -> None: 

49 """Called when message is parsed and identified.""" 

50 pass 

51 

52 @abstractmethod 

53 async def start(self): 

54 pass