add router builder class, fix interupts for ListenerServer
This commit is contained in:
parent
8fb91b52c0
commit
099ebad566
@ -116,6 +116,29 @@ pages:
|
||||
|
||||
Nice syntax cheatsheet [here][4]
|
||||
|
||||
General docstring structure should be structured as follows:
|
||||
|
||||
```python
|
||||
def example_function(a, b):
|
||||
'''
|
||||
Minimal function description. (either first sentence or line; gets used in
|
||||
autosummaries)
|
||||
|
||||
Additional exposition, unwrapped by admonitions.
|
||||
|
||||
.. admonition:: Admonition description
|
||||
Indented content, code blocks, lists, etc
|
||||
|
||||
Parameters:
|
||||
a: a's description
|
||||
b: b's description
|
||||
|
||||
Returns:
|
||||
<return-type>: Description of return value
|
||||
'''
|
||||
...
|
||||
```
|
||||
|
||||
|
||||
[1]: https://pradyunsg.me/furo/
|
||||
[2]: https://www.sphinx-doc.org/en/master/man/sphinx-apidoc.html
|
||||
|
||||
@ -1,9 +1,9 @@
|
||||
from execlog import util
|
||||
from execlog import routers
|
||||
from execlog import listeners
|
||||
|
||||
from execlog.server import Server
|
||||
from execlog.handler import Handler
|
||||
from execlog.listener import Listener
|
||||
from execlog.router import Router, ChainRouter, Event
|
||||
from execlog.server import Server
|
||||
from execlog.event import Event, FileEvent
|
||||
|
||||
from execlog import listeners
|
||||
from execlog import routers
|
||||
from execlog import util
|
||||
from execlog.router import Router, ChainRouter, Event, RouterBuilder, route
|
||||
|
||||
@ -11,4 +11,5 @@ FileEvent = namedtuple(
|
||||
'FileEvent',
|
||||
['endpoint', 'name', 'action'],
|
||||
defaults=[None, None, None],
|
||||
# action is 32bit flag mask
|
||||
)
|
||||
|
||||
@ -5,9 +5,11 @@ import logging
|
||||
from pathlib import Path
|
||||
from collections import defaultdict
|
||||
|
||||
from colorama import Fore, Back, Style
|
||||
from inotify_simple import INotify, Event as iEvent, flags as iflags, masks as imasks
|
||||
|
||||
from execlog import util
|
||||
from execlog.util.generic import color_text
|
||||
from execlog.event import FileEvent
|
||||
from execlog.listener import Listener
|
||||
|
||||
@ -119,7 +121,8 @@ class PathListener(Listener[FileEvent]):
|
||||
flags: inotify_simple flags matching FS event types allowed to trigger the
|
||||
callback
|
||||
'''
|
||||
path = Path(path)
|
||||
#path = Path(path)
|
||||
path = str(path)
|
||||
|
||||
if flags is None:
|
||||
flags = iflags.CREATE | iflags.DELETE | iflags.MODIFY | iflags.DELETE_SELF | iflags.MOVED_TO
|
||||
@ -145,13 +148,23 @@ class PathListener(Listener[FileEvent]):
|
||||
to make sure to call ``.stop()``
|
||||
'''
|
||||
self.started = True
|
||||
logger.info(f'Starting listener for {len(self.watchmap)} paths')
|
||||
logger.info(
|
||||
color_text(
|
||||
f'Starting listener for {len(self.watchmap)} paths',
|
||||
Fore.GREEN,
|
||||
)
|
||||
)
|
||||
|
||||
for path, flags in self.canonmap.values():
|
||||
logger.info(f'> Listening on path {path} for flags {iflags.from_mask(flags)}')
|
||||
|
||||
for (callback, pattern, debounce, delay, *_) in self.router.routemap[path]:
|
||||
logger.info(f'| > {pattern} -> {callback.__name__} (debounce {debounce}ms, delay {delay}ms)')
|
||||
logger.info(
|
||||
color_text(
|
||||
f'| > {pattern} -> {callback.__name__} (debounce {debounce}ms, delay {delay}ms)',
|
||||
Style.DIM,
|
||||
)
|
||||
)
|
||||
|
||||
while True:
|
||||
rlist, _, _ = select.select(
|
||||
@ -300,7 +313,7 @@ class PathListener(Listener[FileEvent]):
|
||||
'''
|
||||
Note:
|
||||
If ``handle_events`` is called externally, note that this loop will block in the
|
||||
calling thread until the jobs have been submitted. It will _not_ block until
|
||||
calling thread until the jobs have been submitted. It will *not* block until
|
||||
jobs have completed, however, as a list of futures is returned. The calling
|
||||
Watcher instance may have already been started, in which case ``run()`` will
|
||||
already be executing in a separate thread. Calling this method externally will
|
||||
|
||||
@ -8,14 +8,15 @@ import inspect
|
||||
import traceback
|
||||
import threading
|
||||
from pathlib import Path
|
||||
from typing import Callable
|
||||
from functools import partial
|
||||
from typing import Any, Callable
|
||||
from colorama import Fore, Style
|
||||
from collections import defaultdict
|
||||
from functools import partial, update_wrapper
|
||||
from concurrent.futures import ThreadPoolExecutor, wait, as_completed
|
||||
|
||||
from tqdm.auto import tqdm
|
||||
|
||||
from execlog.util.generic import color_text
|
||||
from execlog.event import Event
|
||||
from execlog.listener import Listener
|
||||
|
||||
@ -156,7 +157,7 @@ class Router[E: Event]:
|
||||
route_tuple = (callback, pattern, debounce, delay, listener_kwargs)
|
||||
self.routemap[endpoint].append(route_tuple)
|
||||
|
||||
def submit(self, events:E | list[E], callbacks:list[Callable]|None=None):
|
||||
def submit(self, events: E | list[E], callbacks: list[Callable] | None = None):
|
||||
'''
|
||||
Handle a list of events. Each event is matched against the registered callbacks,
|
||||
and those callbacks are ran concurrently (be it via a thread pool or an asyncio
|
||||
@ -173,7 +174,7 @@ class Router[E: Event]:
|
||||
|
||||
return futures
|
||||
|
||||
def submit_event(self, event: E, callbacks:list[Callable]|None=None):
|
||||
def submit_event(self, event: E, callbacks: list[Callable] | None = None):
|
||||
'''
|
||||
Group up and submit all matching callbacks for ``event``. All callbacks are ran
|
||||
concurrently in their own threads, and this method blocks until all are completed.
|
||||
@ -294,6 +295,15 @@ class Router[E: Event]:
|
||||
# reject event
|
||||
continue
|
||||
|
||||
callback_name = str(callback)
|
||||
if hasattr(callback, '__name__'):
|
||||
callback_name = callback.__name__
|
||||
|
||||
name_text = color_text(name, Fore.BLUE)
|
||||
pattern_text = color_text(pattern, Fore.BLUE)
|
||||
endpoint_text = color_text(endpoint, Fore.BLUE)
|
||||
callback_text = color_text(callback_name[:50], Fore.BLUE)
|
||||
|
||||
if self.filter(event, pattern, **listen_kwargs):
|
||||
# note that delayed callbacks are added
|
||||
matches.append(self.get_delayed_callback(callback, delay, index))
|
||||
@ -301,19 +311,14 @@ class Router[E: Event]:
|
||||
# set next debounce
|
||||
self.next_allowed_time[index] = event_time + debounce
|
||||
|
||||
match_text = Style.BRIGHT + Fore.GREEN + 'matched' + Fore.RESET
|
||||
|
||||
callback_name = str(callback)
|
||||
if hasattr(callback, '__name__'):
|
||||
callback_name = callback.__name__
|
||||
|
||||
match_text = color_text('matched', Style.BRIGHT, Fore.GREEN)
|
||||
logger.info(
|
||||
f'Event [{name}] {match_text} [{pattern}] under [{endpoint}] for [{callback_name}]'
|
||||
f'Event [{name_text}] {match_text} [{pattern_text}] under [{endpoint_text}] for [{callback_text}]'
|
||||
)
|
||||
else:
|
||||
match_text = Style.BRIGHT + Fore.RED + 'rejected' + Fore.RESET
|
||||
match_text = color_text('rejected', Style.BRIGHT, Fore.RED)
|
||||
logger.debug(
|
||||
f'Event [{name}] {match_text} against [{pattern}] under [{endpoint}] for [{callback.__name__}]'
|
||||
f'Event [{name_text}] {match_text} against [{pattern_text}] under [{endpoint_text}] for [{callback_text}]'
|
||||
)
|
||||
|
||||
return matches
|
||||
@ -372,7 +377,7 @@ class Router[E: Event]:
|
||||
|
||||
def filter(self, event: E, pattern, **listen_kwargs) -> bool:
|
||||
'''
|
||||
Determine if a given event matches the providedpattern
|
||||
Determine if a given event matches the provided pattern
|
||||
|
||||
Parameters:
|
||||
event:
|
||||
@ -546,3 +551,141 @@ def handle_exception(future):
|
||||
except Exception as e:
|
||||
print(f"Exception occurred: {e}")
|
||||
traceback.print_exc()
|
||||
|
||||
|
||||
# RouterBuilder
|
||||
def route(router, route_group, **route_kwargs):
|
||||
def decorator(f):
|
||||
f._route_data = (router, route_group, route_kwargs)
|
||||
return f
|
||||
|
||||
return decorator
|
||||
|
||||
class RouteRegistryMeta(type):
|
||||
'''
|
||||
Metaclass handling route registry at the class level.
|
||||
'''
|
||||
def __new__(cls, name, bases, attrs):
|
||||
route_registry = defaultdict(lambda: defaultdict(list))
|
||||
|
||||
def register_route(method):
|
||||
nonlocal route_registry
|
||||
|
||||
if hasattr(method, '_route_data'):
|
||||
router, route_group, route_kwargs = method._route_data
|
||||
route_registry[router][route_group].append((method, route_kwargs))
|
||||
|
||||
# add registered superclass methods; iterate over bases (usually just one), then
|
||||
# that base's chain down (reversed), then methods from each subclass
|
||||
for base in bases:
|
||||
for _class in reversed(base.mro()):
|
||||
methods = inspect.getmembers(_class, predicate=inspect.isfunction)
|
||||
for _, method in methods:
|
||||
register_route(method)
|
||||
|
||||
# add final registered formats for the current class, overwriting any found in
|
||||
# superclass chain
|
||||
for attr_name, attr_value in attrs.items():
|
||||
register_route(attr_value)
|
||||
|
||||
attrs['route_registry'] = route_registry
|
||||
|
||||
return super().__new__(cls, name, bases, attrs)
|
||||
|
||||
class RouterBuilder(metaclass=RouteRegistryMeta):
|
||||
'''
|
||||
Builds a (Chain)Router using attached methods and passed options.
|
||||
|
||||
This class can be subtyped and desired router methods attached using the provided
|
||||
``route`` decorator. This facilitates two separate grouping mechanisms:
|
||||
|
||||
1. Group methods by frame (i.e., attach to the same router in a chain router)
|
||||
2. Group by registry equivalence (i.e, within a frame, registered with the same
|
||||
parameters)
|
||||
|
||||
These groups are indicated by the following collation syntax:
|
||||
|
||||
.. code-block:: python
|
||||
|
||||
@route('<router>/<frame>', '<route-group>', **route_kwargs)
|
||||
def method(...):
|
||||
...
|
||||
|
||||
and the following is a specific example:
|
||||
|
||||
.. code-block:: python
|
||||
|
||||
@route(router='convert', route_group='file', debounce=500)
|
||||
def file_convert_1(self, event):
|
||||
...
|
||||
|
||||
which will attach the method to the "convert" router (or "frame" in a chain router
|
||||
context) using parameters (endpoint, pattern, and other keyword args) associated with
|
||||
the "file" route group (as indexed by the ``register_map`` provided on instantiation)
|
||||
with the ``debounce`` route keyword (which will override the same keyword values if
|
||||
set in the route group). Note that the exact same ``@route`` signature can be used for
|
||||
an arbitrary number of methods to be handled in parallel by the associated Router.
|
||||
|
||||
Note that there is one reserved route group keyword: "post," for post callbacks.
|
||||
Multiple post-callbacks for a particular router can be specified with the same ID
|
||||
syntax above.
|
||||
|
||||
.. admonition:: Map structures
|
||||
|
||||
The following is a more intuitive breakdown of the maps involved, provided and
|
||||
computed on instantiation:
|
||||
|
||||
.. code-block:: python
|
||||
|
||||
# provided
|
||||
register_map[<router-name>] -> ( Router, { <type>: ( ( endpoint, pattern ), **kwargs ) } )
|
||||
|
||||
# computed
|
||||
routers[<router-name>][<type>] -> [... <methods> ...]
|
||||
|
||||
.. admonition:: TODO
|
||||
|
||||
Consider "flattening" the ``register_map`` to be indexed only by ``<type>``,
|
||||
effectively forcing the 2nd grouping mechanism to be provided here (while the 1st
|
||||
is handled by the method registration within the body of the class). This properly
|
||||
separates the group mechanisms and is a bit more elegant, but reduces the
|
||||
flexibility a bit (possibly in a good way, though).
|
||||
'''
|
||||
def __init__(
|
||||
self,
|
||||
register_map: dict[str, tuple[Router, dict[str, tuple[tuple[str, str], dict[str, Any]]]]],
|
||||
):
|
||||
self.register_map = register_map
|
||||
|
||||
# register
|
||||
for router_name, (router, router_options) in self.register_map.items():
|
||||
for route_group, method_arg_list in self.route_registry[router_name].items():
|
||||
# get post-callbacks for reserved key "post"
|
||||
# assumed no kwargs for passthrough
|
||||
if route_group == 'post':
|
||||
for method, _ in method_arg_list:
|
||||
router.add_post_callback(method)
|
||||
continue
|
||||
|
||||
group_options = router_options.get(route_group)
|
||||
if group_options is None:
|
||||
continue
|
||||
|
||||
# "group_route_kwargs" are route kwargs provided @ group level
|
||||
# "method_route_kwargs" are route kwargs provided @ method level
|
||||
# |-> considered more specific and will override group kwargs
|
||||
(endpoint, pattern), group_route_kwargs = group_options
|
||||
for method, method_route_kwargs in method_arg_list:
|
||||
router.register(
|
||||
endpoint,
|
||||
update_wrapper(partial(method, self), method),
|
||||
pattern,
|
||||
**{
|
||||
**group_route_kwargs,
|
||||
**method_route_kwargs
|
||||
}
|
||||
)
|
||||
|
||||
def get_router(self, router_key_list: list[str]):
|
||||
return ChainRouter([self.register_map[k][0] for k in router_key_list])
|
||||
|
||||
|
||||
@ -7,6 +7,7 @@ managed independently, but a single Server instance can house, start, and shutdo
|
||||
listeners in one place.
|
||||
|
||||
.. admonition:: todo
|
||||
|
||||
As it stands, the Server requires address and port details, effectively needing one
|
||||
of the HTTP items (static file serving or livereloading) to be initialized appropriately.
|
||||
But there is a clear use case for just managing disparate Routers and their associated
|
||||
@ -14,6 +15,7 @@ listeners in one place.
|
||||
make the Server definition more flexible.
|
||||
'''
|
||||
import re
|
||||
import signal
|
||||
import asyncio
|
||||
import logging
|
||||
import threading
|
||||
@ -290,3 +292,36 @@ class Server:
|
||||
|
||||
self.loop.call_soon_threadsafe(set_should_exit)
|
||||
|
||||
|
||||
class ListenerServer:
|
||||
'''
|
||||
Server abstraction to handle disparate listeners.
|
||||
'''
|
||||
def __init__(
|
||||
self,
|
||||
managed_listeners : list | None = None,
|
||||
):
|
||||
if managed_listeners is None:
|
||||
managed_listeners = []
|
||||
|
||||
self.managed_listeners = managed_listeners
|
||||
|
||||
def start(self):
|
||||
signal.signal(signal.SIGINT, lambda s,f: self.shutdown())
|
||||
signal.signal(signal.SIGTERM, lambda s,f: self.shutdown())
|
||||
|
||||
for listener in self.managed_listeners:
|
||||
#loop.run_in_executor(None, partial(self.listener.start, loop=loop))
|
||||
if not listener.started:
|
||||
listener.start()
|
||||
|
||||
for listener in self.managed_listeners:
|
||||
listener.join()
|
||||
|
||||
def shutdown(self):
|
||||
# stop attached auxiliary listeners, both internal & external
|
||||
if self.managed_listeners:
|
||||
logger.info(f"Stopping {len(self.managed_listeners)} listeners...")
|
||||
|
||||
for listener in self.managed_listeners:
|
||||
listener.stop()
|
||||
|
||||
@ -5,6 +5,10 @@ import colorama
|
||||
from colorama import Fore, Back, Style
|
||||
|
||||
|
||||
def color_text(text, *colorama_args):
|
||||
return f"{''.join(colorama_args)}{text}{Style.RESET_ALL}"
|
||||
|
||||
|
||||
class ColorFormatter(logging.Formatter):
|
||||
_format = '%(levelname)-8s :: %(name)s %(message)s'
|
||||
colorama.init(autoreset=True)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user