from logging import Logger
from multiprocessing.process import BaseProcess
from time import monotonic
from typing import Iterable, Callable, Any, Optional
from edri.dataclass.injection import Injection
__all__ = ["inject", "camel2snake", "snake2camel"]
# Shutdown grace windows (seconds). The component window must stay strictly
# larger than the worker window: a Manager reaps its own worker processes within
# WORKER_SHUTDOWN_GRACE, so the parent must wait longer than that before force-
# killing the Manager — otherwise it could SIGKILL a Manager mid-reap and orphan
# the worker it was about to stop.
WORKER_SHUTDOWN_GRACE = 5.0
COMPONENT_SHUTDOWN_GRACE = 15.0
def current_rss_bytes() -> Optional[int]:
"""
Resident set size (physical memory) of the current process, in bytes.
Reads ``/proc/self/statm`` (resident pages × page size), which is cheap and
dependency-free on Linux. Returns ``None`` when the file is unavailable
(other platforms) or unreadable, so callers can report "unknown" rather than
fail — this is best-effort telemetry, never a hard requirement.
:return: Resident memory in bytes, or ``None`` if it cannot be determined.
:rtype: Optional[int]
"""
try:
with open("/proc/self/statm") as statm:
resident_pages = int(statm.read().split()[1])
except (OSError, IndexError, ValueError):
return None
from os import sysconf
try:
page_size = sysconf("SC_PAGE_SIZE")
except (OSError, ValueError):
return None
return resident_pages * page_size
def peak_rss_bytes() -> Optional[int]:
"""
Peak resident memory (high-water mark) of the current process, in bytes.
Reads ``ru_maxrss`` from :func:`resource.getrusage`, which on Linux is
reported in KiB. Best-effort: returns ``None`` when ``resource`` is
unavailable (e.g. Windows) or the call fails.
:return: Peak resident memory in bytes, or ``None`` if it cannot be determined.
:rtype: Optional[int]
"""
try:
from resource import getrusage, RUSAGE_SELF
return int(getrusage(RUSAGE_SELF).ru_maxrss) * 1024
except (ImportError, OSError, ValueError):
return None
def cpu_seconds_total() -> float:
"""
Total CPU time (user + system) consumed by the current process, in seconds.
A monotonic counter; a caller that wants utilization takes the delta between
two reads and divides by wall-clock elapsed. Best-effort: returns ``0.0``
when ``resource`` is unavailable or the call fails.
:return: Cumulative CPU seconds, or ``0.0`` if it cannot be determined.
:rtype: float
"""
try:
from resource import getrusage, RUSAGE_SELF
usage = getrusage(RUSAGE_SELF)
return float(usage.ru_utime + usage.ru_stime)
except (ImportError, OSError, ValueError):
return 0.0
def open_fd_count() -> Optional[int]:
"""
Number of open file descriptors held by the current process.
Counts entries in ``/proc/self/fd`` (Linux), a cheap fd-leak detector.
Best-effort: returns ``None`` when the directory is unavailable or unreadable.
:return: Open file-descriptor count, or ``None`` if it cannot be determined.
:rtype: Optional[int]
"""
from os import listdir
try:
return len(listdir("/proc/self/fd"))
except OSError:
return None
[docs]
def camel2snake(title: str) -> str:
"""
Converts a camelCase or PascalCase string to a snake_case string.
:param title: The camelCase or PascalCase string to be converted.
:type title: str
:return: The converted snake_case string.
:rtype: str
:Examples:
>>> camel2snake("camelCaseExample")
'camel_case_example'
>>> camel2snake("CamelCaseExample")
'camel_case_example'
"""
new_title = ""
underscore = False
for letter in title:
if 65 <= ord(letter) <= 90:
if underscore:
new_title += '_'
underscore = False
new_title += chr(ord(letter) + 32)
else:
underscore = True
new_title += letter
return new_title
[docs]
def snake2camel(title: str, /, *, exceptions: Optional[tuple[str, ...]] = None) -> str:
"""
Converts a snake_case string to a camelCase string, with optional exceptions
where the conversion should not be applied.
:param title: The snake_case string to be converted.
:type title: str
:param exceptions: A tuple of strings that, when matched at the beginning of the title,
prevent conversion and return the title as is. Defaults to None.
:type exceptions: Optional[Tuple[str, ...]]
:return: The converted camelCase string if no exception is matched; otherwise, the original string.
:rtype: str
:raises KeyError: If a non-trailing URL parameter is missing and subsequent placeholders
are present.
:Examples:
>>> snake2camel("snake_case_example")
'snakeCaseExample'
>>> snake2camel("id_example", exceptions=("id_",))
'id_example'
"""
if exceptions is None:
exceptions = ("id_",)
if title.startswith(exceptions):
return title
title_list = title.split('_')
title_list = [title_list[0]] + [word.capitalize() for word in title_list[1:]]
return ''.join(title_list)
def partition[T](pred: Callable[[T], bool], iterable: Iterable[T]) -> tuple[list[T], list[T]]:
"""
Splits an iterable into two lists based on a predicate function.
:param pred: A function that takes an item and returns a boolean value.
Items for which this function returns True will go into the first list;
items for which it returns False will go into the second list.
:type pred: Callable[[T], bool]
:param iterable: The iterable to be partitioned.
:type iterable: Iterable[T]
:return: A tuple containing two lists:
- The first list contains all items from the iterable for which `pred` returned True.
- The second list contains all items from the iterable for which `pred` returned False.
:rtype: Tuple[List[T], List[T]]
:Examples:
>>> partition(lambda x: x % 2 == 0, [1, 2, 3, 4])
([2, 4], [1, 3])
"""
trues = []
falses = []
for item in iterable:
if pred(item):
trues.append(item)
else:
falses.append(item)
return trues, falses
def format_url(url: str, **kwargs: Any) -> str:
"""
Formats a URL string by replacing placeholders with keyword arguments. If a parameter is missing,
it is considered optional only if it appears at the end of the URL or if all subsequent placeholders
are also missing. Otherwise, raises a KeyError.
:param url: The URL template containing placeholders in the format `{placeholder}`.
:type url: str
:param kwargs: Keyword arguments representing the replacement values for the placeholders in the URL.
:type kwargs: Any
:return: The formatted URL with all placeholders replaced by their corresponding values.
If optional parameters are missing, trailing placeholders are removed.
:rtype: str
:raises KeyError: If a required URL parameter is missing, and it is not at the end of the URL,
or if other placeholders to the right are present.
:Examples:
>>> format_url('https://example.com/{user}/{id}', user='john', id=42)
'https://example.com/john/42'
>>> format_url('https://example.com/{user}/{id}', user='john')
'https://example.com/john'
>>> format_url('https://example.com/{user}/{id}/{page}', user='john')
'https://example.com/john'
>>> format_url('https://example.com/{user}/{id}/detail', user='john')
Traceback (most recent call last):
...
KeyError: 'Optional url parameter is missing: id'
>>> format_url('https://example.com/{user}/{id}/{page}', user='john', page='home')
Traceback (most recent call last):
...
KeyError: 'Optional url parameter is missing: id'
"""
class Helper(dict[str, Any]):
def __init__(self, **kwargs: Any) -> None:
super().__init__(kwargs)
self.missing: list[str] = []
def __missing__(self, key: str) -> str:
self.missing.append(key)
return f"{{{key}}}"
kwargs_helper = Helper(**kwargs)
formatted_url = url.format_map(kwargs_helper)
for missing in reversed(kwargs_helper.missing):
placeholder = f"{{{missing}}}"
if formatted_url.endswith(placeholder):
# Remove the trailing placeholder along with the preceding slash if present
if len(formatted_url) > len(placeholder) and formatted_url[-len(placeholder) - 1] == '/':
formatted_url = formatted_url[:-len(placeholder) - 1]
else:
formatted_url = formatted_url[:-len(placeholder)]
else:
raise KeyError(f"Optional url parameter is missing: {missing}")
return formatted_url
[docs]
def inject(*args: Callable[..., Any], **kwargs: Any) -> Injection:
"""
Creates an `Injection` for validator callables and their parameters.
This function is typically used to declaratively define dependency injections
by specifying a class and any keyword arguments to pass to its constructor.
Args:
*args: Validator classes or functions called in order.
**kwargs: Keyword arguments passed to validators whose signatures accept them.
Returns:
Injection[T]: An `Injection` object wrapping the class and its parameters.
Example:
inject(MyService, config=..., db=...)
Use ``| None`` to make the injection optional::
inject(StringValidator, minimum_length=2) | None
"""
return Injection(classes=args, parameters=kwargs)
def install_graceful_sigterm() -> None:
"""Map ``SIGTERM`` onto ``KeyboardInterrupt`` for the current process.
EDRI's process loops (the Router, Managers, the Broker) already exit
cleanly on ``KeyboardInterrupt`` — i.e. on ``SIGINT`` / Ctrl-C. Process
managers (systemd, Docker, Kubernetes) stop a process with ``SIGTERM`` by
default, which would otherwise bypass that shutdown. Installing this handler
makes ``SIGTERM`` follow the same graceful path.
Must be called from a process's main thread; on a non-main thread (where
``signal.signal`` is not allowed) it is a no-op.
"""
import signal
def _raise_keyboard_interrupt(signum: int, frame: Any) -> None:
raise KeyboardInterrupt
try:
signal.signal(signal.SIGTERM, _raise_keyboard_interrupt)
except ValueError:
# signal.signal() outside the main thread raises ValueError; nothing to do.
pass
def terminate_processes(processes: Iterable[BaseProcess], *, grace: float,
logger: Logger | None = None) -> None:
"""Stop child processes: ``terminate()`` (SIGTERM) all, wait one shared ``grace``
window, then ``kill()`` (SIGKILL) any survivors.
The wait uses a single shared deadline rather than a per-process timeout, so the
total time spent here is bounded by ``grace`` regardless of how many processes are
slow to stop. Callers can therefore size an outer grace window above this one and
rely on a child finishing its own reap before being force-killed.
"""
alive = [process for process in processes if process.is_alive()]
if not alive:
return
for process in alive:
process.terminate()
deadline = monotonic() + grace
for process in alive:
process.join(timeout=max(0.0, deadline - monotonic()))
survivors = [process for process in alive if process.is_alive()]
for process in survivors:
if logger is not None:
logger.warning("Force-killing process that did not stop: %s", process.name)
process.kill()
for process in survivors:
process.join(timeout=grace)