6724041435

Committed 01 Nov 2023 07:12PM UTC coverage: 86.654% (+0.02%) from 86.634%

Build # 6724041435

Build Type

push

github

Committed by

web-flow

Commit Message

Added ``off`` method for both local and distributed APIs. (#2)

* Added off method for both local and distributed APIs.

* Lowered stress test, added version.py, and bumped to 0.0.2.

Run Details

215 of 278 branches covered (0.0%)

Branch coverage included in aggregate %.

17 of 20 new or added lines in 6 files covered. (85.0%)

3 existing lines in 2 files now uncovered.

694 of 771 relevant lines covered (90.01%)

0.9 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.8

/aiodistbus/entrypoint/aentrypoint.py

import asyncio
import logging
import uuid
from abc import ABC, abstractmethod
from collections import deque
from typing import Any, Callable, Coroutine, Dict, List, Optional, Type

from ..protocols import Event, Handler
from ..registry import Registry
from ..utils import safe_coro

logger = logging.getLogger("aiodistbus")


class AEntryPoint(ABC):
    def __init__(self):
        """Abstract entrypoint for eventbus"""

        # State information
        self.id = str(uuid.uuid4())
        self._handlers: Dict[str, Handler] = {}
        self._wildcards: Dict[str, Handler] = {}
        self._received: deque[str] = deque(maxlen=1000)
        self._tasks: List[asyncio.Task] = []

    def _wrapper(
        self, func: Callable, unpack: bool = True, create_task: bool = False
    ) -> Callable:
        """Wrapper for handlers

        The wrapper provides an approach for logging which events have been transmitted

        Args:
            func (Callable): Function to wrap
            unpack (bool, optional): Unpack the event. Defaults to True.
            create_task (bool, optional): Create a task for the handler. Defaults to False.

        Returns:
            Callable: Wrapped function

        """
        # Wrapper for async functions
        async def awrapper(event: Event):
            coro: Optional[Coroutine] = None
            if unpack:
                if (
                    type(event.data) is not type(None)
                    and self._handlers[event.type].dtype
                ):
                    coro = func(event.data)
                else:
                    coro = func()
            else:
                coro = func(event)

            if coro:
                # Create safe coro with error msg
                scoro = safe_coro(
                    coro, f"Error in (type: {event.type}, handler {func.__name__})"
                )
                if create_task:
                    self._tasks.append(asyncio.create_task(scoro))
                else:
                    await scoro

            self._received.append(event.id)

        # Wrapper for sync functions
        async def wrapper(event: Event):

            try:
                if unpack:
                    func(event.data)
                else:
                    func(event)
            except Exception as e:
                logger.error(
                    f"Error in handler (type: {event.type}, handler {func.__name__}): {e}"
                )

            self._received.append(event.id)

        # Select according to function type
        if asyncio.iscoroutinefunction(func):
            return awrapper
        else:
            return wrapper

    @abstractmethod
    async def _update_handlers(
        self, event_type: Optional[str] = None, remove: bool = False
    ):
        ...

    ####################################################################
    ## Public API
    ####################################################################

    async def use(self, registry: Registry, namespace: str = "default"):
        """Use a registry

        Args:
            registry (Registry): Registry to use
            namespace (str, optional): Namespace. Defaults to "default".

        """
        # Obtain the handlers
        for event_type, handler in registry.get_handlers(namespace).items():
            await self.on(event_type, handler.function, handler.dtype)

    async def on(
        self,
        event_type: str,
        func: Callable,
        dtype: Optional[Type] = None,
        create_task: bool = False,
    ):
        """Register a handler

        Args:
            event_type (str): Event type
            func (Callable): Function to call
            dtype (Optional[Type], optional): Data type. Defaults to None.
            create_task (bool, optional): Create a task for the handler. Defaults to False.

        Examples:
            >>> from aiodistbus import AEntryPoint
            >>> ep = AEntryPoint()
            >>> async def test_handler():
            ...     print("Hello World")
            ...
            >>> await ep.on("test", test_handler)

        """
        # Track handlers (supporting wildcards)
        if "*" not in event_type:
            wrapped_func = self._wrapper(func, create_task=create_task)
            handler = Handler(event_type, wrapped_func, dtype)
            self._handlers[event_type] = handler
        else:
            wrapped_func = self._wrapper(func, unpack=False, create_task=create_task)
            handler = Handler(event_type, wrapped_func, dtype)
            self._wildcards[event_type] = handler

        await self._update_handlers(event_type)

    async def off(self, event_type: str):
        """Remove a handler

        Args:
            event_type (str): Event type

        """
        # Track handlers (supporting wildcards)
        if "*" not in event_type:
            del self._handlers[event_type]
        else:
            del self._wildcards[event_type]

        await self._update_handlers(event_type, remove=True)

    @abstractmethod
    async def emit(
        self, event_type: str, data: Any, id: Optional[str] = None
    ) -> Optional[Event]:
        """Emit an event

        Args:
            event_type (str): Event type
            data (Any): Data to send
            id (Optional[str], optional): Event ID. Defaults to None.

        Returns:
            Optional[Event]: Event object

        """
        ...

    @abstractmethod
    async def close(self):
        """Close the entrypoint"""
        ...

1	import asyncio	1✔
2	import logging	1✔
3	import uuid	1✔
4	from abc import ABC, abstractmethod	1✔
5	from collections import deque	1✔
6	from typing import Any, Callable, Coroutine, Dict, List, Optional, Type	1✔
7
8	from ..protocols import Event, Handler	1✔
9	from ..registry import Registry	1✔
10	from ..utils import safe_coro	1✔
11
12	logger = logging.getLogger("aiodistbus")	1✔
13
14
15	class AEntryPoint(ABC):	1✔
16	def __init__(self):	1✔
17	"""Abstract entrypoint for eventbus"""
18
19	# State information
20	self.id = str(uuid.uuid4())	1✔
21	self._handlers: Dict[str, Handler] = {}	1✔
22	self._wildcards: Dict[str, Handler] = {}	1✔
23	self._received: deque[str] = deque(maxlen=1000)	1✔
24	self._tasks: List[asyncio.Task] = []	1✔
25
26	def _wrapper(	1✔
27	self, func: Callable, unpack: bool = True, create_task: bool = False
28	) -> Callable:
29	"""Wrapper for handlers
30
31	The wrapper provides an approach for logging which events have been transmitted
32
33	Args:
34	func (Callable): Function to wrap
35	unpack (bool, optional): Unpack the event. Defaults to True.
36	create_task (bool, optional): Create a task for the handler. Defaults to False.
37
38	Returns:
39	Callable: Wrapped function
40
41	"""
42	# Wrapper for async functions
43	async def awrapper(event: Event):	1✔
44	coro: Optional[Coroutine] = None	1✔
45	if unpack:	1✔
46	if (	1✔
47	type(event.data) is not type(None)
48	and self._handlers[event.type].dtype
49	):
50	coro = func(event.data)	1✔
51	else:
52	coro = func()	1✔
53	else:
54	coro = func(event)	1✔
55
56	if coro:	1!
57	# Create safe coro with error msg
58	scoro = safe_coro(	1✔
59	coro, f"Error in (type: {event.type}, handler {func.__name__})"
60	)
61	if create_task:	1✔
62	self._tasks.append(asyncio.create_task(scoro))	1✔
63	else:
64	await scoro	1✔
65
66	self._received.append(event.id)	1✔
67
68	# Wrapper for sync functions
69	async def wrapper(event: Event):	1✔
70
71	try:	1✔
72	if unpack:	1!
73	func(event.data)	1✔
74	else:
75	func(event)	×
76	except Exception as e:	1✔
77	logger.error(	1✔
78	f"Error in handler (type: {event.type}, handler {func.__name__}): {e}"
79	)
80
81	self._received.append(event.id)	1✔
82
83	# Select according to function type
84	if asyncio.iscoroutinefunction(func):	1✔
85	return awrapper	1✔
86	else:
87	return wrapper	1✔
88
89	@abstractmethod	1✔
90	async def _update_handlers(	1✔
91	self, event_type: Optional[str] = None, remove: bool = False
92	):
UNCOV 93	...	×
94
95	####################################################################
96	## Public API
97	####################################################################
98
99	async def use(self, registry: Registry, namespace: str = "default"):	1✔
100	"""Use a registry
101
102	Args:
103	registry (Registry): Registry to use
104	namespace (str, optional): Namespace. Defaults to "default".
105
106	"""
107	# Obtain the handlers
108	for event_type, handler in registry.get_handlers(namespace).items():	1✔
109	await self.on(event_type, handler.function, handler.dtype)	1✔
110
111	async def on(	1✔
112	self,
113	event_type: str,
114	func: Callable,
115	dtype: Optional[Type] = None,
116	create_task: bool = False,
117	):
118	"""Register a handler
119
120	Args:
121	event_type (str): Event type
122	func (Callable): Function to call
123	dtype (Optional[Type], optional): Data type. Defaults to None.
124	create_task (bool, optional): Create a task for the handler. Defaults to False.
125
126	Examples:
127	>>> from aiodistbus import AEntryPoint
128	>>> ep = AEntryPoint()
129	>>> async def test_handler():
130	... print("Hello World")
131	...
132	>>> await ep.on("test", test_handler)
133
134	"""
135	# Track handlers (supporting wildcards)
136	if "*" not in event_type:	1✔
137	wrapped_func = self._wrapper(func, create_task=create_task)	1✔
138	handler = Handler(event_type, wrapped_func, dtype)	1✔
139	self._handlers[event_type] = handler	1✔
140	else:
141	wrapped_func = self._wrapper(func, unpack=False, create_task=create_task)	1✔
142	handler = Handler(event_type, wrapped_func, dtype)	1✔
143	self._wildcards[event_type] = handler	1✔
144
145	await self._update_handlers(event_type)	1✔
146
147	async def off(self, event_type: str):	1✔
148	"""Remove a handler
149
150	Args:
151	event_type (str): Event type
152
153	"""
154	# Track handlers (supporting wildcards)
155	if "*" not in event_type:	1!
156	del self._handlers[event_type]	1✔
157	else:
NEW 158	del self._wildcards[event_type]	×
159
160	await self._update_handlers(event_type, remove=True)	1✔
161
162	@abstractmethod	1✔
163	async def emit(	1✔
164	self, event_type: str, data: Any, id: Optional[str] = None
165	) -> Optional[Event]:
166	"""Emit an event
167
168	Args:
169	event_type (str): Event type
170	data (Any): Data to send
171	id (Optional[str], optional): Event ID. Defaults to None.
172
173	Returns:
174	Optional[Event]: Event object
175
176	"""
177	...	×
178
179	@abstractmethod	1✔
180	async def close(self):	1✔
181	"""Close the entrypoint"""
182	...	×

ChimeraPy / aiodistbus / 6724041435

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous