olog/execlib
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

add router builder class, fix interupts for ListenerServer

Browse Source
This commit is contained in:
Sam G. 2024-05-09 00:39:42 -07:00
parent 22f24ab73f
commit 3b6a09bb72
7 changed files with 243 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
docs/reference/documentation/sphinx.md
View File

@ -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

12
execlog/__init__.py
View File

@ -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

1
execlog/event.py
View File

@ -11,4 +11,5 @@ FileEvent = namedtuple(
'FileEvent',
['endpoint', 'name', 'action'],
defaults=[None, None, None],
# action is 32bit flag mask
)

21
execlog/listeners/path.py
View File

@ -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

171
execlog/router.py
View File

@ -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])

35
execlog/server.py
View File

@ -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()

4
execlog/util/generic.py
View File

@ -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)