modules-and-libs/pyappengine
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Documentation added

Browse Source
This commit is contained in:
Foué
2026-04-12 12:03:17 +02:00
parent 5a60e47c12
commit 5d8865a9fa
6 changed files with 621 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
docs/api.md Normal file
View File

@@ -0,0 +1,25 @@
# API Reference
## AppEngine
::: appengine.AppEngine
---
## Commands
::: appengine.Commands
---
## CommandsLoader
::: appengine.CommandsLoader
---
## Errors
::: appengine.AEErrs
::: appengine.AppEngineException

103
docs/getting-started.md Normal file
View File

@@ -0,0 +1,103 @@
# Getting started
## Project structure
```
my_project/
├── main.py
├── config.ini
└── modules/
├── cmds_foo.py
└── cmds_bar.py
```
## Configuration file
```ini
[general]
default = foo # default module (used when no module is specified)
# methods_prefix = cmd_ # optional: override command method prefix
# modules_prefix = cmds_ # optional: override module file prefix
[foo]
alias = f # optional: short name for the module
my_setting = value # any key accessible via self.config.get("my_setting")
[bar]
# ...
```
## Writing a module
```python
from appengine import Commands, AppEngineException, AEErrs
class Foo(Commands):
# Set to True to run this module in a background thread
threaded = False
# Declare dependencies on other modules
# dependencies = ["bar"] # list form → self.bar = cmods["bar"]
# dependencies = {"b": "bar"} # dict form → self.b = cmods["bar"]
def cmd_add(self, a: float, b: float):
"""Add two numbers.
Args:
a: First operand.
b: Second operand.
Returns:
float: The sum of *a* and *b*.
"""
return float(a) + float(b)
def cmd_fail(self):
"""Demonstrate structured error reporting."""
raise AppEngineException(AEErrs.INVALID_PARAMS, "Nothing to do here")
def free(self):
# Close connections, files, etc.
pass
```
## Threaded module
```python
import time
from appengine import Commands
class Worker(Commands):
threaded = True # will be started as a Thread
def run(self):
while not self.stopped:
self.info("tick")
time.sleep(1)
def free(self):
self.info("worker cleaned up")
```
## Dispatching commands
Commands are dispatched via `execute_command(module, method, *args)`:
```python
success, result = my_commands_instance.execute_command("foo", "add", 1, 2)
```
### Built-in help
| Call | Result |
|------|--------|
| `execute_command("", "help")` | List of all modules |
| `execute_command("", "help", "foo")` | List of commands in `foo` |
| `execute_command("", "help", "foo.add")` | Docstring of `foo.add` |
## Error handling
Commands return `(success: bool, result)`.
On error, `result` is a tuple `(error_code: int, message: str)`.
Error codes follow the JSON-RPC convention (see `AEErrs`).

65
docs/index.md Normal file
View File

@@ -0,0 +1,65 @@
# pyappengine
A Python framework for building **modular, command-based applications** with dynamic module loading, thread-safe command dispatch, and structured lifecycle management.
## Features
- **Dynamic module loading** — drop a `cmds_*.py` file in your module directory, it's loaded automatically
- **Thread-safe command dispatch** — shared lock across all modules
- **Threaded modules** — opt-in background thread execution per module
- **Inter-module dependencies** — declarative dependency injection between modules
- **Built-in help system** — introspects docstrings at runtime
- **INI configuration** — per-module config sections
- **Graceful shutdown** — SIGINT handler + global stop event
## Installation
```bash
pip install pyappengine
```
## Quick start
**1. Create your application entry point:**
```python
from appengine import AppEngine
app = AppEngine(
"my_app",
conf_file="config.ini",
log_file="my_app.log",
debug=True,
)
app.exec(modpath="./modules")
```
**2. Create a module** — save as `modules/cmds_hello.py`:
```python
from appengine import Commands
class Hello(Commands):
def cmd_greet(self, name: str):
"""Greet someone.
Args:
name: The person to greet.
Returns:
str: A greeting string.
"""
return f"Hello, {name}!"
```
**3. Create a minimal config** — `config.ini`:
```ini
[general]
default = hello
```
## License
Released under the [CeCILL-C](https://cecill.info/licences/Licence_CeCILL-C_V1-en.html) license.
Author: François Dausseur — fdausseur@free.fr

51
mkdocs.yml Normal file
View File

@@ -0,0 +1,51 @@
site_name: pyappengine
site_description: Python Application Engine — modular command-based application framework
site_author: François Dausseur
repo_url: https://git.beafrancois.fr/Foue/pyappengine
repo_name: pyappengine
theme:
name: material
palette:
- scheme: default
primary: indigo
accent: indigo
toggle:
icon: material/brightness-7
name: Switch to dark mode
- scheme: slate
primary: indigo
accent: indigo
toggle:
icon: material/brightness-4
name: Switch to light mode
features:
- navigation.sections
- navigation.top
- search.suggest
- content.code.copy
plugins:
- search
- mkdocstrings:
handlers:
python:
options:
show_source: true
show_root_heading: true
docstring_style: google
members_order: source
nav:
- Home: index.md
- Getting started: getting-started.md
- API Reference: api.md
markdown_extensions:
- admonition
- pymdownx.highlight:
anchor_linenums: true
- pymdownx.superfences
- pymdownx.inlinehilite
- toc:
permalink: true

7
pyproject.toml
View File

@@ -18,6 +18,13 @@ classifiers = [
dependencies = [ ]
dynamic = ["version"]
[project.optional-dependencies]
docs = [
"mkdocs>=1.5",
"mkdocs-material>=9.0",
"mkdocstrings[python]>=0.24",
]
[project.urls]
"Homepage" = "https://git.beafrancois.fr/Foue/pyappengine"

377
src/appengine/__init__.py
View File

@@ -1,4 +1,18 @@
#!/usr/bin/python3
"""pyappengine — Python Application Engine.
A framework for building modular, command-based Python applications
with dynamic module loading, thread-safe command dispatch, and
structured lifecycle management.
Typical usage:
```python
from appengine import AppEngine
app = AppEngine("my_app", conf_file="config.ini", debug=True)
app.exec(modpath="./modules")
```
"""
import os
import sys
@@ -16,6 +30,14 @@ from threading import Thread, Event
def is_number(s):
"""Check whether a string represents a numeric value.
Args:
s (str): The string to test.
Returns:
bool: ``True`` if *s* can be parsed as a float, ``False`` otherwise.
"""
try:
float(s)
return True
@@ -25,6 +47,16 @@ def is_number(s):
class AEErrs(Enum):
"""JSON-RPC-inspired error codes used throughout the engine.
Attributes:
PARSE_ERROR: Invalid JSON received (-32700).
INVALID_REQUEST: Request is not a valid object (-32600).
METH_NOT_FOUND: Method does not exist or is not available (-32601).
INVALID_PARAMS: Invalid method parameters (-32602).
INTERNAL_ERROR: Generic internal error (-32000).
"""
PARSE_ERROR = -32700
INVALID_REQUEST = -32600
METH_NOT_FOUND = -32601
@@ -45,6 +77,26 @@ ERROR_MESSAGES = {
class AppEngineException(Exception):
"""Exception raised by command modules to signal a structured error.
Wraps an :class:`AEErrs` code so that the engine can return a
machine-readable ``(code, message)`` pair to the caller.
Args:
error (AEErrs): The error type.
mesg (str, optional): Custom message. Defaults to the standard
message associated with *error*.
Attributes:
value (int): Numeric error code (from :class:`AEErrs`).
mesg (str): Human-readable error message.
Example:
```python
raise AppEngineException(AEErrs.INVALID_PARAMS, "Expected an integer")
```
"""
def __init__(self, error: AEErrs, mesg=None) -> None:
if mesg is None:
self.mesg = str(error)
@@ -55,6 +107,58 @@ class AppEngineException(Exception):
class Commands(Thread):
"""Base class for all command modules.
Subclass ``Commands`` to create a module. Any method whose name starts
with :attr:`prefcmd` (default ``"cmd_"``) is automatically exposed as a
callable command and listed by the built-in ``help`` system.
A module can optionally run as a background thread by setting
``self.threaded = True`` in ``__init__``. When threaded, the module's
``run()`` method will be executed in a dedicated thread managed by
:class:`CommandsLoader`.
Inter-module dependencies can be declared via a ``dependencies``
attribute (see :meth:`CommandsLoader._load_dependencies`).
Class Attributes:
defmod (str | None): Name of the default module, set by
:class:`CommandsLoader` from the ``[general]`` config section.
prefcmd (str): Prefix for command methods. Defaults to ``"cmd_"``.
Attributes:
threaded (bool): If ``True``, the module runs as a thread.
Defaults to ``False``.
cmods (dict): Shared dictionary of all loaded modules
(``{name: Commands}``), injected by :class:`CommandsLoader`.
config (ConfigParser | None): Configuration section for this module.
log (logging.Logger): Logger instance shared across all modules.
nickname (str | None): Module alias read from ``config["alias"]``.
stopped (bool): Set to ``True`` by :meth:`stop`.
stop_all_event (threading.Event | None): Global stop event,
injected by :class:`CommandsLoader`.
Example:
```python
from appengine import Commands
class MyModule(Commands):
threaded = False
def cmd_greet(self, name: str):
\"\"\"Greet someone.
Args:
name: The person to greet.
Returns:
str: A greeting string.
\"\"\"
return f"Hello, {name}!"
```
Save the file as ``cmds_my_module.py`` in your ``modpath``.
"""
defmod = None
prefcmd = "cmd_"
@@ -71,6 +175,15 @@ class Commands(Thread):
self.nickname = self.config.get("alias")
def val_to_print(self, val, dig=1):
"""Format a value as a printable string.
Args:
val (str | float | Any): The value to format.
dig (int): Number of decimal places for float values. Defaults to 1.
Returns:
str: The formatted string representation of *val*.
"""
if isinstance(val, str):
return val
if isinstance(val, float):
@@ -80,39 +193,73 @@ class Commands(Thread):
return str(val)
def info(self, msg):
"""Log an informational message.
Args:
msg (str): Message to log.
"""
if self.log:
self.log.info(msg)
else:
print("info: " + msg)
def debug(self, msg):
"""Log a debug message.
Args:
msg (str): Message to log.
"""
if self.log:
self.log.debug(msg)
else:
print("debug: " + msg)
def warning(self, msg):
"""Log a warning message.
Args:
msg (str): Message to log.
"""
if self.log:
self.log.warning(msg)
else:
print("warning: " + msg)
def error(self, msg):
"""Log an error message.
Args:
msg (str): Message to log.
"""
if self.log:
self.log.error(msg)
else:
print("error: " + msg)
def stop(self):
"""Signal the module to stop.
Sets :attr:`stopped` to ``True``. Threaded subclasses should check
this flag in their ``run()`` loop to exit gracefully.
"""
self.stopped = True
def free(self):
""" Virtual method used to clean resources for all Commands
when the application is exited.
"""Release resources held by this module.
This is a no-op by default. Override it in subclasses to close
connections, files, or other resources when the application exits.
Called by :meth:`CommandsLoader.free` for every loaded module.
"""
pass
def list_modules(self):
"""Return a formatted list of all loaded modules.
Returns:
tuple[bool, str]: ``(True, listing)`` where *listing* is a
newline-separated string of module names.
"""
success = True
ret = "List of modules:\n"
for module in self.cmods.keys():
@@ -120,6 +267,16 @@ class Commands(Thread):
return success, ret.strip()
def help_module(self, module, *args):
"""Return the help text for a given module or one of its commands.
Args:
module (str): Name of the target module.
*args: If provided, the first element is the command name whose
docstring should be returned.
Returns:
str: Help text, or an error message if the module is not found.
"""
if module in self.cmods.keys():
if len(args) > 0:
return self.cmods[module].cmd_help(args[0])
@@ -129,6 +286,23 @@ class Commands(Thread):
return "No module with this name"
def _execute_command(self, method: str, *args, **kwargs) -> tuple:
"""Execute a command on this module in a thread-safe manner.
Looks up the method ``<prefcmd><method>`` on ``self``, acquires the
shared lock, calls it, and returns the result. Handles
:class:`AppEngineException` and bare exceptions gracefully.
Args:
method (str): Command name (without the ``cmd_`` prefix).
*args: Positional arguments forwarded to the command method.
**kwargs: Keyword arguments forwarded to the command method.
Returns:
tuple[bool, Any]: ``(success, result)`` where *success* is
``False`` if the method was not found or raised an error, and
*result* is either the command return value or an
``(error_code, message)`` tuple.
"""
success = False
ret = (AEErrs.INTERNAL_ERROR.value, "function not found")
if hasattr(self, self.prefcmd + method) and inspect.ismethod(
@@ -171,6 +345,24 @@ class Commands(Thread):
return success, ret
def execute_command(self, module: str, method: str, *args, **kwargs):
"""Dispatch a command to the appropriate module.
Handles the special ``help`` command (with no module prefix) and
resolves underscore/hyphen aliases for module names before
delegating to :meth:`_execute_command`.
Args:
module (str): Target module name. Pass an empty string ``""``
to use :attr:`defmod`, or for the built-in ``help`` routing.
method (str): Command name (without the ``cmd_`` prefix).
*args: Positional arguments forwarded to the command.
**kwargs: Keyword arguments forwarded to the command.
Returns:
tuple[bool, Any]: ``(success, result)`` from
:meth:`_execute_command`, or ``(False, (error_code, message))``
if the module is not found.
"""
# isolate the module called
success = False
@@ -211,14 +403,26 @@ class Commands(Thread):
return self.cmods[module]._execute_command(method, *args, **kwargs)
def stop_all(self):
"""Trigger a global application stop.
Sets the shared :attr:`stop_all_event`, which signals
:class:`AppEngine` to initiate an orderly shutdown of all modules.
"""
if self.stop_all_event is not None:
self.stop_all_event.set()
def cmd_help(self, *args, **kwargs):
"""Help of module commands.
Params:
- if No param: list of commands
- otherwise: help of the first arg"""
"""List available commands or show the docstring of a specific one.
Args:
*args: If empty, all command names are listed. If one argument
is given, it is treated as a command name and its docstring
is returned.
Returns:
str: A newline-separated list of command names, or the
docstring of the requested command.
"""
ret = ""
@@ -245,6 +449,30 @@ class Commands(Thread):
class CommandsLoader:
"""Discover, load, and manage the lifecycle of command modules.
Scans *modpath* for Python files whose names match
``<modules_prefix><name>.py`` (default prefix: ``"cmds_"``),
imports each one, and instantiates the first :class:`Commands`
subclass found.
All modules share the same :class:`threading.Lock` and
:class:`threading.Event` (stop signal) injected by :class:`AppEngine`.
Args:
stop_event (threading.Event): Event used to signal a global stop.
config (ConfigParser): Full application configuration.
log (logging.Handler): Logger shared across all modules.
modpath (str): Directory to scan for module files.
Note:
The ``[general]`` config section may override engine defaults:
- ``default``: name of the default module.
- ``methods_prefix``: prefix for command methods (default ``"cmd_"``).
- ``modules_prefix``: prefix for module files (default ``"cmds_"``).
"""
def __init__(self, stop_event: Event, config: ConfigParser, log: logging.Handler, modpath: str):
self.config = config
self.modpath = modpath
@@ -264,6 +492,12 @@ class CommandsLoader:
self._set_cmods()
def _load_commands(self):
"""Scan *modpath* and load all matching module files.
Populates :attr:`cmods` with successfully instantiated modules.
Each module is keyed by its ``nickname`` (from config) or by its
filename stem stripped of the module prefix.
"""
cmds = {}
with os.scandir(self.modpath) as it:
for entry in it:
@@ -288,6 +522,19 @@ class CommandsLoader:
self.cmods = cmds
def _load_module(self, name: str):
"""Import a module file and instantiate its :class:`Commands` subclass.
Args:
name (str): Module filename stem (without ``.py``).
Returns:
Commands | None: The instantiated module object, or ``None`` if
no valid :class:`Commands` subclass could be instantiated.
Raises:
Exception: If the Python file cannot be imported at all
(e.g. missing dependency).
"""
try:
module = import_module(name)
except ModuleNotFoundError:
@@ -314,10 +561,30 @@ class CommandsLoader:
return obj
def _set_cmods(self):
"""Inject the shared ``cmods`` dict into every loaded module.
This allows any module to call commands on any other module via
:meth:`Commands.execute_command`.
"""
for k, v in self.cmods.items():
v.cmods = self.cmods
def _load_dependencies(self):
"""Resolve and inject inter-module dependencies.
For each module that declares a ``dependencies`` attribute, the
corresponding module objects are looked up in :attr:`cmods` and set
as attributes on the dependent module.
``dependencies`` can be:
- A **list** of module names: each name becomes both the attribute
name and the key in ``cmods``.
- A **dict** ``{attr_name: module_name}``: allows renaming the
injected attribute.
Logs an error for any dependency that cannot be satisfied.
"""
for k, v in self.cmods.items():
if hasattr(v, "dependencies"):
deps = v.dependencies
@@ -337,27 +604,75 @@ class CommandsLoader:
)
def start(self):
"""Resolve dependencies and start all threaded modules.
Must be called before :meth:`join`. Non-threaded modules are not
started here — they are invoked on demand via
:meth:`Commands.execute_command`.
"""
self._load_dependencies()
for k, v in self.cmods.items():
if v.threaded:
v.start()
def stop(self):
"""Signal all threaded modules to stop.
Calls :meth:`Commands.stop` on each module whose ``threaded``
attribute is ``True``.
"""
for k, v in self.cmods.items():
if v.threaded:
v.stop()
def join(self):
"""Wait for all threaded modules to finish.
Blocks until every module started as a thread has exited its
``run()`` method.
"""
for k, v in self.cmods.items():
if v.threaded:
v.join()
def free(self):
"""Release resources for all loaded modules.
Calls :meth:`Commands.free` on every module regardless of whether
it is threaded.
"""
for k, v in self.cmods.items():
v.free()
class AppEngine:
"""Main entry point for a pyappengine application.
Handles configuration loading, logging setup, signal handling, and
the full module lifecycle (load → start → run → stop → free).
Args:
app_name (str): Application name, also used as the logger name.
conf_file (str): Path to an INI configuration file. Optional.
log_file (str): Path to the log file. If empty, logs go to stderr.
Relative paths are resolved from the current working directory.
debug (bool): If ``True``, sets the log level to ``DEBUG``.
Defaults to ``False`` (``WARNING`` level).
Example:
```python
from appengine import AppEngine
app = AppEngine(
"my_app",
conf_file="config.ini",
log_file="my_app.log",
debug=False,
)
app.exec(modpath="./modules")
```
"""
def __init__(
self,
app_name: str,
@@ -382,16 +697,41 @@ class AppEngine:
self.stop_thread.start()
def signal_handler(self, sig, frame):
"""Handle SIGINT (Ctrl+C) by initiating a clean shutdown.
Args:
sig: Signal number (unused).
frame: Current stack frame (unused).
"""
print("\nExiting.")
self.stop()
def parse_config(self, cf: str):
"""Load an INI configuration file.
Args:
cf (str): Path to the configuration file.
Raises:
Exception: If the file does not exist or is not a regular file.
"""
if os.path.exists(cf) and os.path.isfile(cf):
self.conf.read(cf)
else:
raise Exception("Configuration file not found")
def def_log(self, log_file: str):
"""Configure the application logger.
Creates a :class:`logging.Logger` named after :attr:`app_name`.
If *log_file* is provided and the parent directory is writable,
a :class:`logging.FileHandler` is added. Otherwise logs go to
the default stream handler (stderr).
Args:
log_file (str): Destination log file path. Empty string disables
file logging.
"""
self.log = logging.getLogger(self.app_name)
if log_file != "":
fname = log_file
@@ -409,14 +749,37 @@ class AppEngine:
self.log.error('No write permissions: "{}"'.format(fname))
def exec(self, modpath: str = ""):
"""Load modules and run the application until stopped.
Creates a :class:`CommandsLoader`, starts all threaded modules,
then blocks until they all finish. Calls :meth:`CommandsLoader.free`
on exit.
Args:
modpath (str): Directory containing the module files
(``cmds_*.py``).
"""
self.cl = CommandsLoader(self.stop_event, self.conf, self.log, modpath)
self.cl.start()
self.cl.join()
self.cl.free()
def wait_stop(self, evnt):
"""Wait for the stop event and then halt all modules.
Runs in a dedicated background thread started in ``__init__``.
Blocks on *evnt* and calls :meth:`CommandsLoader.stop` once it fires.
Args:
evnt (threading.Event): The event to wait on.
"""
evnt.wait()
self.cl.stop()
def stop(self):
self.stop_event.set()
"""Stop the application.
Sets the internal stop event, which unblocks :meth:`wait_stop`
and triggers an orderly shutdown of all modules.
"""
self.stop_event.set()