toolboxv2 API Reference

This section provides an API reference for key components directly available from the toolboxv2 package.

Core Application & Tooling

toolboxv2.AppType

Source code in toolboxv2/utils/system/types.py

class AppType:
    prefix: str
    id: str
    globals: dict[str, Any] = {"root": dict, }
    locals: dict[str, Any] = {"user": {'app': "self"}, }

    manifest: TBManifest

    local_test: bool = False
    start_dir: str
    data_dir: str
    config_dir: str
    info_dir: str
    appdata: str
    is_server:bool = False

    logger: logging.Logger
    logging_filename: str
    audit_logger: AuditLogger

    api_allowed_mods_list: list[str] = []

    version: str
    loop: asyncio.AbstractEventLoop

    keys: dict[str, str] = {
        "MACRO": "macro~~~~:",
        "MACRO_C": "m_color~~:",
        "HELPER": "helper~~~:",
        "debug": "debug~~~~:",
        "id": "name-spa~:",
        "st-load": "mute~load:",
        "comm-his": "comm-his~:",
        "develop-mode": "dev~mode~:",
        "provider::": "provider::",
    }

    defaults: dict[
        str,
        (bool or dict or dict[str, dict[str, str]] or str or list[str] or list[list])
        | None,
    ] = {
        "MACRO": list[str],
        "MACRO_C": dict,
        "HELPER": dict,
        "debug": str,
        "id": str,
        "st-load": False,
        "comm-his": list[list],
        "develop-mode": bool,
    }

    root_blob_storage: BlobStorage
    config_fh: FileHandler
    _debug: bool
    flows: dict[str, Callable]
    dev_modi: bool
    functions: dict[str, Any]
    modules: dict[str, Any]

    interface_type: ToolBoxInterfaces
    REFIX: str
    logger_prefix:str

    alive: bool
    called_exit: tuple[bool, float]
    args_sto: AppArgs
    system_flag = None
    session: 'AioSession' = None
    appdata = None
    exit_tasks = []

    enable_profiling: bool = False
    sto = None

    websocket_handlers: dict[str, dict[str, Callable]] = {}
    _rust_ws_bridge: Any = None

    ip: str | None
    location: str | None
    ping: int | None
    _ping_task: Callable
    _ping_interval: int | None


    def __init__(self, prefix=None, args=None):
        self.args_sto = args
        self.prefix = prefix
        self._footprint_start_time = time.time()
        if psutil:
            self._process = psutil.Process(os.getpid())

        # Tracking-Daten für Min/Max/Avg
        self._footprint_metrics = {
            'memory': {'max': 0, 'min': float('inf'), 'samples': []},
            'cpu': {'max': 0, 'min': float('inf'), 'samples': []},
            'disk_read': {'max': 0, 'min': float('inf'), 'samples': []},
            'disk_write': {'max': 0, 'min': float('inf'), 'samples': []},
            'network_sent': {'max': 0, 'min': float('inf'), 'samples': []},
            'network_recv': {'max': 0, 'min': float('inf'), 'samples': []},
        }

        # Initial Disk/Network Counters
        try:
            io_counters = self._process.io_counters()
            self._initial_disk_read = io_counters.read_bytes
            self._initial_disk_write = io_counters.write_bytes
        except (AttributeError, OSError):
            self._initial_disk_read = 0
            self._initial_disk_write = 0

        try:
            net_io = psutil.net_io_counters()
            self._initial_network_sent = net_io.bytes_sent
            self._initial_network_recv = net_io.bytes_recv
        except (AttributeError, OSError):
            self._initial_network_sent = 0
            self._initial_network_recv = 0

        # State
        self.ip: Optional[str] = None
        self.location: Optional[dict] = None
        self.ping: Optional[int] = None
        self.ping_record: List[int] = []

        self._ping_interval = 0
        self._lock = threading.Lock()
        self._stop_event = threading.Event()
        self._base_url = None

    async def _determine_base_url(self) -> str:
        """Prüft, ob der lokale Server erreichbar ist, sonst Remote."""
        local_url = f"http://{os.getenv('TOOLBOXV2_BASE')}:5000"
        try:
            # Kurzer Timeout für den Check
            response = await self.session.fetch(f"{local_url}/api/ping", timeout=0.5)
            if response.status_code == 200:
                return local_url
        except:
            pass
        local_url = f"http://{os.getenv('TOOLBOXV2_BASE')}:8001"
        try:
            # Kurzer Timeout für den Check
            response = await self.session.fetch(f"{local_url}/api/ping", timeout=0.5)
            if response.status_code == 200:
                return local_url
        except:
            pass
        local_url = f"http://{os.getenv('TOOLBOXV2_BASE')}:{os.getenv('TOOLBOXV2_BASE_PORT')}"
        try:
            # Kurzer Timeout für den Check
            response = await self.session.fetch(f"{local_url}/api/ping", timeout=0.5)
            if response.status_code == 200:
                return local_url
        except:
            pass
        return os.getenv('TOOLBOXV2_REMOTE_BASE')

    async def _fetch_ip_and_location(self):
        """Holt IP und Geo-Daten vom eigenen Server (minimiert externe API Aufrufe)."""
        done = False
        if self._base_url is None:
            self._base_url = await self._determine_base_url()
        response = "No Internet"
        try:
            # Kombinierter Call an deine interne API
            response = await (await self.session.fetch(f"{self._base_url}/api/geo", timeout=3)).json()
            with self._lock:
                self.ip = response.get("ip")
                self.location = response
            done = 'error' not in response
        except Exception as e:
            pass
        if done:
            return
        try:
            response = await (await self.session.fetch("https://api64.ipify.org?format=json")).json()
            self.ip = response["ip"]
            response = await (await self.session.fetch(f"https://ipapi.co/{response['ip']}/json/")).json()
            self.location = response
        except Exception as e:

            self.print("ERROR getting loc rem", e)
            pass

    async def _measure_ping(self, timeout=2.):
        """Führt 5 Round-Trips zur API/ping aus und mittelt das Ergebnis."""
        pings = []

        if self._base_url is None:
            self._base_url = await self._determine_base_url()
        for _ in range(5):
            try:
                start = time.perf_counter()
                resp = await self.session.fetch(f"{self._base_url}/api/ping", timeout=timeout)
                end = time.perf_counter()
                if resp.status_code == 200:
                    pings.append(int((end - start) * 1000))
                    timeout = pings[-1]*2
            except Exception as e:
                try:
                    start = time.perf_counter()
                    resp = await self.session.fetch(f"https://google.com", timeout=timeout)
                    end = time.perf_counter()
                    if resp.status_code == 200:
                        pings.append(int((end - start) * 1000))
                        timeout = pings[-1] * 2
                except:
                    continue


        if pings:
            avg_ping = sum(pings) // len(pings)
            with self._lock:
                self.ping = avg_ping
                self.ping_record.insert(0, avg_ping)
                self.ping_record = self.ping_record[:10]  # Nur die letzten 10

    async def _initialize_network(self):
        """Logik für initialen Start und Thread-Management."""
        # Immer einmal IP und Location holen
        if not self.alive:
            return

        await self._fetch_ip_and_location()

        if self._ping_interval == 0:
            # shnelles ping holen
            await self._measure_ping(timeout=1)
            return

        if self._ping_interval == -1:
            # Genau einmal Ping messen und dann nie wieder
            await self._measure_ping()
            return

        if self._ping_interval > 0:
            # Thread für regelmäßiges Update starten
            await self._measure_ping()  # Erstes Mal sofort
            self.run_bg_task_advanced(self._ping_worker)

        self.print(
                   (f"\n  {'IP':<8} -> {self.ip}" if self.ip else "") +
                   (f"\n  {'PING':<8} -> {self.ping}" if self.ping else "") +
                   (f"\n  {'LOC':<8} -> {self.location.get('city')}" if self.location and self.location.get('city') else ""))

    async def _ping_worker(self):
        """Hintergrund-Thread für regelmäßige Pings."""
        while not self._stop_event.is_set():
            time.sleep(self._ping_interval)
            await self._measure_ping()

    def _update_metric_tracking(self, metric_name: str, value: float):
        """Aktualisiert Min/Max/Avg für eine Metrik"""
        metrics = self._footprint_metrics[metric_name]
        metrics['max'] = max(metrics['max'], value)
        metrics['min'] = min(metrics['min'], value)
        metrics['samples'].append(value)

        # Begrenze die Anzahl der Samples (letzte 1000)
        if len(metrics['samples']) > 1000:
            metrics['samples'] = metrics['samples'][-1000:]

    def _get_metric_avg(self, metric_name: str) -> float:
        """Berechnet Durchschnitt einer Metrik"""
        samples = self._footprint_metrics[metric_name]['samples']
        return sum(samples) / len(samples) if samples else 0

    def footprint(self, update_tracking: bool = True) -> FootprintMetrics:
        """
        Erfasst den aktuellen Ressourcen-Footprint der Toolbox-Instanz.

        Args:
            update_tracking: Wenn True, aktualisiert Min/Max/Avg-Tracking

        Returns:
            FootprintMetrics mit allen erfassten Metriken
        """
        current_time = time.time()
        uptime_seconds = current_time - self._footprint_start_time

        # Formatierte Uptime
        uptime_delta = timedelta(seconds=int(uptime_seconds))
        uptime_formatted = str(uptime_delta)

        # Memory Metrics (in MB)
        try:
            mem_info = self._process.memory_info()
            memory_current = mem_info.rss / (1024 * 1024)  # Bytes zu MB
            memory_percent = self._process.memory_percent()

            if update_tracking:
                self._update_metric_tracking('memory', memory_current)

            memory_max = self._footprint_metrics['memory']['max']
            memory_min = self._footprint_metrics['memory']['min']
            if memory_min == float('inf'):
                memory_min = memory_current
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            memory_current = memory_max = memory_min = memory_percent = 0

        # CPU Metrics
        try:
            cpu_percent_current = self._process.cpu_percent(interval=0.1)
            cpu_times = self._process.cpu_times()
            cpu_time_seconds = cpu_times.user + cpu_times.system

            if update_tracking:
                self._update_metric_tracking('cpu', cpu_percent_current)

            cpu_percent_max = self._footprint_metrics['cpu']['max']
            cpu_percent_min = self._footprint_metrics['cpu']['min']
            cpu_percent_avg = self._get_metric_avg('cpu')

            if cpu_percent_min == float('inf'):
                cpu_percent_min = cpu_percent_current
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            cpu_percent_current = cpu_percent_max = 0
            cpu_percent_min = cpu_percent_avg = cpu_time_seconds = 0

        # Disk I/O Metrics (in MB)
        try:
            io_counters = self._process.io_counters()
            disk_read_bytes = io_counters.read_bytes - self._initial_disk_read
            disk_write_bytes = io_counters.write_bytes - self._initial_disk_write

            disk_read_mb = disk_read_bytes / (1024 * 1024)
            disk_write_mb = disk_write_bytes / (1024 * 1024)

            if update_tracking:
                self._update_metric_tracking('disk_read', disk_read_mb)
                self._update_metric_tracking('disk_write', disk_write_mb)

            disk_read_max = self._footprint_metrics['disk_read']['max']
            disk_read_min = self._footprint_metrics['disk_read']['min']
            disk_write_max = self._footprint_metrics['disk_write']['max']
            disk_write_min = self._footprint_metrics['disk_write']['min']

            if disk_read_min == float('inf'):
                disk_read_min = disk_read_mb
            if disk_write_min == float('inf'):
                disk_write_min = disk_write_mb
        except (AttributeError, OSError, psutil.NoSuchProcess, psutil.AccessDenied):
            disk_read_mb = disk_write_mb = 0
            disk_read_max = disk_read_min = disk_write_max = disk_write_min = 0

        # Network I/O Metrics (in MB)
        try:
            net_io = psutil.net_io_counters()
            network_sent_bytes = net_io.bytes_sent - self._initial_network_sent
            network_recv_bytes = net_io.bytes_recv - self._initial_network_recv

            network_sent_mb = network_sent_bytes / (1024 * 1024)
            network_recv_mb = network_recv_bytes / (1024 * 1024)

            if update_tracking:
                self._update_metric_tracking('network_sent', network_sent_mb)
                self._update_metric_tracking('network_recv', network_recv_mb)

            network_sent_max = self._footprint_metrics['network_sent']['max']
            network_sent_min = self._footprint_metrics['network_sent']['min']
            network_recv_max = self._footprint_metrics['network_recv']['max']
            network_recv_min = self._footprint_metrics['network_recv']['min']

            if network_sent_min == float('inf'):
                network_sent_min = network_sent_mb
            if network_recv_min == float('inf'):
                network_recv_min = network_recv_mb
        except (AttributeError, OSError):
            network_sent_mb = network_recv_mb = 0
            network_sent_max = network_sent_min = 0
            network_recv_max = network_recv_min = 0

        # Process Info
        try:
            process_id = self._process.pid
            threads = self._process.num_threads()
            open_files_path = [str(x.path).replace("\\", "/") for x in self._process.open_files()]
            connections_uri = [f"{x.laddr}:{x.raddr} {str(x.status)}" for x in self._process.connections()]

            open_files = len(open_files_path)
            connections = len(connections_uri)
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            process_id = os.getpid()
            threads = open_files = connections = 0
            open_files_path = []
            connections_uri = []

        return FootprintMetrics(
            start_time=self._footprint_start_time,
            uptime_seconds=uptime_seconds,
            uptime_formatted=uptime_formatted,
            memory_current=memory_current,
            memory_max=memory_max,
            memory_min=memory_min,
            memory_percent=memory_percent,
            cpu_percent_current=cpu_percent_current,
            cpu_percent_max=cpu_percent_max,
            cpu_percent_min=cpu_percent_min,
            cpu_percent_avg=cpu_percent_avg,
            cpu_time_seconds=cpu_time_seconds,
            disk_read_mb=disk_read_mb,
            disk_write_mb=disk_write_mb,
            disk_read_max=disk_read_max,
            disk_read_min=disk_read_min,
            disk_write_max=disk_write_max,
            disk_write_min=disk_write_min,
            network_sent_mb=network_sent_mb,
            network_recv_mb=network_recv_mb,
            network_sent_max=network_sent_max,
            network_sent_min=network_sent_min,
            network_recv_max=network_recv_max,
            network_recv_min=network_recv_min,
            process_id=process_id,
            threads=threads,
            open_files=open_files,
            connections=connections,
            open_files_path=open_files_path,
            connections_uri=connections_uri,
        )

    def print_footprint(self, detailed: bool = True) -> str:
        """
        Gibt den Footprint formatiert aus.

        Args:
            detailed: Wenn True, zeigt alle Details, sonst nur Zusammenfassung

        Returns:
            Formatierter Footprint-String
        """
        metrics = self.footprint()

        output = [
            "=" * 70,
            f"TOOLBOX FOOTPRINT - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
            "=" * 70,
            f"\n📊 UPTIME",
            f"  Runtime: {metrics.uptime_formatted}",
            f"  Seconds: {metrics.uptime_seconds:.2f}s",
            f"\n💾 MEMORY USAGE",
            f"  Current:  {metrics.memory_current:.2f} MB ({metrics.memory_percent:.2f}%)",
            f"  Maximum:  {metrics.memory_max:.2f} MB",
            f"  Minimum:  {metrics.memory_min:.2f} MB",
        ]

        if detailed:
            helper_ = '\n\t- '.join(metrics.open_files_path)
            helper__ = '\n\t- '.join(metrics.connections_uri)
            output.extend([
                f"\n⚙️  CPU USAGE",
                f"  Current:  {metrics.cpu_percent_current:.2f}%",
                f"  Maximum:  {metrics.cpu_percent_max:.2f}%",
                f"  Minimum:  {metrics.cpu_percent_min:.2f}%",
                f"  Average:  {metrics.cpu_percent_avg:.2f}%",
                f"  CPU Time: {metrics.cpu_time_seconds:.2f}s",
                f"\n💿 DISK I/O",
                f"  Read:     {metrics.disk_read_mb:.2f} MB (Max: {metrics.disk_read_max:.2f}, Min: {metrics.disk_read_min:.2f})",
                f"  Write:    {metrics.disk_write_mb:.2f} MB (Max: {metrics.disk_write_max:.2f}, Min: {metrics.disk_write_min:.2f})",
                f"\n🌐 NETWORK I/O",
                f"  Sent:     {metrics.network_sent_mb:.2f} MB (Max: {metrics.network_sent_max:.2f}, Min: {metrics.network_sent_min:.2f})",
                f"  Received: {metrics.network_recv_mb:.2f} MB (Max: {metrics.network_recv_max:.2f}, Min: {metrics.network_recv_min:.2f})",
                f"\n🔧 PROCESS INFO",
                f"  PID:         {metrics.process_id}",
                f"  Threads:     {metrics.threads}",
                f"\n📂 OPEN FILES",
                f"  Open Files:  {metrics.open_files}",
                f"  Open Files Path: \n\t- {helper_}",
                f"\n🔗 NETWORK CONNECTIONS",
                f"  Connections: {metrics.connections}",
                f"  Connections URI: \n\t- {helper__}",
            ])

        output.append("=" * 70)

        return "\n".join(output)



    def start_server(self):
        from toolboxv2.utils.clis.cli_worker_manager import WorkerManager
        from toolboxv2.utils.workers.config import load_config
        config = load_config()
        print(config.nginx.static_root)
        if self.is_server:
            return
        WorkerManager(config).start_all()
        self.is_server = False

    @staticmethod
    def exit_main(*args, **kwargs):
        """proxi attr"""

    @staticmethod
    async def hide_console(*args, **kwargs):
        """proxi attr"""

    @staticmethod
    async def show_console(*args, **kwargs):
        """proxi attr"""

    @staticmethod
    async def disconnect(*args, **kwargs):
        """proxi attr"""

    def set_logger(self, debug=False, logger_prefix=None):
        """proxi attr"""

    @property
    def debug(self):
        """proxi attr"""
        return self._debug

    def debug_rains(self, e):
        """proxi attr"""

    def set_flows(self, r):
        """proxi attr"""

    async def run_flows(self, name, **kwargs):
        """proxi attr"""

    def rrun_flows(self, name, **kwargs):
        """proxi attr"""

    def idle(self):
        import time
        self.print("idle")
        try:
            while self.alive:
                time.sleep(1)
        except KeyboardInterrupt:
            pass
        self.print("idle done")

    async def a_idle(self):
        self.print("a idle (running :"+("online)" if hasattr(self, 'daemon_app') else "offline)"))
        try:
            if hasattr(self, 'daemon_app'):
                await self.daemon_app.connect(self)
            else:
                while self.alive:
                    await asyncio.sleep(1)
        except KeyboardInterrupt:
            pass
        self.print("a idle done")

    @debug.setter
    def debug(self, value):
        """proxi attr"""

    def _coppy_mod(self, content, new_mod_dir, mod_name, file_type='py'):
        """proxi attr"""

    def _pre_lib_mod(self, mod_name, path_to="./runtime", file_type='py'):
        """proxi attr"""

    def _copy_load(self, mod_name, file_type='py', **kwargs):
        """proxi attr"""

    def inplace_load_instance(self, mod_name, loc="toolboxv2.mods.", spec='app', save=True):
        """proxi attr"""

    def save_instance(self, instance, modular_id, spec='app', instance_type="file/application", tools_class=None):
        """proxi attr"""

    def save_initialized_module(self, tools_class, spec):
        """proxi attr"""

    def mod_online(self, mod_name, installed=False):
        """proxi attr"""

    def _get_function(self,
                      name: Enum or None,
                      state: bool = True,
                      specification: str = "app",
                      metadata=False, as_str: tuple or None = None, r=0):
        """proxi attr"""

    def save_exit(self):
        """proxi attr"""

    def load_mod(self, mod_name: str, mlm='I', **kwargs):
        """proxi attr"""

    async def init_module(self, modular):
        return await self.load_mod(modular)

    async def load_external_mods(self):
        """proxi attr"""

    async def load_all_mods_in_file(self, working_dir="mods"):
        """proxi attr"""

    def get_all_mods(self, working_dir="mods", path_to="./runtime"):
        """proxi attr"""

    def remove_all_modules(self, delete=False):
        for mod in list(self.functions.keys()):
            self.logger.info(f"closing: {mod}")
            self.remove_mod(mod, delete=delete)

    async def a_remove_all_modules(self, delete=False):
        for mod in list(self.functions.keys()):
            self.logger.info(f"closing: {mod}")
            await self.a_remove_mod(mod, delete=delete)

    def print_ok(self):
        """proxi attr"""
        self.logger.info("OK")

    def reload_mod(self, mod_name, spec='app', is_file=True, loc="toolboxv2.mods."):
        """proxi attr"""

    def watch_mod(self, mod_name, spec='app', loc="toolboxv2.mods.", use_thread=True, path_name=None):
        """proxi attr"""

    def remove_mod(self, mod_name, spec='app', delete=True):
        """proxi attr"""

    async def a_remove_mod(self, mod_name, spec='app', delete=True):
        """proxi attr"""

    def exit(self):
        """proxi attr"""

    def web_context(self) -> str:
        """returns the build index ( toolbox web component )"""

    async def a_exit(self):
        """proxi attr"""

    def save_load(self, modname, spec='app'):
        """proxi attr"""

    def get_function(self, name: Enum or tuple, **kwargs):
        """
        Kwargs for _get_function
            metadata:: return the registered function dictionary
                stateless: (function_data, None), 0
                stateful: (function_data, higher_order_function), 0
            state::boolean
                specification::str default app
        """

    def run_a_from_sync(self, function, *args):
        """
        run a async fuction
        """

    def run_bg_task_advanced(self, task, *args, **kwargs):
        """
        proxi attr
        """

    def wait_for_bg_tasks(self, timeout=None):
        """
        proxi attr
        """

    def run_bg_task(self, task):
        """
                run a async fuction
                """
    def run_function(self, mod_function_name: Enum or tuple,
                     tb_run_function_with_state=True,
                     tb_run_with_specification='app',
                     args_=None,
                     kwargs_=None,
                     *args,
                     **kwargs) -> Result:

        """proxi attr"""

    async def a_run_function(self, mod_function_name: Enum or tuple,
                             tb_run_function_with_state=True,
                             tb_run_with_specification='app',
                             args_=None,
                             kwargs_=None,
                             *args,
                             **kwargs) -> Result:

        """proxi attr"""

    def fuction_runner(self, function, function_data: dict, args: list, kwargs: dict, t0=.0):
        """
        parameters = function_data.get('params')
        modular_name = function_data.get('module_name')
        function_name = function_data.get('func_name')
        mod_function_name = f"{modular_name}.{function_name}"

        proxi attr
        """

    async def a_fuction_runner(self, function, function_data: dict, args: list, kwargs: dict):
        """
        parameters = function_data.get('params')
        modular_name = function_data.get('module_name')
        function_name = function_data.get('func_name')
        mod_function_name = f"{modular_name}.{function_name}"

        proxi attr
        """

    async def run_http(
        self,
        mod_function_name: Enum or str or tuple,
        function_name=None,
        method="GET",
        args_=None,
        kwargs_=None,
        *args,
        **kwargs,
    ):
        """run a function remote via http / https"""

    def run_any(self, mod_function_name: Enum or str or tuple, backwords_compability_variabel_string_holder=None,
                get_results=False, tb_run_function_with_state=True, tb_run_with_specification='app', args_=None,
                kwargs_=None,
                *args, **kwargs):
        """proxi attr"""

    async def a_run_any(self, mod_function_name: Enum or str or tuple,
                        backwords_compability_variabel_string_holder=None,
                        get_results=False, tb_run_function_with_state=True, tb_run_with_specification='app', args_=None,
                        kwargs_=None,
                        *args, **kwargs):
        """proxi attr"""

    def get_mod(self, name, spec='app') -> ModuleType or MainToolType:
        """proxi attr"""

    @staticmethod
    def print(text, *args, **kwargs):
        """proxi attr"""

    @staticmethod
    def sprint(text, *args, **kwargs):
        """proxi attr"""

    # ----------------------------------------------------------------
    # Decorators for the toolbox

    def _register_function(self, module_name, func_name, data):
        """proxi attr"""

    def _create_decorator(
        self,
        type_: str,
        name: str = "",
        mod_name: str = "",
        level: int = -1,
        restrict_in_virtual_mode: bool = False,
        api: bool = False,
        helper: str = "",
        version: str or None = None,
        initial=False,
        exit_f=False,
        test=True,
        samples=None,
        state=None,
        pre_compute=None,
        post_compute=None,
        memory_cache=False,
        file_cache=False,
        row=False,
        request_as_kwarg=False,
        memory_cache_max_size=100,
        memory_cache_ttl=300,
        websocket_handler: str | None = None,
        websocket_context: bool = False,
    ):
        """proxi attr"""

        # data = {
        #     "type": type_,
        #     "module_name": module_name,
        #     "func_name": func_name,
        #     "level": level,
        #     "restrict_in_virtual_mode": restrict_in_virtual_mode,
        #     "func": func,
        #     "api": api,
        #     "helper": helper,
        #     "version": version,
        #     "initial": initial,
        #     "exit_f": exit_f,
        #     "__module__": func.__module__,
        #     "signature": sig,
        #     "params": params,
        #     "state": (
        #         False if len(params) == 0 else params[0] in ['self', 'state', 'app']) if state is None else state,
        #     "do_test": test,
        #     "samples": samples,
        #     "request_as_kwarg": request_as_kwarg,

    def tb(self, name=None,
           mod_name: str = "",
           helper: str = "",
           version: str or None = None,
           test: bool = True,
           restrict_in_virtual_mode: bool = False,
           api: bool = False,
           initial: bool = False,
           exit_f: bool = False,
           test_only: bool = False,
           memory_cache: bool = False,
           file_cache: bool = False,
           row=False,
           request_as_kwarg: bool = False,
           state: bool or None = None,
           level: int = 0,
           memory_cache_max_size: int = 100,
           memory_cache_ttl: int = 300,
           samples: list or dict or None = None,
           interface: ToolBoxInterfaces or None or str = None,
           pre_compute=None,
           post_compute=None,
           api_methods=None,
           websocket_handler: str | None = None,
           websocket_context: bool = False,
           ):
        """
    A decorator for registering and configuring functions within a module.

    This decorator is used to wrap functions with additional functionality such as caching, API conversion, and lifecycle management (initialization and exit). It also handles the registration of the function in the module's function registry.

    Args:
        name (str, optional): The name to register the function under. Defaults to the function's own name.
        mod_name (str, optional): The name of the module the function belongs to.
        helper (str, optional): A helper string providing additional information about the function.
        version (str or None, optional): The version of the function or module.
        test (bool, optional): Flag to indicate if the function is for testing purposes.
        restrict_in_virtual_mode (bool, optional): Flag to restrict the function in virtual mode.
        api (bool, optional): Flag to indicate if the function is part of an API.
        initial (bool, optional): Flag to indicate if the function should be executed at initialization.
        exit_f (bool, optional): Flag to indicate if the function should be executed at exit.
        test_only (bool, optional): Flag to indicate if the function should only be used for testing.
        memory_cache (bool, optional): Flag to enable memory caching for the function.
        request_as_kwarg (bool, optional): Flag to get request if the fuction is calld from api.
        file_cache (bool, optional): Flag to enable file caching for the function.
        row (bool, optional): rather to auto wrap the result in Result type default False means no row data aka result type
        state (bool or None, optional): Flag to indicate if the function maintains state.
        level (int, optional): The level of the function, used for prioritization or categorization.
        memory_cache_max_size (int, optional): Maximum size of the memory cache.
        memory_cache_ttl (int, optional): Time-to-live for the memory cache entries.
        samples (list or dict or None, optional): Samples or examples of function usage.
        interface (str, optional): The interface type for the function.
        pre_compute (callable, optional): A function to be called before the main function.
        post_compute (callable, optional): A function to be called after the main function.
        api_methods (list[str], optional): default ["AUTO"] (GET if not params, POST if params) , GET, POST, PUT or DELETE.

    Returns:
        function: The decorated function with additional processing and registration capabilities.
    """
        if interface is None:
            interface = "tb"
        if test_only and 'test' not in self.id:
            return lambda *args, **kwargs: args
        return self._create_decorator(
            interface,
            name,
            mod_name,
            version=version,
            test=test,
            restrict_in_virtual_mode=restrict_in_virtual_mode,
            api=api,
            initial=initial,
            exit_f=exit_f,
            test_only=test_only,
            memory_cache=memory_cache,
            file_cache=file_cache,
            row=row,
            request_as_kwarg=request_as_kwarg,
            state=state,
            level=level,
            memory_cache_max_size=memory_cache_max_size,
            memory_cache_ttl=memory_cache_ttl,
            samples=samples,
            interface=interface,
            pre_compute=pre_compute,
            post_compute=post_compute,
            api_methods=api_methods,
            websocket_handler=websocket_handler,
            websocket_context=websocket_context,
        )

    def print_functions(self, name=None):
        if not self.functions:
            return

        def helper(_functions):
            for func_name, data in _functions.items():
                if not isinstance(data, dict):
                    continue

                func_type = data.get("type", "Unknown")
                func_level = "r" if data["level"] == -1 else data["level"]
                api_status = "Api" if data.get("api", False) else "Non-Api"

                print(
                    f"  Function: {func_name}{data.get('signature', '()')}; "
                    f"Type: {func_type}, Level: {func_level}, {api_status}"
                )

        if name is not None:
            functions = self.functions.get(name)
            if functions is not None:
                print(
                    f"\nModule: {name}; Type: {functions.get('app_instance_type', 'Unknown')}"
                )
                helper(functions)
                return
        for module, functions in self.functions.items():
            print(
                f"\nModule: {module}; Type: {functions.get('app_instance_type', 'Unknown')}"
            )
            helper(functions)

    def save_autocompletion_dict(self):
        """proxi attr"""

    def get_autocompletion_dict(self):
        """proxi attr"""

    def get_username(self, get_input=False, default="loot") -> str:
        """proxi attr"""

    def save_registry_as_enums(self, directory: str, filename: str):
        """proxi attr"""

    async def docs_reader(
        self,
        query: Optional[str] = None,
        section_id: Optional[str] = None,
        file_path: Optional[str] = None,
        tags: Optional[List[str]] = None,
        max_results: int = 25,
        format_type: str = "structured",
    ) -> dict:
        """"mkdocs system [extra]"""
    async def docs_writer(self, action: str, **kwargs) -> dict:
        """"mkdocs system [extra]
        Actions:
            - create_file
                Kwargs: file_path, content
                Returns: {"status": "created", "file": file_path, "sections": num_sections}
            - add_section
                Kwargs: file_path, section_title, content, position, level
                Returns: {"status": "added", "section": section_id}
            - update_section
                Kwargs: section_id, content
                Returns: {"status": "updated", "section": section_id}
            - delete_section
                Kwargs: section_id
                Returns: {"status": "deleted", "section": section_id}

            on error
                Returns: {"error": "error_message"}
        """
    async def docs_lookup(self,
                          name: Optional[str] = None,
                          element_type: Optional[str] = None,
                          file_path: Optional[str] = None,
                          language: Optional[str] = None,
                          include_code: bool = False,
                          max_results: int = 25,
                          ) -> dict:
        """"mkdocs system [extra]"""
    async def docs_suggestions(self, max_suggestions: int = 20) -> dict:
        """mkdocs system [extra]
            Returns:
                {"suggestions": [{"type": "unclear_section", "section_id": "123", "title": "Section Title", "priority": "low"}, ...], "total": 100, "time_ms": 123}
        """

    async def docs_sync(self):
        """"mkdocs system [extra]"""
    async def docs_init(self, force_rebuild: bool = False) -> dict:
        """mkdocs system [extra]
            Returns:
                {"status": "loaded", "sections": num_sections, "elements": num_elements, "time_ms": time_taken}
        """
    async def get_task_context(self, files: List[str], intent: str) -> dict:
        """mkdocs system [extra]
        Get optimized context for a specific editing task.

        Args:
            files: List of file paths relevant to the task.
            intent: Description of what the user wants to do (e.g., "Add logging to auth").

        Returns:
            ContextBundle dictionary ready for LLM injection.
        """

    async def execute_all_functions_(self, m_query='', f_query='', test_class=None):
        from ..extras import generate_test_cases
        all_data = {
            "modular_run": 0,
            "modular_fatal_error": 0,
            "errors": 0,
            "modular_sug": 0,
            "coverage": [],
            "total_coverage": {},
        }
        items = list(self.functions.items()).copy()
        print("Executing all functions", len(items))

        for module_name, functions in items:
            infos = {
                "functions_run": 0,
                "functions_fatal_error": 0,
                "error": 0,
                "functions_sug": 0,
                'calls': {},
                'callse': {},
                "coverage": [0, 0],
                "timeouts": [],  # neu
            }
            all_data['modular_run'] += 1
            if not module_name.startswith(m_query):
                all_data['modular_sug'] += 1
                continue

            with Spinner(message=f"In {module_name}|"):
                f_items = list(functions.items()).copy()

                # ── 1. Tasks sammeln ──────────────────────────────────────────
                tasks_meta = []  # (function_name, test_kwargs, task)
                for function_name, function_data in f_items:
                    if not isinstance(function_data, dict):
                        continue
                    if not function_name.startswith(f_query):
                        continue

                    infos["coverage"][0] += 1
                    if function_data.get('do_test') is False:
                        continue
                    infos["coverage"][1] += 1

                    params = function_data.get('params')
                    sig = function_data.get('signature')
                    state = function_data.get('state')
                    samples = function_data.get('samples')

                    test_kwargs_list = [{}]
                    if params is not None:
                        test_kwargs_list = samples if samples is not None else generate_test_cases(sig=sig)

                    for test_kwargs in test_kwargs_list:
                        coro = self.a_run_function(
                            (module_name, function_name),
                            tb_run_function_with_state=state,
                            **test_kwargs,
                        )
                        # asyncio.Task → kann einzeln gecancelt / inspiziert werden
                        task = asyncio.ensure_future(coro)
                        tasks_meta.append((function_name, test_kwargs, task))

                # ── 2. Alle parallel mit globalem 15 s Timeout warten ────────
                all_tasks = [t for _, _, t in tasks_meta]
                # return_exceptions=True → kein Task bricht die anderen ab
                if all_tasks:
                    await asyncio.wait(all_tasks, timeout=15)
                else:
                    pass  # Modul hat keine testbaren Funktionen → direkt zu Schritt 3
                # ── 3. Ergebnisse auswerten ───────────────────────────────────
                for function_name, test_kwargs, task in tasks_meta:
                    infos['functions_run'] += 1
                    ctx = f"{module_name}.{function_name}"

                    if not task.done():
                        # Nach 15 s noch nicht fertig → Timeout
                        task.cancel()
                        infos['functions_fatal_error'] += 1
                        infos['timeouts'].append(function_name)
                        infos['callse'][function_name] = [test_kwargs, "TIMEOUT (>15s)"]
                        if test_class is not None:
                            with test_class.subTest(ctx):
                                test_class.fail(f"TIMEOUT: {ctx}")
                        continue

                    exc = task.exception()
                    if exc is not None:
                        infos['functions_fatal_error'] += 1
                        infos['callse'][function_name] = [test_kwargs, str(exc)]
                        if test_class is not None:
                            with test_class.subTest(ctx):
                                test_class.fail(str(exc))
                        continue

                    result = task.result()
                    if not isinstance(result, Result):
                        result = Result.ok(result)

                    if test_class is not None:
                        with test_class.subTest(ctx):
                            test_class.assertTrue(
                                not result.is_error(),
                                result.print(show=False, full_data=True),
                            )

                    infos['functions_sug'] += 1
                    if result.info.exec_code == 0:
                        infos['calls'][function_name] = [test_kwargs, str(result)]
                    else:
                        infos['error'] += 1
                        infos['callse'][function_name] = [test_kwargs, str(result)]

                # ── 4. Modul-Aggregation ──────────────────────────────────────
                if infos['functions_run'] == infos['functions_sug']:
                    all_data['modular_sug'] += 1
                else:
                    all_data['modular_fatal_error'] += 1
                if infos['error'] > 0:
                    all_data['errors'] += infos['error']

                all_data[module_name] = infos
                c = infos['coverage'][1] / infos['coverage'][0] if infos['coverage'][0] else 0
                all_data["coverage"].append(f"{module_name}:{c:.2f}\n")

        # ── 5. Finaler Bericht ────────────────────────────────────────────────
        total_coverage = (
            sum(float(t.split(":")[-1]) for t in all_data["coverage"]) / len(all_data["coverage"])
            if all_data["coverage"] else 0.0
        )

        # Timeouts über alle Module sammeln
        all_timeouts = {
            mod: data["timeouts"]
            for mod, data in all_data.items()
            if isinstance(data, dict) and data.get("timeouts")
        }

        print(
            f"\n{all_data['modular_run']=}"
            f"\n{all_data['modular_sug']=}"
            f"\n{all_data['modular_fatal_error']=}"
            f"\n{total_coverage=:.2f}"
            f"\nTimeouts: {all_timeouts or 'keine'}"
        )

        d = analyze_data(all_data)
        return Result.ok(data=all_data, data_info=d)

    async def execute_function_test(self, module_name: str, function_name: str,
                                    function_data: dict, test_kwargs: dict,
                                    profiler: cProfile.Profile) -> tuple[bool, str, dict, float]:
        start_time = time.time()
        with profile_section(profiler, hasattr(self, 'enable_profiling') and self.enable_profiling):
            try:
                result = await self.a_run_function(
                    (module_name, function_name),
                    tb_run_function_with_state=function_data.get('state'),
                    **test_kwargs
                )

                if not isinstance(result, Result):
                    result = Result.ok(result)

                success = result.info.exec_code == 0
                execution_time = time.time() - start_time
                return success, str(result), test_kwargs, execution_time
            except Exception as e:
                execution_time = time.time() - start_time
                return False, str(e), test_kwargs, execution_time

    async def process_function(self, module_name: str, function_name: str,
                               function_data: dict, profiler: cProfile.Profile) -> tuple[str, ModuleInfo]:
        start_time = time.time()
        info = ModuleInfo()

        with profile_section(profiler, hasattr(self, 'enable_profiling') and self.enable_profiling):
            if not isinstance(function_data, dict):
                return function_name, info

            test = function_data.get('do_test')
            info.coverage[0] += 1

            if test is False:
                return function_name, info

            params = function_data.get('params')
            sig = function_data.get('signature')
            samples = function_data.get('samples')

            test_kwargs_list = [{}] if params is None else (
                samples if samples is not None else generate_test_cases(sig=sig)
            )

            info.coverage[1] += 1

            # Create tasks for all test cases
            tasks = [
                self.execute_function_test(module_name, function_name, function_data, test_kwargs, profiler)
                for test_kwargs in test_kwargs_list
            ]

            # Execute all tests concurrently
            results = await asyncio.gather(*tasks)

            total_execution_time = 0
            for success, result_str, test_kwargs, execution_time in results:
                info.functions_run += 1
                total_execution_time += execution_time

                if success:
                    info.functions_sug += 1
                    info.calls[function_name] = [test_kwargs, result_str]
                else:
                    info.functions_sug += 1
                    info.error += 1
                    info.callse[function_name] = [test_kwargs, result_str]

            info.execution_time = time.time() - start_time
            return function_name, info

    async def process_module(self, module_name: str, functions: dict,
                             f_query: str, profiler: cProfile.Profile) -> tuple[str, ModuleInfo]:
        start_time = time.time()

        with profile_section(profiler, hasattr(self, 'enable_profiling') and self.enable_profiling):
            async with asyncio.Semaphore(mp.cpu_count()):
                tasks = [
                    self.process_function(module_name, fname, fdata, profiler)
                    for fname, fdata in functions.items()
                    if fname.startswith(f_query)
                ]

                if not tasks:
                    return module_name, ModuleInfo()

                results = await asyncio.gather(*tasks)

                # Combine results from all functions in the module
                combined_info = ModuleInfo()
                total_execution_time = 0

                for _, info in results:
                    combined_info.functions_run += info.functions_run
                    combined_info.functions_fatal_error += info.functions_fatal_error
                    combined_info.error += info.error
                    combined_info.functions_sug += info.functions_sug
                    combined_info.calls.update(info.calls)
                    combined_info.callse.update(info.callse)
                    combined_info.coverage[0] += info.coverage[0]
                    combined_info.coverage[1] += info.coverage[1]
                    total_execution_time += info.execution_time

                combined_info.execution_time = time.time() - start_time
                return module_name, combined_info

    async def execute_all_functions(self, m_query='', f_query='', enable_profiling=True):
        """
        Execute all functions with parallel processing and optional profiling.

        Args:
            m_query (str): Module name query filter
            f_query (str): Function name query filter
            enable_profiling (bool): Enable detailed profiling information
        """
        print("Executing all functions in parallel" + (" with profiling" if enable_profiling else ""))

        start_time = time.time()
        stats = ExecutionStats()
        items = list(self.functions.items()).copy()

        # Set up profiling
        self.enable_profiling = enable_profiling
        profiler = cProfile.Profile()

        with profile_section(profiler, enable_profiling):
            # Filter modules based on query
            filtered_modules = [
                (mname, mfuncs) for mname, mfuncs in items
                if mname.startswith(m_query)
            ]

            stats.modular_run = len(filtered_modules)

            # Process all modules concurrently
            async with asyncio.Semaphore(mp.cpu_count()):
                tasks = [
                    self.process_module(mname, mfuncs, f_query, profiler)
                    for mname, mfuncs in filtered_modules
                ]

                results = await asyncio.gather(*tasks)

            # Combine results and calculate statistics
            for module_name, info in results:
                if info.functions_run == info.functions_sug:
                    stats.modular_sug += 1
                else:
                    stats.modular_fatal_error += 1

                stats.errors += info.error

                # Calculate coverage
                coverage = (
                    (info.coverage[1] / info.coverage[0]) if info.coverage[0] > 0 else 0
                )
                stats.coverage.append(f"{module_name}:{coverage:.2f}\n")

                # Store module info
                stats.__dict__[module_name] = info

            # Calculate total coverage
            total_coverage = (
                sum(float(t.split(":")[-1]) for t in stats.coverage) / len(stats.coverage)
                if stats.coverage
                else 0
            )

            stats.total_execution_time = time.time() - start_time

            # Generate profiling stats if enabled
            if enable_profiling:
                s = io.StringIO()
                ps = pstats.Stats(profiler, stream=s).sort_stats("cumulative")
                ps.print_stats()
                stats.profiling_data = {
                    "detailed_stats": s.getvalue(),
                    "total_time": stats.total_execution_time,
                    "function_count": stats.modular_run,
                    "successful_functions": stats.modular_sug,
                }

            print(
                f"\n{stats.modular_run=}"
                f"\n{stats.modular_sug=}"
                f"\n{stats.modular_fatal_error=}"
                f"\n{total_coverage=}"
                f"\nTotal execution time: {stats.total_execution_time:.2f}s"
            )

            if enable_profiling:
                print("\nProfiling Summary:")
                print(f"{'=' * 50}")
                print("Top 10 time-consuming functions:")
                ps.print_stats(10)

            analyzed_data = analyze_data(stats.__dict__)
            return Result.ok(data=stats.__dict__, data_info=analyzed_data)

    def generate_openapi_html(self):
        """
        Generiert eine HTML-Datei mit OpenAPI/Swagger UI für API-Routen.

        Args:
        """

        # OpenAPI Spec erstellen
        openapi_spec = {
            "openapi": "3.0.0",
            "info": {
                "title": "CloudM API Services",
                "version": "0.1.24",
                "description": "API Documentation für CloudM Email Services",
            },
            "servers": [{"url": "/api", "description": "API Server"}],
            "paths": {},
        }

        # Durch alle Services iterieren
        for service_name, functions in self.functions.items():
            for func_name, func_info in functions.items():
                # Nur API-Funktionen verarbeiten
                if not isinstance(func_info, dict):
                    continue
                if not func_info.get("api", False):
                    continue

                # Parameter aus der Signatur extrahieren
                params = func_info.get("params", [])
                # 'app' Parameter ausschließen (interner Parameter)
                api_params = [p for p in params if p != "app"]

                # Request Body Schema erstellen
                properties = {}
                required = []

                for param in api_params:
                    properties[param] = {
                        "type": "string",
                        "description": f"Parameter: {param}",
                    }
                    # Prüfen ob Parameter optional ist (hat default value)
                    if "=" not in str(func_info.get("signature", "")):
                        required.append(param)

                # API Path erstellen
                path = f"/{service_name}/{func_name}"

                # Path Operation definieren
                openapi_spec["paths"][path] = {
                    "post": {
                        "summary": func_name.replace("_", " ").title(),
                        "description": f"Funktion: {func_name} aus Modul {func_info.get('module_name', 'unknown')}",
                        "tags": [service_name],
                        "requestBody": {
                            "required": True,
                            "content": {
                                "application/json": {
                                    "schema": {
                                        "type": "object",
                                        "properties": properties,
                                        "required": required,
                                    }
                                }
                            },
                        },
                        "responses": {
                            "200": {
                                "description": "Erfolgreiche Antwort",
                                "content": {
                                    "application/json": {"schema": {"type": "object"}}
                                },
                            },
                            "400": {"description": "Ungültige Anfrage"},
                            "500": {"description": "Serverfehler"},
                        },
                    }
                }

        # HTML Template mit Swagger UI
        html_content = f"""<!DOCTYPE html>
    <html lang="de">
    <head>
        <meta charset="UTF-8">
        <meta name="viewport" content="width=device-width, initial-scale=1.0">
        <title>CloudM API Documentation</title>
        <link rel="stylesheet" type="text/css" href="https://cdnjs.cloudflare.com/ajax/libs/swagger-ui/5.10.5/swagger-ui.min.css">
        <style>
            body {{
                margin: 0;
                padding: 0;
            }}
            #swagger-ui {{
                max-width: 1460px;
                margin: 0 auto;
            }}
        </style>
    </head>
    <body>
        <div id="swagger-ui"></div>

        <script src="https://cdnjs.cloudflare.com/ajax/libs/swagger-ui/5.10.5/swagger-ui-bundle.min.js"></script>
        <script src="https://cdnjs.cloudflare.com/ajax/libs/swagger-ui/5.10.5/swagger-ui-standalone-preset.min.js"></script>
        <script unsave="true">
            const onload = function() {{
                const spec = {json.dumps(openapi_spec, indent=2)};

                window.ui = SwaggerUIBundle({{
                    spec: spec,
                    dom_id: '#swagger-ui',
                    deepLinking: true,
                    presets: [
                        SwaggerUIBundle.presets.apis,
                        SwaggerUIStandalonePreset
                    ],
                    plugins: [
                        SwaggerUIBundle.plugins.DownloadUrl
                    ],
                    layout: "StandaloneLayout"
                }});
            }};
            if (window.TB?.onLoaded) {{
                window.TB.onLoaded(onload());
            }} else {{
               window.addEventListener('DOMContentLoaded', onload)
            }}
        </script>
    </body>
    </html>"""
        print(f"✓ Gefundene API-Routen: {len(openapi_spec['paths'])}")
        return Result.html(html_content, row=True)

debug property writable

proxi attr

a_exit() async

proxi attr

Source code in toolboxv2/utils/system/types.py

async def a_exit(self):
    """proxi attr"""

a_fuction_runner(function, function_data, args, kwargs) async

parameters = function_data.get('params') modular_name = function_data.get('module_name') function_name = function_data.get('func_name') mod_function_name = f"{modular_name}.{function_name}"

proxi attr

Source code in toolboxv2/utils/system/types.py

async def a_fuction_runner(self, function, function_data: dict, args: list, kwargs: dict):
    """
    parameters = function_data.get('params')
    modular_name = function_data.get('module_name')
    function_name = function_data.get('func_name')
    mod_function_name = f"{modular_name}.{function_name}"

    proxi attr
    """

a_remove_mod(mod_name, spec='app', delete=True) async

proxi attr

Source code in toolboxv2/utils/system/types.py

async def a_remove_mod(self, mod_name, spec='app', delete=True):
    """proxi attr"""

a_run_any(mod_function_name, backwords_compability_variabel_string_holder=None, get_results=False, tb_run_function_with_state=True, tb_run_with_specification='app', args_=None, kwargs_=None, *args, **kwargs) async

proxi attr

Source code in toolboxv2/utils/system/types.py

async def a_run_any(self, mod_function_name: Enum or str or tuple,
                    backwords_compability_variabel_string_holder=None,
                    get_results=False, tb_run_function_with_state=True, tb_run_with_specification='app', args_=None,
                    kwargs_=None,
                    *args, **kwargs):
    """proxi attr"""

a_run_function(mod_function_name, tb_run_function_with_state=True, tb_run_with_specification='app', args_=None, kwargs_=None, *args, **kwargs) async

proxi attr

Source code in toolboxv2/utils/system/types.py

async def a_run_function(self, mod_function_name: Enum or tuple,
                         tb_run_function_with_state=True,
                         tb_run_with_specification='app',
                         args_=None,
                         kwargs_=None,
                         *args,
                         **kwargs) -> Result:

    """proxi attr"""

debug_rains(e)

proxi attr

Source code in toolboxv2/utils/system/types.py

def debug_rains(self, e):
    """proxi attr"""

disconnect(*args, **kwargs) async staticmethod

proxi attr

Source code in toolboxv2/utils/system/types.py

@staticmethod
async def disconnect(*args, **kwargs):
    """proxi attr"""

docs_init(force_rebuild=False) async

mkdocs system [extra] Returns:

Source code in toolboxv2/utils/system/types.py

async def docs_init(self, force_rebuild: bool = False) -> dict:
    """mkdocs system [extra]
        Returns:
            {"status": "loaded", "sections": num_sections, "elements": num_elements, "time_ms": time_taken}
    """

docs_lookup(name=None, element_type=None, file_path=None, language=None, include_code=False, max_results=25) async

"mkdocs system [extra]

Source code in toolboxv2/utils/system/types.py

async def docs_lookup(self,
                      name: Optional[str] = None,
                      element_type: Optional[str] = None,
                      file_path: Optional[str] = None,
                      language: Optional[str] = None,
                      include_code: bool = False,
                      max_results: int = 25,
                      ) -> dict:
    """"mkdocs system [extra]"""

docs_reader(query=None, section_id=None, file_path=None, tags=None, max_results=25, format_type='structured') async

"mkdocs system [extra]

Source code in toolboxv2/utils/system/types.py

async def docs_reader(
    self,
    query: Optional[str] = None,
    section_id: Optional[str] = None,
    file_path: Optional[str] = None,
    tags: Optional[List[str]] = None,
    max_results: int = 25,
    format_type: str = "structured",
) -> dict:
    """"mkdocs system [extra]"""

docs_suggestions(max_suggestions=20) async

mkdocs system [extra] Returns: {"suggestions": [{"type": "unclear_section", "section_id": "123", "title": "Section Title", "priority": "low"}, ...], "total": 100, "time_ms": 123}

Source code in toolboxv2/utils/system/types.py

async def docs_suggestions(self, max_suggestions: int = 20) -> dict:
    """mkdocs system [extra]
        Returns:
            {"suggestions": [{"type": "unclear_section", "section_id": "123", "title": "Section Title", "priority": "low"}, ...], "total": 100, "time_ms": 123}
    """

docs_sync() async

"mkdocs system [extra]

Source code in toolboxv2/utils/system/types.py

async def docs_sync(self):
    """"mkdocs system [extra]"""

docs_writer(action, **kwargs) async

"mkdocs system [extra] Actions: - create_file Kwargs: file_path, content Returns: {"status": "created", "file": file_path, "sections": num_sections} - add_section Kwargs: file_path, section_title, content, position, level Returns: {"status": "added", "section": section_id} - update_section Kwargs: section_id, content Returns: {"status": "updated", "section": section_id} - delete_section Kwargs: section_id Returns: {"status": "deleted", "section": section_id}

on error
    Returns: {"error": "error_message"}

Source code in toolboxv2/utils/system/types.py

async def docs_writer(self, action: str, **kwargs) -> dict:
    """"mkdocs system [extra]
    Actions:
        - create_file
            Kwargs: file_path, content
            Returns: {"status": "created", "file": file_path, "sections": num_sections}
        - add_section
            Kwargs: file_path, section_title, content, position, level
            Returns: {"status": "added", "section": section_id}
        - update_section
            Kwargs: section_id, content
            Returns: {"status": "updated", "section": section_id}
        - delete_section
            Kwargs: section_id
            Returns: {"status": "deleted", "section": section_id}

        on error
            Returns: {"error": "error_message"}
    """

execute_all_functions(m_query='', f_query='', enable_profiling=True) async

Execute all functions with parallel processing and optional profiling.

Parameters:

Name	Type	Description	Default
m_query	str	Module name query filter	''
f_query	str	Function name query filter	''
enable_profiling	bool	Enable detailed profiling information	True

Source code in toolboxv2/utils/system/types.py

async def execute_all_functions(self, m_query='', f_query='', enable_profiling=True):
    """
    Execute all functions with parallel processing and optional profiling.

    Args:
        m_query (str): Module name query filter
        f_query (str): Function name query filter
        enable_profiling (bool): Enable detailed profiling information
    """
    print("Executing all functions in parallel" + (" with profiling" if enable_profiling else ""))

    start_time = time.time()
    stats = ExecutionStats()
    items = list(self.functions.items()).copy()

    # Set up profiling
    self.enable_profiling = enable_profiling
    profiler = cProfile.Profile()

    with profile_section(profiler, enable_profiling):
        # Filter modules based on query
        filtered_modules = [
            (mname, mfuncs) for mname, mfuncs in items
            if mname.startswith(m_query)
        ]

        stats.modular_run = len(filtered_modules)

        # Process all modules concurrently
        async with asyncio.Semaphore(mp.cpu_count()):
            tasks = [
                self.process_module(mname, mfuncs, f_query, profiler)
                for mname, mfuncs in filtered_modules
            ]

            results = await asyncio.gather(*tasks)

        # Combine results and calculate statistics
        for module_name, info in results:
            if info.functions_run == info.functions_sug:
                stats.modular_sug += 1
            else:
                stats.modular_fatal_error += 1

            stats.errors += info.error

            # Calculate coverage
            coverage = (
                (info.coverage[1] / info.coverage[0]) if info.coverage[0] > 0 else 0
            )
            stats.coverage.append(f"{module_name}:{coverage:.2f}\n")

            # Store module info
            stats.__dict__[module_name] = info

        # Calculate total coverage
        total_coverage = (
            sum(float(t.split(":")[-1]) for t in stats.coverage) / len(stats.coverage)
            if stats.coverage
            else 0
        )

        stats.total_execution_time = time.time() - start_time

        # Generate profiling stats if enabled
        if enable_profiling:
            s = io.StringIO()
            ps = pstats.Stats(profiler, stream=s).sort_stats("cumulative")
            ps.print_stats()
            stats.profiling_data = {
                "detailed_stats": s.getvalue(),
                "total_time": stats.total_execution_time,
                "function_count": stats.modular_run,
                "successful_functions": stats.modular_sug,
            }

        print(
            f"\n{stats.modular_run=}"
            f"\n{stats.modular_sug=}"
            f"\n{stats.modular_fatal_error=}"
            f"\n{total_coverage=}"
            f"\nTotal execution time: {stats.total_execution_time:.2f}s"
        )

        if enable_profiling:
            print("\nProfiling Summary:")
            print(f"{'=' * 50}")
            print("Top 10 time-consuming functions:")
            ps.print_stats(10)

        analyzed_data = analyze_data(stats.__dict__)
        return Result.ok(data=stats.__dict__, data_info=analyzed_data)

exit()

proxi attr

Source code in toolboxv2/utils/system/types.py

def exit(self):
    """proxi attr"""

exit_main(*args, **kwargs) staticmethod

proxi attr

Source code in toolboxv2/utils/system/types.py

@staticmethod
def exit_main(*args, **kwargs):
    """proxi attr"""

footprint(update_tracking=True)

Erfasst den aktuellen Ressourcen-Footprint der Toolbox-Instanz.

Parameters:

Name	Type	Description	Default
update_tracking	bool	Wenn True, aktualisiert Min/Max/Avg-Tracking	True

Returns:

Type	Description
FootprintMetrics	FootprintMetrics mit allen erfassten Metriken

Source code in toolboxv2/utils/system/types.py

def footprint(self, update_tracking: bool = True) -> FootprintMetrics:
    """
    Erfasst den aktuellen Ressourcen-Footprint der Toolbox-Instanz.

    Args:
        update_tracking: Wenn True, aktualisiert Min/Max/Avg-Tracking

    Returns:
        FootprintMetrics mit allen erfassten Metriken
    """
    current_time = time.time()
    uptime_seconds = current_time - self._footprint_start_time

    # Formatierte Uptime
    uptime_delta = timedelta(seconds=int(uptime_seconds))
    uptime_formatted = str(uptime_delta)

    # Memory Metrics (in MB)
    try:
        mem_info = self._process.memory_info()
        memory_current = mem_info.rss / (1024 * 1024)  # Bytes zu MB
        memory_percent = self._process.memory_percent()

        if update_tracking:
            self._update_metric_tracking('memory', memory_current)

        memory_max = self._footprint_metrics['memory']['max']
        memory_min = self._footprint_metrics['memory']['min']
        if memory_min == float('inf'):
            memory_min = memory_current
    except (psutil.NoSuchProcess, psutil.AccessDenied):
        memory_current = memory_max = memory_min = memory_percent = 0

    # CPU Metrics
    try:
        cpu_percent_current = self._process.cpu_percent(interval=0.1)
        cpu_times = self._process.cpu_times()
        cpu_time_seconds = cpu_times.user + cpu_times.system

        if update_tracking:
            self._update_metric_tracking('cpu', cpu_percent_current)

        cpu_percent_max = self._footprint_metrics['cpu']['max']
        cpu_percent_min = self._footprint_metrics['cpu']['min']
        cpu_percent_avg = self._get_metric_avg('cpu')

        if cpu_percent_min == float('inf'):
            cpu_percent_min = cpu_percent_current
    except (psutil.NoSuchProcess, psutil.AccessDenied):
        cpu_percent_current = cpu_percent_max = 0
        cpu_percent_min = cpu_percent_avg = cpu_time_seconds = 0

    # Disk I/O Metrics (in MB)
    try:
        io_counters = self._process.io_counters()
        disk_read_bytes = io_counters.read_bytes - self._initial_disk_read
        disk_write_bytes = io_counters.write_bytes - self._initial_disk_write

        disk_read_mb = disk_read_bytes / (1024 * 1024)
        disk_write_mb = disk_write_bytes / (1024 * 1024)

        if update_tracking:
            self._update_metric_tracking('disk_read', disk_read_mb)
            self._update_metric_tracking('disk_write', disk_write_mb)

        disk_read_max = self._footprint_metrics['disk_read']['max']
        disk_read_min = self._footprint_metrics['disk_read']['min']
        disk_write_max = self._footprint_metrics['disk_write']['max']
        disk_write_min = self._footprint_metrics['disk_write']['min']

        if disk_read_min == float('inf'):
            disk_read_min = disk_read_mb
        if disk_write_min == float('inf'):
            disk_write_min = disk_write_mb
    except (AttributeError, OSError, psutil.NoSuchProcess, psutil.AccessDenied):
        disk_read_mb = disk_write_mb = 0
        disk_read_max = disk_read_min = disk_write_max = disk_write_min = 0

    # Network I/O Metrics (in MB)
    try:
        net_io = psutil.net_io_counters()
        network_sent_bytes = net_io.bytes_sent - self._initial_network_sent
        network_recv_bytes = net_io.bytes_recv - self._initial_network_recv

        network_sent_mb = network_sent_bytes / (1024 * 1024)
        network_recv_mb = network_recv_bytes / (1024 * 1024)

        if update_tracking:
            self._update_metric_tracking('network_sent', network_sent_mb)
            self._update_metric_tracking('network_recv', network_recv_mb)

        network_sent_max = self._footprint_metrics['network_sent']['max']
        network_sent_min = self._footprint_metrics['network_sent']['min']
        network_recv_max = self._footprint_metrics['network_recv']['max']
        network_recv_min = self._footprint_metrics['network_recv']['min']

        if network_sent_min == float('inf'):
            network_sent_min = network_sent_mb
        if network_recv_min == float('inf'):
            network_recv_min = network_recv_mb
    except (AttributeError, OSError):
        network_sent_mb = network_recv_mb = 0
        network_sent_max = network_sent_min = 0
        network_recv_max = network_recv_min = 0

    # Process Info
    try:
        process_id = self._process.pid
        threads = self._process.num_threads()
        open_files_path = [str(x.path).replace("\\", "/") for x in self._process.open_files()]
        connections_uri = [f"{x.laddr}:{x.raddr} {str(x.status)}" for x in self._process.connections()]

        open_files = len(open_files_path)
        connections = len(connections_uri)
    except (psutil.NoSuchProcess, psutil.AccessDenied):
        process_id = os.getpid()
        threads = open_files = connections = 0
        open_files_path = []
        connections_uri = []

    return FootprintMetrics(
        start_time=self._footprint_start_time,
        uptime_seconds=uptime_seconds,
        uptime_formatted=uptime_formatted,
        memory_current=memory_current,
        memory_max=memory_max,
        memory_min=memory_min,
        memory_percent=memory_percent,
        cpu_percent_current=cpu_percent_current,
        cpu_percent_max=cpu_percent_max,
        cpu_percent_min=cpu_percent_min,
        cpu_percent_avg=cpu_percent_avg,
        cpu_time_seconds=cpu_time_seconds,
        disk_read_mb=disk_read_mb,
        disk_write_mb=disk_write_mb,
        disk_read_max=disk_read_max,
        disk_read_min=disk_read_min,
        disk_write_max=disk_write_max,
        disk_write_min=disk_write_min,
        network_sent_mb=network_sent_mb,
        network_recv_mb=network_recv_mb,
        network_sent_max=network_sent_max,
        network_sent_min=network_sent_min,
        network_recv_max=network_recv_max,
        network_recv_min=network_recv_min,
        process_id=process_id,
        threads=threads,
        open_files=open_files,
        connections=connections,
        open_files_path=open_files_path,
        connections_uri=connections_uri,
    )

fuction_runner(function, function_data, args, kwargs, t0=0.0)

parameters = function_data.get('params') modular_name = function_data.get('module_name') function_name = function_data.get('func_name') mod_function_name = f"{modular_name}.{function_name}"

proxi attr

Source code in toolboxv2/utils/system/types.py

def fuction_runner(self, function, function_data: dict, args: list, kwargs: dict, t0=.0):
    """
    parameters = function_data.get('params')
    modular_name = function_data.get('module_name')
    function_name = function_data.get('func_name')
    mod_function_name = f"{modular_name}.{function_name}"

    proxi attr
    """

generate_openapi_html()

Generiert eine HTML-Datei mit OpenAPI/Swagger UI für API-Routen.

Args:

Source code in toolboxv2/utils/system/types.py

def generate_openapi_html(self):
    """
    Generiert eine HTML-Datei mit OpenAPI/Swagger UI für API-Routen.

    Args:
    """

    # OpenAPI Spec erstellen
    openapi_spec = {
        "openapi": "3.0.0",
        "info": {
            "title": "CloudM API Services",
            "version": "0.1.24",
            "description": "API Documentation für CloudM Email Services",
        },
        "servers": [{"url": "/api", "description": "API Server"}],
        "paths": {},
    }

    # Durch alle Services iterieren
    for service_name, functions in self.functions.items():
        for func_name, func_info in functions.items():
            # Nur API-Funktionen verarbeiten
            if not isinstance(func_info, dict):
                continue
            if not func_info.get("api", False):
                continue

            # Parameter aus der Signatur extrahieren
            params = func_info.get("params", [])
            # 'app' Parameter ausschließen (interner Parameter)
            api_params = [p for p in params if p != "app"]

            # Request Body Schema erstellen
            properties = {}
            required = []

            for param in api_params:
                properties[param] = {
                    "type": "string",
                    "description": f"Parameter: {param}",
                }
                # Prüfen ob Parameter optional ist (hat default value)
                if "=" not in str(func_info.get("signature", "")):
                    required.append(param)

            # API Path erstellen
            path = f"/{service_name}/{func_name}"

            # Path Operation definieren
            openapi_spec["paths"][path] = {
                "post": {
                    "summary": func_name.replace("_", " ").title(),
                    "description": f"Funktion: {func_name} aus Modul {func_info.get('module_name', 'unknown')}",
                    "tags": [service_name],
                    "requestBody": {
                        "required": True,
                        "content": {
                            "application/json": {
                                "schema": {
                                    "type": "object",
                                    "properties": properties,
                                    "required": required,
                                }
                            }
                        },
                    },
                    "responses": {
                        "200": {
                            "description": "Erfolgreiche Antwort",
                            "content": {
                                "application/json": {"schema": {"type": "object"}}
                            },
                        },
                        "400": {"description": "Ungültige Anfrage"},
                        "500": {"description": "Serverfehler"},
                    },
                }
            }

    # HTML Template mit Swagger UI
    html_content = f"""<!DOCTYPE html>
<html lang="de">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>CloudM API Documentation</title>
    <link rel="stylesheet" type="text/css" href="https://cdnjs.cloudflare.com/ajax/libs/swagger-ui/5.10.5/swagger-ui.min.css">
    <style>
        body {{
            margin: 0;
            padding: 0;
        }}
        #swagger-ui {{
            max-width: 1460px;
            margin: 0 auto;
        }}
    </style>
</head>
<body>
    <div id="swagger-ui"></div>

    <script src="https://cdnjs.cloudflare.com/ajax/libs/swagger-ui/5.10.5/swagger-ui-bundle.min.js"></script>
    <script src="https://cdnjs.cloudflare.com/ajax/libs/swagger-ui/5.10.5/swagger-ui-standalone-preset.min.js"></script>
    <script unsave="true">
        const onload = function() {{
            const spec = {json.dumps(openapi_spec, indent=2)};

            window.ui = SwaggerUIBundle({{
                spec: spec,
                dom_id: '#swagger-ui',
                deepLinking: true,
                presets: [
                    SwaggerUIBundle.presets.apis,
                    SwaggerUIStandalonePreset
                ],
                plugins: [
                    SwaggerUIBundle.plugins.DownloadUrl
                ],
                layout: "StandaloneLayout"
            }});
        }};
        if (window.TB?.onLoaded) {{
            window.TB.onLoaded(onload());
        }} else {{
           window.addEventListener('DOMContentLoaded', onload)
        }}
    </script>
</body>
</html>"""
    print(f"✓ Gefundene API-Routen: {len(openapi_spec['paths'])}")
    return Result.html(html_content, row=True)

get_all_mods(working_dir='mods', path_to='./runtime')

proxi attr

Source code in toolboxv2/utils/system/types.py

def get_all_mods(self, working_dir="mods", path_to="./runtime"):
    """proxi attr"""

get_autocompletion_dict()

proxi attr

Source code in toolboxv2/utils/system/types.py

def get_autocompletion_dict(self):
    """proxi attr"""

get_function(name, **kwargs)

Kwargs for _get_function metadata:: return the registered function dictionary stateless: (function_data, None), 0 stateful: (function_data, higher_order_function), 0 state::boolean specification::str default app

Source code in toolboxv2/utils/system/types.py

def get_function(self, name: Enum or tuple, **kwargs):
    """
    Kwargs for _get_function
        metadata:: return the registered function dictionary
            stateless: (function_data, None), 0
            stateful: (function_data, higher_order_function), 0
        state::boolean
            specification::str default app
    """

get_mod(name, spec='app')

proxi attr

Source code in toolboxv2/utils/system/types.py

def get_mod(self, name, spec='app') -> ModuleType or MainToolType:
    """proxi attr"""

get_task_context(files, intent) async

mkdocs system [extra] Get optimized context for a specific editing task.

Parameters:

Name	Type	Description	Default
files	List[str]	List of file paths relevant to the task.	required
intent	str	Description of what the user wants to do (e.g., "Add logging to auth").	required

Returns:

Type	Description
dict	ContextBundle dictionary ready for LLM injection.

Source code in toolboxv2/utils/system/types.py

async def get_task_context(self, files: List[str], intent: str) -> dict:
    """mkdocs system [extra]
    Get optimized context for a specific editing task.

    Args:
        files: List of file paths relevant to the task.
        intent: Description of what the user wants to do (e.g., "Add logging to auth").

    Returns:
        ContextBundle dictionary ready for LLM injection.
    """

get_username(get_input=False, default='loot')

proxi attr

Source code in toolboxv2/utils/system/types.py

def get_username(self, get_input=False, default="loot") -> str:
    """proxi attr"""

hide_console(*args, **kwargs) async staticmethod

proxi attr

Source code in toolboxv2/utils/system/types.py

@staticmethod
async def hide_console(*args, **kwargs):
    """proxi attr"""

inplace_load_instance(mod_name, loc='toolboxv2.mods.', spec='app', save=True)

proxi attr

Source code in toolboxv2/utils/system/types.py

def inplace_load_instance(self, mod_name, loc="toolboxv2.mods.", spec='app', save=True):
    """proxi attr"""

load_all_mods_in_file(working_dir='mods') async

proxi attr

Source code in toolboxv2/utils/system/types.py

async def load_all_mods_in_file(self, working_dir="mods"):
    """proxi attr"""

load_external_mods() async

proxi attr

Source code in toolboxv2/utils/system/types.py

async def load_external_mods(self):
    """proxi attr"""

load_mod(mod_name, mlm='I', **kwargs)

proxi attr

Source code in toolboxv2/utils/system/types.py

def load_mod(self, mod_name: str, mlm='I', **kwargs):
    """proxi attr"""

mod_online(mod_name, installed=False)

proxi attr

Source code in toolboxv2/utils/system/types.py

def mod_online(self, mod_name, installed=False):
    """proxi attr"""

print(text, *args, **kwargs) staticmethod

proxi attr

Source code in toolboxv2/utils/system/types.py

@staticmethod
def print(text, *args, **kwargs):
    """proxi attr"""

print_footprint(detailed=True)

Gibt den Footprint formatiert aus.

Parameters:

Name	Type	Description	Default
detailed	bool	Wenn True, zeigt alle Details, sonst nur Zusammenfassung	True

Returns:

Type	Description
str	Formatierter Footprint-String

Source code in toolboxv2/utils/system/types.py

def print_footprint(self, detailed: bool = True) -> str:
    """
    Gibt den Footprint formatiert aus.

    Args:
        detailed: Wenn True, zeigt alle Details, sonst nur Zusammenfassung

    Returns:
        Formatierter Footprint-String
    """
    metrics = self.footprint()

    output = [
        "=" * 70,
        f"TOOLBOX FOOTPRINT - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
        "=" * 70,
        f"\n📊 UPTIME",
        f"  Runtime: {metrics.uptime_formatted}",
        f"  Seconds: {metrics.uptime_seconds:.2f}s",
        f"\n💾 MEMORY USAGE",
        f"  Current:  {metrics.memory_current:.2f} MB ({metrics.memory_percent:.2f}%)",
        f"  Maximum:  {metrics.memory_max:.2f} MB",
        f"  Minimum:  {metrics.memory_min:.2f} MB",
    ]

    if detailed:
        helper_ = '\n\t- '.join(metrics.open_files_path)
        helper__ = '\n\t- '.join(metrics.connections_uri)
        output.extend([
            f"\n⚙️  CPU USAGE",
            f"  Current:  {metrics.cpu_percent_current:.2f}%",
            f"  Maximum:  {metrics.cpu_percent_max:.2f}%",
            f"  Minimum:  {metrics.cpu_percent_min:.2f}%",
            f"  Average:  {metrics.cpu_percent_avg:.2f}%",
            f"  CPU Time: {metrics.cpu_time_seconds:.2f}s",
            f"\n💿 DISK I/O",
            f"  Read:     {metrics.disk_read_mb:.2f} MB (Max: {metrics.disk_read_max:.2f}, Min: {metrics.disk_read_min:.2f})",
            f"  Write:    {metrics.disk_write_mb:.2f} MB (Max: {metrics.disk_write_max:.2f}, Min: {metrics.disk_write_min:.2f})",
            f"\n🌐 NETWORK I/O",
            f"  Sent:     {metrics.network_sent_mb:.2f} MB (Max: {metrics.network_sent_max:.2f}, Min: {metrics.network_sent_min:.2f})",
            f"  Received: {metrics.network_recv_mb:.2f} MB (Max: {metrics.network_recv_max:.2f}, Min: {metrics.network_recv_min:.2f})",
            f"\n🔧 PROCESS INFO",
            f"  PID:         {metrics.process_id}",
            f"  Threads:     {metrics.threads}",
            f"\n📂 OPEN FILES",
            f"  Open Files:  {metrics.open_files}",
            f"  Open Files Path: \n\t- {helper_}",
            f"\n🔗 NETWORK CONNECTIONS",
            f"  Connections: {metrics.connections}",
            f"  Connections URI: \n\t- {helper__}",
        ])

    output.append("=" * 70)

    return "\n".join(output)

print_ok()

proxi attr

Source code in toolboxv2/utils/system/types.py

def print_ok(self):
    """proxi attr"""
    self.logger.info("OK")

reload_mod(mod_name, spec='app', is_file=True, loc='toolboxv2.mods.')

proxi attr

Source code in toolboxv2/utils/system/types.py

def reload_mod(self, mod_name, spec='app', is_file=True, loc="toolboxv2.mods."):
    """proxi attr"""

remove_mod(mod_name, spec='app', delete=True)

proxi attr

Source code in toolboxv2/utils/system/types.py

def remove_mod(self, mod_name, spec='app', delete=True):
    """proxi attr"""

rrun_flows(name, **kwargs)

proxi attr

Source code in toolboxv2/utils/system/types.py

def rrun_flows(self, name, **kwargs):
    """proxi attr"""

run_a_from_sync(function, *args)

run a async fuction

Source code in toolboxv2/utils/system/types.py

def run_a_from_sync(self, function, *args):
    """
    run a async fuction
    """

run_any(mod_function_name, backwords_compability_variabel_string_holder=None, get_results=False, tb_run_function_with_state=True, tb_run_with_specification='app', args_=None, kwargs_=None, *args, **kwargs)

proxi attr

Source code in toolboxv2/utils/system/types.py

def run_any(self, mod_function_name: Enum or str or tuple, backwords_compability_variabel_string_holder=None,
            get_results=False, tb_run_function_with_state=True, tb_run_with_specification='app', args_=None,
            kwargs_=None,
            *args, **kwargs):
    """proxi attr"""

run_bg_task(task)

run a async fuction

Source code in toolboxv2/utils/system/types.py

def run_bg_task(self, task):
    """
            run a async fuction
            """

run_bg_task_advanced(task, *args, **kwargs)

proxi attr

Source code in toolboxv2/utils/system/types.py

def run_bg_task_advanced(self, task, *args, **kwargs):
    """
    proxi attr
    """

run_flows(name, **kwargs) async

proxi attr

Source code in toolboxv2/utils/system/types.py

async def run_flows(self, name, **kwargs):
    """proxi attr"""

run_function(mod_function_name, tb_run_function_with_state=True, tb_run_with_specification='app', args_=None, kwargs_=None, *args, **kwargs)

proxi attr

Source code in toolboxv2/utils/system/types.py

def run_function(self, mod_function_name: Enum or tuple,
                 tb_run_function_with_state=True,
                 tb_run_with_specification='app',
                 args_=None,
                 kwargs_=None,
                 *args,
                 **kwargs) -> Result:

    """proxi attr"""

run_http(mod_function_name, function_name=None, method='GET', args_=None, kwargs_=None, *args, **kwargs) async

run a function remote via http / https

Source code in toolboxv2/utils/system/types.py

async def run_http(
    self,
    mod_function_name: Enum or str or tuple,
    function_name=None,
    method="GET",
    args_=None,
    kwargs_=None,
    *args,
    **kwargs,
):
    """run a function remote via http / https"""

save_autocompletion_dict()

proxi attr

Source code in toolboxv2/utils/system/types.py

def save_autocompletion_dict(self):
    """proxi attr"""

save_exit()

proxi attr

Source code in toolboxv2/utils/system/types.py

def save_exit(self):
    """proxi attr"""

save_initialized_module(tools_class, spec)

proxi attr

Source code in toolboxv2/utils/system/types.py

def save_initialized_module(self, tools_class, spec):
    """proxi attr"""

save_instance(instance, modular_id, spec='app', instance_type='file/application', tools_class=None)

proxi attr

Source code in toolboxv2/utils/system/types.py

def save_instance(self, instance, modular_id, spec='app', instance_type="file/application", tools_class=None):
    """proxi attr"""

save_load(modname, spec='app')

proxi attr

Source code in toolboxv2/utils/system/types.py

def save_load(self, modname, spec='app'):
    """proxi attr"""

save_registry_as_enums(directory, filename)

proxi attr

Source code in toolboxv2/utils/system/types.py

def save_registry_as_enums(self, directory: str, filename: str):
    """proxi attr"""

set_flows(r)

proxi attr

Source code in toolboxv2/utils/system/types.py

def set_flows(self, r):
    """proxi attr"""

set_logger(debug=False, logger_prefix=None)

proxi attr

Source code in toolboxv2/utils/system/types.py

def set_logger(self, debug=False, logger_prefix=None):
    """proxi attr"""

show_console(*args, **kwargs) async staticmethod

proxi attr

Source code in toolboxv2/utils/system/types.py

@staticmethod
async def show_console(*args, **kwargs):
    """proxi attr"""

sprint(text, *args, **kwargs) staticmethod

proxi attr

Source code in toolboxv2/utils/system/types.py

@staticmethod
def sprint(text, *args, **kwargs):
    """proxi attr"""

tb(name=None, mod_name='', helper='', version=None, test=True, restrict_in_virtual_mode=False, api=False, initial=False, exit_f=False, test_only=False, memory_cache=False, file_cache=False, row=False, request_as_kwarg=False, state=None, level=0, memory_cache_max_size=100, memory_cache_ttl=300, samples=None, interface=None, pre_compute=None, post_compute=None, api_methods=None, websocket_handler=None, websocket_context=False)

A decorator for registering and configuring functions within a module.

This decorator is used to wrap functions with additional functionality such as caching, API conversion, and lifecycle management (initialization and exit). It also handles the registration of the function in the module's function registry.

Parameters:

Name	Type	Description	Default
name	str	The name to register the function under. Defaults to the function's own name.	None
mod_name	str	The name of the module the function belongs to.	''
helper	str	A helper string providing additional information about the function.	''
version	str or None	The version of the function or module.	None
test	bool	Flag to indicate if the function is for testing purposes.	True
restrict_in_virtual_mode	bool	Flag to restrict the function in virtual mode.	False
api	bool	Flag to indicate if the function is part of an API.	False
initial	bool	Flag to indicate if the function should be executed at initialization.	False
exit_f	bool	Flag to indicate if the function should be executed at exit.	False
test_only	bool	Flag to indicate if the function should only be used for testing.	False
memory_cache	bool	Flag to enable memory caching for the function.	False
request_as_kwarg	bool	Flag to get request if the fuction is calld from api.	False
file_cache	bool	Flag to enable file caching for the function.	False
row	bool	rather to auto wrap the result in Result type default False means no row data aka result type	False
state	bool or None	Flag to indicate if the function maintains state.	None
level	int	The level of the function, used for prioritization or categorization.	0
memory_cache_max_size	int	Maximum size of the memory cache.	100
memory_cache_ttl	int	Time-to-live for the memory cache entries.	300
samples	list or dict or None	Samples or examples of function usage.	None
interface	str	The interface type for the function.	None
pre_compute	callable	A function to be called before the main function.	None
post_compute	callable	A function to be called after the main function.	None
api_methods	list[str]	default ["AUTO"] (GET if not params, POST if params) , GET, POST, PUT or DELETE.	None

Returns:

Name	Type	Description
function		The decorated function with additional processing and registration capabilities.

Source code in toolboxv2/utils/system/types.py

def tb(self, name=None,
       mod_name: str = "",
       helper: str = "",
       version: str or None = None,
       test: bool = True,
       restrict_in_virtual_mode: bool = False,
       api: bool = False,
       initial: bool = False,
       exit_f: bool = False,
       test_only: bool = False,
       memory_cache: bool = False,
       file_cache: bool = False,
       row=False,
       request_as_kwarg: bool = False,
       state: bool or None = None,
       level: int = 0,
       memory_cache_max_size: int = 100,
       memory_cache_ttl: int = 300,
       samples: list or dict or None = None,
       interface: ToolBoxInterfaces or None or str = None,
       pre_compute=None,
       post_compute=None,
       api_methods=None,
       websocket_handler: str | None = None,
       websocket_context: bool = False,
       ):
    """
A decorator for registering and configuring functions within a module.

This decorator is used to wrap functions with additional functionality such as caching, API conversion, and lifecycle management (initialization and exit). It also handles the registration of the function in the module's function registry.

Args:
    name (str, optional): The name to register the function under. Defaults to the function's own name.
    mod_name (str, optional): The name of the module the function belongs to.
    helper (str, optional): A helper string providing additional information about the function.
    version (str or None, optional): The version of the function or module.
    test (bool, optional): Flag to indicate if the function is for testing purposes.
    restrict_in_virtual_mode (bool, optional): Flag to restrict the function in virtual mode.
    api (bool, optional): Flag to indicate if the function is part of an API.
    initial (bool, optional): Flag to indicate if the function should be executed at initialization.
    exit_f (bool, optional): Flag to indicate if the function should be executed at exit.
    test_only (bool, optional): Flag to indicate if the function should only be used for testing.
    memory_cache (bool, optional): Flag to enable memory caching for the function.
    request_as_kwarg (bool, optional): Flag to get request if the fuction is calld from api.
    file_cache (bool, optional): Flag to enable file caching for the function.
    row (bool, optional): rather to auto wrap the result in Result type default False means no row data aka result type
    state (bool or None, optional): Flag to indicate if the function maintains state.
    level (int, optional): The level of the function, used for prioritization or categorization.
    memory_cache_max_size (int, optional): Maximum size of the memory cache.
    memory_cache_ttl (int, optional): Time-to-live for the memory cache entries.
    samples (list or dict or None, optional): Samples or examples of function usage.
    interface (str, optional): The interface type for the function.
    pre_compute (callable, optional): A function to be called before the main function.
    post_compute (callable, optional): A function to be called after the main function.
    api_methods (list[str], optional): default ["AUTO"] (GET if not params, POST if params) , GET, POST, PUT or DELETE.

Returns:
    function: The decorated function with additional processing and registration capabilities.
"""
    if interface is None:
        interface = "tb"
    if test_only and 'test' not in self.id:
        return lambda *args, **kwargs: args
    return self._create_decorator(
        interface,
        name,
        mod_name,
        version=version,
        test=test,
        restrict_in_virtual_mode=restrict_in_virtual_mode,
        api=api,
        initial=initial,
        exit_f=exit_f,
        test_only=test_only,
        memory_cache=memory_cache,
        file_cache=file_cache,
        row=row,
        request_as_kwarg=request_as_kwarg,
        state=state,
        level=level,
        memory_cache_max_size=memory_cache_max_size,
        memory_cache_ttl=memory_cache_ttl,
        samples=samples,
        interface=interface,
        pre_compute=pre_compute,
        post_compute=post_compute,
        api_methods=api_methods,
        websocket_handler=websocket_handler,
        websocket_context=websocket_context,
    )

wait_for_bg_tasks(timeout=None)

proxi attr

Source code in toolboxv2/utils/system/types.py

def wait_for_bg_tasks(self, timeout=None):
    """
    proxi attr
    """

watch_mod(mod_name, spec='app', loc='toolboxv2.mods.', use_thread=True, path_name=None)

proxi attr

Source code in toolboxv2/utils/system/types.py

def watch_mod(self, mod_name, spec='app', loc="toolboxv2.mods.", use_thread=True, path_name=None):
    """proxi attr"""

web_context()

returns the build index ( toolbox web component )

Source code in toolboxv2/utils/system/types.py

def web_context(self) -> str:
    """returns the build index ( toolbox web component )"""

toolboxv2.MainTool

Source code in toolboxv2/utils/system/main_tool.py

class MainTool:
    toolID: str = ""
    # app = None
    interface = None
    spec = "app"
    name = ""
    color = "Bold"
    stuf = False # Shut The Up fuck

    def __init__(self, *args, **kwargs):
        """
        Standard constructor used for arguments pass
        Do not override. Use __ainit__ instead
        """
        self.__storedargs = args, kwargs
        self.tools = kwargs.get("tool", {})
        self.logger = kwargs.get("logs", get_logger())
        self.color = kwargs.get("color", "WHITE")
        self.todo = kwargs.get("load", kwargs.get("on_start", lambda: None))
        if "on_exit" in kwargs and isinstance(kwargs.get("on_exit"), Callable):
            self.on_exit =self.app.tb(
                mod_name=self.name,
                name=kwargs.get("on_exit").__name__,
                version=self.version if hasattr(self, 'version') else "x.x.x",
            )(kwargs.get("on_exit"))
        self.async_initialized = False
        if self.todo:
            try:
                if inspect.iscoroutinefunction(self.todo):
                    pass
                else:
                    self.todo()
                get_logger().info(f"{self.name} on load suspended")
            except Exception as e:
                get_logger().error(f"Error loading mod {self.name} {e}")
                if self.app.debug:
                    import traceback
                    traceback.print_exc()
        else:
            get_logger().info(f"{self.name} no load require")

    async def __ainit__(self, *args, **kwargs):
        self.version = kwargs.get("v", kwargs.get("version", "x.x.x"))
        self.tools = kwargs.get("tool", {})
        self.name = kwargs["name"]
        self.logger = kwargs.get("logs", get_logger())
        self.color = kwargs.get("color", "WHITE")
        self.todo = kwargs.get("load", kwargs.get("on_start"))
        if not hasattr(self, 'config'):
            self.config = {}
        self.user = None
        self.description = "A toolbox mod" if kwargs.get("description") is None else kwargs.get("description")
        if MainTool.interface is None:
            MainTool.interface = self.app.interface_type
        # Result.default(self.app.interface)

        if self.todo:
            try:
                if inspect.iscoroutinefunction(self.todo):
                    await self.todo()
                else:
                    pass
                await asyncio.sleep(0.1)
                get_logger().info(f"{self.name} on load a suspended")
            except Exception as e:
                get_logger().error(f"Error loading mod {self.name} {e}")
                if self.app.debug:
                    import traceback
                    traceback.print_exc()
        else:
            get_logger().info(f"{self.name} no load require")
        self.app.print(f"TOOL : {self.spec}.{self.name} online")



    @property
    def app(self):
        return get_app(
            from_=f"{self.spec}.{self.name}|{self.toolID if self.toolID else '*' + MainTool.toolID} {self.interface if self.interface else MainTool.interface}")

    @app.setter
    def app(self, v):
        raise PermissionError(f"You cannot set the App Instance! {v=}")

    @staticmethod
    def return_result(error: ToolBoxError = ToolBoxError.none,
                      exec_code: int = 0,
                      help_text: str = "",
                      data_info=None,
                      data=None,
                      data_to=None):

        if data_to is None:
            data_to = MainTool.interface if MainTool.interface is not None else ToolBoxInterfaces.cli

        if data is None:
            data = {}

        if data_info is None:
            data_info = {}

        return Result(
            error,
            ToolBoxResult(data_info=data_info, data=data, data_to=data_to),
            ToolBoxInfo(exec_code=exec_code, help_text=help_text)
        )

    def print(self, message, end="\n", **kwargs):
        if self.stuf:
            return

        self.app.print(Style.style_dic[self.color] + self.name + Style.style_dic["END"] + ":", message, end=end,
                       **kwargs)

    def add_str_to_config(self, command):
        if len(command) != 2:
            self.logger.error('Invalid command must be key value')
            return False
        self.config[command[0]] = command[1]

    def webInstall(self, user_instance, construct_render) -> str:
        """"Returns a web installer for the given user instance and construct render template"""

    def get_version(self) -> str:
        """"Returns the version"""
        return self.version

    async def get_user(self, username: str) -> Result:
        return await self.app.a_run_any(CLOUDM_AUTHMANAGER.GET_USER_BY_NAME, username=username, get_results=True)

    async def __initobj(self):
        """Crutch used for __await__ after spawning"""
        assert not self.async_initialized
        self.async_initialized = True
        # pass the parameters to __ainit__ that passed to __init__
        await self.__ainit__(*self.__storedargs[0], **self.__storedargs[1])
        return self

    def __await__(self):
        return self.__initobj().__await__()

get_version()

"Returns the version

Source code in toolboxv2/utils/system/main_tool.py

def get_version(self) -> str:
    """"Returns the version"""
    return self.version

webInstall(user_instance, construct_render)

"Returns a web installer for the given user instance and construct render template

Source code in toolboxv2/utils/system/main_tool.py

def webInstall(self, user_instance, construct_render) -> str:
    """"Returns a web installer for the given user instance and construct render template"""

toolboxv2.get_app(from_=None, name=None, args=AppArgs().default(), app_con=None, sync=False)

Source code in toolboxv2/utils/system/getting_and_closing_app.py

def get_app(from_=None, name=None, args=AppArgs().default(), app_con=None, sync=False) -> AppType:
    global registered_apps

    # print(f"get app requested from: {from_} withe name: {name}")
    logger = get_logger()
    caller_src = "set debug mod"
    if from_ is None and logger.level >= logging.DEBUG:
        from inspect import getouterframes, currentframe
        caller_src = f"{getouterframes(currentframe(), 2)[1].filename}::{getouterframes(currentframe(), 2)[1].lineno}"
    logger.info(Style.GREYBG(f"get app requested from: {from_ if from_ is not None else caller_src}"))
    if registered_apps[0] is not None:
        return registered_apps[0]

    # Fail-safe: someone called get_app() straight from their own code/menu
    # before any onboarding ran. If no manifest exists, prepare 'mini headless'
    # (secret + offline env + mini manifest) so the App boots cleanly and login
    # works. Non-recursive: prepare only, this function builds the App below.
    try:
        from toolboxv2.init_onboarding import prepare_mini_failsafe
        prepare_mini_failsafe()
    except Exception:
        pass  # never block app creation on onboarding prep

    if app_con is None:
        try:
            from ... import App
        except ImportError:
            try:
                from ..toolbox import App
            except ImportError:
                from toolboxv2 import App

        app_con = App
    app = app_con(name, args=args) if name else app_con()
    registered_apps[0] = app
    return app

System Utilities & Configuration

toolboxv2.FileHandler = FileHandlerV2 module-attribute

toolboxv2.show_console = _guarded_import('toolboxv2.utils.extras.show_and_hide_console', 'show_console', feature='desktop') module-attribute

Logging

toolboxv2.get_logger()

Return the active toolboxv2 logger. Same API as before – but now backed by JSON formatting + MobileDB.

Source code in toolboxv2/utils/system/tb_logger.py

def get_logger() -> logging.Logger:
    """
    Return the active toolboxv2 logger.
    Same API as before – but now backed by JSON formatting + MobileDB.
    """
    return logging.getLogger(loggerNameOfToolboxv2)

toolboxv2.setup_logging(level, name=loggerNameOfToolboxv2, online_level=None, is_online=False, file_level=None, interminal=False, logs_directory=None, app_name='main')

Unified logger setup.

Keeps the old (logger, filename) return signature so every existing call site stays untouched, but internally: - Console + File handlers use JsonLogFormatter (JSONL) - If a MobileDB was registered via set_log_db(), a MobileDBLogHandler is attached automatically. - SocketHandler still works for is_online=True.

Returns:

Type	Description
Tuple[Logger, str]	(logger, log_filename)

Source code in toolboxv2/utils/system/tb_logger.py

def setup_logging(
    level: int,
    name: str = loggerNameOfToolboxv2,
    online_level: Optional[int] = None,
    is_online: bool = False,
    file_level: Optional[int] = None,
    interminal: bool = False,
    logs_directory: str|None = None,
    app_name: str = "main",
) -> Tuple[logging.Logger, str]:
    """
    Unified logger setup.

    Keeps the old (logger, filename) return signature so every existing
    call site stays untouched, but internally:
      - Console + File handlers use JsonLogFormatter (JSONL)
      - If a MobileDB was registered via set_log_db(), a
        MobileDBLogHandler is attached automatically.
      - SocketHandler still works for is_online=True.

    Returns:
        (logger, log_filename)
    """
    global loggerNameOfToolboxv2, _app_id

    if logs_directory is None:
        from toolboxv2 import tb_root_dir
        logs_directory = str(tb_root_dir.parent / "logs")

    if not online_level:
        online_level = level
    if not file_level:
        file_level = level

    loggerNameOfToolboxv2 = name
    _app_id = app_name

    # ---- Log file rotation (unchanged logic) ----
    if not os.path.exists(logs_directory):
        os.makedirs(logs_directory, exist_ok=True)
    if not os.path.exists(os.path.join(logs_directory , "Logs.info")):
        open(os.path.join(f"{logs_directory}","Logs.info"), "a").close()

    available_log_levels = [
        logging.CRITICAL, logging.FATAL, logging.ERROR, logging.WARNING,
        logging.WARN, logging.INFO, logging.DEBUG, logging.NOTSET,
    ]
    for lbl, val in [("level", level), ("online_level", online_level), ("file_level", file_level)]:
        if val not in available_log_levels:
            raise ValueError(f"{lbl} must be one of {available_log_levels}, but is {val}")

    log_date = datetime.datetime.today().strftime('%Y-%m-%d')
    log_levels_names = ["CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG", "NOTSET"]
    log_level_index = log_levels_names.index(logging.getLevelName(level))

    filename = f"Logs-{name}-{log_date}-{log_levels_names[log_level_index]}"
    log_filename = os.path.join(f"{logs_directory}",f"{filename}.log")

    log_info_data: Dict[str, Any] = {filename: 0, "H": "localhost", "P": 62435}

    with open(os.path.join(f"{logs_directory}","Logs.info")) as li:
        log_info_data_str = li.read()
        try:
            log_info_data = eval(log_info_data_str)
        except SyntaxError:
            if log_info_data_str:
                print(Style.RED(Style.Bold("Could not parse log info data")))

        if filename not in log_info_data:
            log_info_data[filename] = 0
        if not os.path.exists(log_filename):
            log_info_data[filename] = 0
        if os.path.exists(log_filename):
            log_info_data[filename] += 1
            while os.path.exists(os.path.join(f"{logs_directory}",f"{filename}#{log_info_data[filename]}.log")):
                log_info_data[filename] += 1
            try:
                os.rename(log_filename, os.path.join(f"{logs_directory}",f"{filename}#{log_info_data[filename]}.log"))
            except PermissionError and FileNotFoundError and BaseException:
                pass
    with open(os.path.join(f"{logs_directory}","Logs.info"), "w") as li:
        if len(log_info_data.keys()) >= 7:
            log_info_data = {
                filename: log_info_data[filename],
                "H": log_info_data["H"],
                "P": log_info_data["P"],
            }
        li.write(str(log_info_data))

    try:
        with open(log_filename, "a"):
            pass
    except OSError:
        log_filename = os.path.join(f"{logs_directory}",f"Logs-Test-{log_date}-{log_levels_names[log_level_index]}.log")
        with open(log_filename, "a"):
            pass

    # ---- Configure logger with new JSON formatters ----

    logger = logging.getLogger(name)
    logger.setLevel(level)
    if logger.hasHandlers():
        logger.handlers.clear()
    logger.propagate = False

    json_formatter = JsonLogFormatter(app_id=app_name, node_id=_node_id)

    # File handler – JSONL output (machine-parseable, replaces old plain text)
    file_handler = logging.FileHandler(log_filename)
    file_handler.setFormatter(json_formatter)
    file_handler.setLevel(file_level)
    logger.addHandler(file_handler)

    # Console handler
    if interminal:
        console_handler = logging.StreamHandler()
        log_format = '%(asctime)s - %(name)s - %(levelname)s - %(filename)s - %(funcName)s:%(lineno)d - %(message)s'
        console_handler.setFormatter(logging.Formatter(log_format))
        console_handler.setLevel(level)
        logger.addHandler(console_handler)

    # Socket handler (legacy remote logging)
    if is_online:
        socket_handler = SocketHandler(log_info_data["H"], log_info_data["P"])
        socket_handler.setFormatter(json_formatter)
        socket_handler.setLevel(online_level)
        logger.addHandler(socket_handler)

    # MobileDB handler (encrypted, offline-first, sync-ready)
    if _log_db is not None:
        db_handler = MobileDBLogHandler(db=_log_db, node_id=_node_id)
        db_handler.setFormatter(json_formatter)
        db_handler.setLevel(level)
        logger.addHandler(db_handler)

    return logger, filename

Styling & Console Output

toolboxv2.Style

Source code in toolboxv2/utils/extras/Style.py

class Style:
    _END = '\33[0m'
    _BLACK = '\33[30m'
    _RED = '\33[31m'
    _GREEN = '\33[32m'
    _YELLOW = '\33[33m'
    _BLUE = '\33[34m'
    _MAGENTA = '\33[35m'
    _CYAN = '\33[36m'
    _WHITE = '\33[37m'

    _Bold = '\33[1m'
    _ITALIC = '\33[3m'
    _Underline = '\33[4m'
    _BLINK = '\33[5m'
    _BLINK2 = '\33[6m'
    _Reversed = '\33[7m'

    _BLACKBG = '\33[40m'
    _REDBG = '\33[41m'
    _GREENBG = '\33[42m'
    _YELLOWBG = '\33[43m'
    _BLUEBG = '\33[44m'
    _VIOLETBG = '\33[45m'
    _BEIGEBG = '\33[46m'
    _WHITEBG = '\33[47m'

    _GREY = '\33[90m'
    _RED2 = '\33[91m'
    _GREEN2 = '\33[92m'
    _YELLOW2 = '\33[93m'
    _BLUE2 = '\33[94m'
    _VIOLET2 = '\33[95m'
    _BEIGE2 = '\33[96m'
    _WHITE2 = '\33[97m'

    _GREYBG = '\33[100m'
    _REDBG2 = '\33[101m'
    _GREENBG2 = '\33[102m'
    _YELLOWBG2 = '\33[103m'
    _BLUEBG2 = '\33[104m'
    _VIOLETBG2 = '\33[105m'
    _BEIGEBG2 = '\33[106m'
    _WHITEBG2 = '\33[107m'

    style_dic = {
        "END": _END,
        "BLACK": _BLACK,
        "RED": _RED,
        "GREEN": _GREEN,
        "YELLOW": _YELLOW,
        "BLUE": _BLUE,
        "MAGENTA": _MAGENTA,
        "CYAN": _CYAN,
        "WHITE": _WHITE,
        "Bold": _Bold,
        "Underline": _Underline,
        "Reversed": _Reversed,

        "ITALIC": _ITALIC,
        "BLINK": _BLINK,
        "BLINK2": _BLINK2,
        "BLACKBG": _BLACKBG,
        "REDBG": _REDBG,
        "GREENBG": _GREENBG,
        "YELLOWBG": _YELLOWBG,
        "BLUEBG": _BLUEBG,
        "VIOLETBG": _VIOLETBG,
        "BEIGEBG": _BEIGEBG,
        "WHITEBG": _WHITEBG,
        "GRAY": _GREY,
        "GREY": _GREY,
        "RED2": _RED2,
        "GREEN2": _GREEN2,
        "YELLOW2": _YELLOW2,
        "BLUE2": _BLUE2,
        "VIOLET2": _VIOLET2,
        "BEIGE2": _BEIGE2,
        "WHITE2": _WHITE2,
        "GREYBG": _GREYBG,
        "REDBG2": _REDBG2,
        "GREENBG2": _GREENBG2,
        "YELLOWBG2": _YELLOWBG2,
        "BLUEBG2": _BLUEBG2,
        "VIOLETBG2": _VIOLETBG2,
        "BEIGEBG2": _BEIGEBG2,
        "WHITEBG2": _WHITEBG2,

    }

    @staticmethod
    @text_save
    def END_():
        print(Style._END)

    @staticmethod
    @text_save
    def GREEN_():
        print(Style._GREEN)

    @staticmethod
    @text_save
    def BLUE(text: str):
        return Style._BLUE + text + Style._END

    @staticmethod
    @text_save
    def BLACK(text: str):
        return Style._BLACK + text + Style._END

    @staticmethod
    @text_save
    def RED(text: str):
        return Style._RED + text + Style._END

    @staticmethod
    @text_save
    def GREEN(text: str):
        return Style._GREEN + text + Style._END

    @staticmethod
    @text_save
    def YELLOW(text: str):
        return Style._YELLOW + text + Style._END

    @staticmethod
    @text_save
    def MAGENTA(text: str):
        return Style._MAGENTA + text + Style._END

    @staticmethod
    @text_save
    def CYAN(text: str):
        return Style._CYAN + text + Style._END

    @staticmethod
    @text_save
    def WHITE(text: str):
        return Style._WHITE + text + Style._END

    @staticmethod
    @text_save
    def BOLD(text: str):
        return Style._Bold + text + Style._END

    @staticmethod
    @text_save
    def Bold(text: str):
        return Style._Bold + text + Style._END

    @staticmethod
    @text_save
    def UNDERLINE(text: str):
        return Style._Underline + text + Style._END
    @staticmethod
    @text_save
    def Underline(text: str):
        return Style._Underline + text + Style._END

    @staticmethod
    @text_save
    def Underlined(text: str):
        return Style._Underline + text + Style._END

    @staticmethod
    @text_save
    def Reversed(text: str):
        return Style._Reversed + text + Style._END

    @staticmethod
    @text_save
    def ITALIC(text: str):
        return Style._ITALIC + text + Style._END

    @staticmethod
    @text_save
    def BLINK(text: str):
        return Style._BLINK + text + Style._END

    @staticmethod
    @text_save
    def BLINK2(text: str):
        return Style._BLINK2 + text + Style._END

    @staticmethod
    @text_save
    def BLACKBG(text: str):
        return Style._BLACKBG + text + Style._END

    @staticmethod
    @text_save
    def REDBG(text: str):
        return Style._REDBG + text + Style._END

    @staticmethod
    @text_save
    def GREENBG(text: str):
        return Style._GREENBG + text + Style._END

    @staticmethod
    @text_save
    def YELLOWBG(text: str):
        return Style._YELLOWBG + text + Style._END

    @staticmethod
    @text_save
    def BLUEBG(text: str):
        return Style._BLUEBG + text + Style._END

    @staticmethod
    @text_save
    def VIOLETBG(text: str):
        return Style._VIOLETBG + text + Style._END

    @staticmethod
    @text_save
    def BEIGEBG(text: str):
        return Style._BEIGEBG + text + Style._END

    @staticmethod
    @text_save
    def WHITEBG(text: str):
        return Style._WHITEBG + text + Style._END

    @staticmethod
    @text_save
    def GREY(text: str):
        return Style._GREY + str(text) + Style._END

    @staticmethod
    @text_save
    def RED2(text: str):
        return Style._RED2 + text + Style._END

    @staticmethod
    @text_save
    def GREEN2(text: str):
        return Style._GREEN2 + text + Style._END

    @staticmethod
    @text_save
    def YELLOW2(text: str):
        return Style._YELLOW2 + text + Style._END

    @staticmethod
    @text_save
    def BLUE2(text: str):
        return Style._BLUE2 + text + Style._END

    @staticmethod
    @text_save
    def VIOLET2(text: str):
        return Style._VIOLET2 + text + Style._END

    @staticmethod
    @text_save
    def BEIGE2(text: str):
        return Style._BEIGE2 + text + Style._END

    @staticmethod
    @text_save
    def WHITE2(text: str):
        return Style._WHITE2 + text + Style._END

    @staticmethod
    @text_save
    def GREYBG(text: str):
        return Style._GREYBG + text + Style._END

    @staticmethod
    @text_save
    def REDBG2(text: str):
        return Style._REDBG2 + text + Style._END

    @staticmethod
    @text_save
    def GREENBG2(text: str):
        return Style._GREENBG2 + text + Style._END

    @staticmethod
    @text_save
    def YELLOWBG2(text: str):
        return Style._YELLOWBG2 + text + Style._END

    @staticmethod
    @text_save
    def BLUEBG2(text: str):
        return Style._BLUEBG2 + text + Style._END

    @staticmethod
    @text_save
    def VIOLETBG2(text: str):
        return Style._VIOLETBG2 + text + Style._END

    @staticmethod
    @text_save
    def BEIGEBG2(text: str):
        return Style._BEIGEBG2 + text + Style._END

    @staticmethod
    @text_save
    def WHITEBG2(text: str):
        return Style._WHITEBG2 + text + Style._END

    @staticmethod
    @text_save
    def loading_al(text: str):
        b = f"{text} /"
        print(b)
        sleep(0.05)
        cls()
        b = f"{text} -"
        print(b)
        sleep(0.05)
        cls()
        b = f"{text} \\"
        print(b)
        sleep(0.05)
        cls()
        b = f"{text} |"
        print(b)
        sleep(0.05)
        cls()

    @property
    def END(self):
        return self._END

    def color_demo(self):
        for color in self.style_dic:
            print(f"{color} -> {self.style_dic[color]}Effect{self._END}")

    @property
    def Underline2(self):
        return self._Underline

    def style_text(self, text, color, bold=False):
        text = self.style_dic.get(color, 'WHITE') + text + self._END
        if bold:
            text = self._Bold + text + self._END
        return text

toolboxv2.Spinner

Enhanced Spinner with tqdm-like line rendering.

Source code in toolboxv2/utils/extras/Style.py

class Spinner:
    """
    Enhanced Spinner with tqdm-like line rendering.
    """
    SYMBOL_SETS = {
        "c": ["◐", "◓", "◑", "◒"],
        "b": ["▁", "▃", "▄", "▅", "▆", "▇", "█", "▇", "▆", "▅", "▄", "▃"],
        "d": ["⣾", "⣽", "⣻", "⢿", "⡿", "⣟", "⣯", "⣷"],
        "w": ["🌍", "🌎", "🌏"],
        "s": ["🌀   ", " 🌀  ", "  🌀 ", "   🌀", "  🌀 ", " 🌀  "],
        "+": ["+", "x"],
        "t": ["✶", "✸", "✹", "✺", "◎", "◉", "◯", "✹", "◎", "✷"],
        "q": ["▖▘", "▘▝", "▝▗", "▗▖"],
        "o": ["●○○", "○●○", "○○●", "○●○"],
        "g": ["▏", "▎", "▍", "▌", "▋", "▊", "▉", "█"],
        "i": ["   ", ".  ", ".. ", "..."],
        "e":["[=   ]", "[==  ]", "[=== ]", "[ ===]", "[  ==]", "[   =]"],
        "p": ["◰", "◳", "◲", "◱"],
        "a": ["←", "↖", "↑", "↗", "→", "↘", "↓", "↙"],
        "h": ["▁▁▁", "▂▂▂", "▃▃▃", "▄▄▄", "▅▅▅"]

    }

    def __init__(
        self,
        message: str = "Loading...",
        delay: float = 0.1,
        symbols=None,
        count_down: bool = False,
        time_in_s: float = 0
    ):
        """Initialize spinner with flexible configuration."""
        # Resolve symbol set.
        if isinstance(symbols, str):
            symbols = self.SYMBOL_SETS.get(symbols, None)

        # Default symbols if not provided.
        if symbols is None:
            symbols = ["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"]

        # Test mode symbol set.
        if 'unittest' in sys.argv[0]:
            symbols = ['#', '=', '-']

        self.spinner = itertools.cycle(symbols)
        self.delay = delay
        self.message = message
        self.running = False
        self.spinner_thread = None
        self.max_t = time_in_s
        self.contd = count_down

        # Rendering management.
        self._is_primary = False
        self._start_time = 0

        # Central manager.
        self.manager = SpinnerManager()

    def _generate_render_line(self):
        """Generate the primary render line."""
        current_time = time.time()
        if self.contd:
            remaining = max(0, self.max_t - (current_time - self._start_time))
            time_display = f"{remaining:.2f}"
        else:
            time_display = f"{current_time - self._start_time:.2f}"

        symbol = next(self.spinner)
        return f"{symbol} {self.message} | {time_display}"

    def _generate_secondary_info(self):
        """Generate secondary spinner info for additional spinners."""
        return f"{self.message}"

    def __enter__(self):
        """Start the spinner."""
        if not self.manager.enabled:
            return self
        self.running = True
        self._start_time = time.time()
        self.manager.register_spinner(self)
        return self

    def __exit__(self, exc_type, exc_value, exc_traceback):
        """Stop the spinner."""
        self.running = False
        self.manager.unregister_spinner(self)
        # Clear the spinner's line if it was the primary spinner.
        if self._is_primary:
            if hasattr(sys.stdout, "buffer"):
                sys.stdout.buffer.write("\r\033[K".encode('utf-8'))
                sys.stdout.buffer.flush()

toolboxv2.remove_styles(text, infos=False)

Source code in toolboxv2/utils/extras/Style.py

def remove_styles(text: str, infos=False):
    in_ = []
    for key, style in Style.style_dic.items():
        if style in text:
            text = text.replace(style, '')
            if infos:
                in_.append([key for key, st in Style.style_dic.items() if style == st][0])
    if infos:
        if "END" in in_:
            in_.remove('END')
        return text, in_
    return text

Data Types & Structures

toolboxv2.AppArgs

Source code in toolboxv2/utils/system/types.py

class AppArgs:
    init = None
    init_file = 'init.config'
    get_version = False
    mm = False
    sm = False
    lm = False
    modi = 'cli'
    kill = False
    remote = False
    remote_direct_key = None
    background_application = False
    background_application_runner = False
    docker = False
    build = False
    install = None
    remove = None
    update = None
    name = 'main'
    port = 5000
    host = '0.0.0.0'
    load_all_mod_in_files = False
    mods_folder = 'toolboxv2.mods.'
    debug = None
    test = None
    profiler = None
    hot_reload = False
    live_application = True
    sysPrint = False
    kwargs = {}
    session = None

    def default(self):
        return self

    def set(self, name, value):
        setattr(self, name, value)
        return self

toolboxv2.Result

Bases: Generic[T]

Source code in toolboxv2/utils/system/types.py

class Result(Generic[T]):
    _task = None
    _generic_type: Optional[Type] = None

    def __init__(self,
                 error: ToolBoxError,
                 result: ToolBoxResult,
                 info: ToolBoxInfo,
                 origin: Any | None = None,
                 generic_type: Optional[Type] = None
                 ):
        self.error: ToolBoxError = error
        self.result: ToolBoxResult = result
        self.info: ToolBoxInfo = info
        self.origin = origin
        self._generic_type = generic_type

    def __class_getitem__(cls, item):
        """Enable Result[Type] syntax"""

        class TypedResult(cls):
            _generic_type = item

            def __init__(self, *args, **kwargs):
                super().__init__(*args, **kwargs)
                self._generic_type = item

        return TypedResult

    def typed_get(self, key=None, default=None) -> T:
        """Get data with type validation"""
        data = self.get(key, default)

        if self._generic_type and data is not None:
            # Validate type matches generic parameter
            if not self._validate_type(data, self._generic_type):
                from toolboxv2 import get_logger
                get_logger().warning(f"Type mismatch: expected {self._generic_type}, got {type(data)}")

        return data

    async def typed_aget(self, key=None, default=None) -> T:
        """Async get data with type validation"""
        data = await self.aget(key, default)

        if self._generic_type and data is not None:
            if not self._validate_type(data, self._generic_type):
                from toolboxv2 import get_logger
                get_logger().warning(f"Type mismatch: expected {self._generic_type}, got {type(data)}")

        return data

    def _validate_type(self, data, expected_type) -> bool:
        """Validate data matches expected type"""
        try:
            # Handle List[Type] syntax
            origin = get_origin(expected_type)
            if origin is list or origin is List:
                if not isinstance(data, list):
                    return False

                # Check list element types if specified
                args = get_args(expected_type)
                if args and data:
                    element_type = args[0]
                    return all(isinstance(item, element_type) for item in data)
                return True

            # Handle other generic types
            elif origin is not None:
                return isinstance(data, origin)

            # Handle regular types
            else:
                return isinstance(data, expected_type)

        except Exception:
            return True  # Skip validation on error

    @classmethod
    def typed_ok(cls, data: T, data_info="", info="OK", interface=ToolBoxInterfaces.native) -> 'Result[T]':
        """Create OK result with type information"""
        error = ToolBoxError.none
        info_obj = ToolBoxInfo(exec_code=0, help_text=info)
        result = ToolBoxResult(data_to=interface, data=data, data_info=data_info, data_type=type(data).__name__)

        instance = cls(error=error, info=info_obj, result=result)
        if hasattr(cls, '_generic_type'):
            instance._generic_type = cls._generic_type

        return instance

    @classmethod
    def typed_json(cls, data: T, info="OK", interface=ToolBoxInterfaces.remote, exec_code=0,
                   status_code=None) -> 'Result[T]':
        """Create JSON result with type information"""
        error = ToolBoxError.none
        info_obj = ToolBoxInfo(exec_code=status_code or exec_code, help_text=info)

        result = ToolBoxResult(
            data_to=interface,
            data=data,
            data_info="JSON response",
            data_type="json"
        )

        instance = cls(error=error, info=info_obj, result=result)
        if hasattr(cls, '_generic_type'):
            instance._generic_type = cls._generic_type

        return instance

    def cast_to(self, target_type: Type[T]) -> 'Result[T]':
        """Cast result to different type"""
        new_result = Result(
            error=self.error,
            result=self.result,
            info=self.info,
            origin=self.origin,
            generic_type=target_type
        )
        new_result._generic_type = target_type
        return new_result

    def get_type_info(self) -> Optional[Type]:
        """Get the generic type information"""
        return self._generic_type

    def is_typed(self) -> bool:
        """Check if result has type information"""
        return self._generic_type is not None

    def as_result(self):
        return self

    def as_dict(self):
        return {
            "error":self.error.value if isinstance(self.error, Enum) else self.error,
        "result" : {
            "data_to":self.result.data_to.value if isinstance(self.result.data_to, Enum) else self.result.data_to,
            "data_info":self.result.data_info,
            "data":self.result.data,
            "data_type":self.result.data_type
        } if self.result else None,
        "info" : {
            "exec_code" : self.info.exec_code,  # exec_code umwandel in http resposn codes
        "help_text" : self.info.help_text
        } if self.info else None,
        "origin" : self.origin
        }

    def set_origin(self, origin):
        if self.origin is not None:
            raise ValueError("You cannot Change the origin of a Result!")
        self.origin = origin
        return self

    def set_dir_origin(self, name, extras="assets/"):
        if self.origin is not None:
            raise ValueError("You cannot Change the origin of a Result!")
        self.origin = f"mods/{name}/{extras}"
        return self

    def is_error(self):
        if _test_is_result(self.result.data):
            return self.result.data.is_error()
        if self.error == ToolBoxError.none:
            return False
        if self.info.exec_code == 0:
            return False
        return self.info.exec_code != 200

    def is_ok(self):
        return not self.is_error()

    def is_data(self):
        return self.result.data is not None

    def to_api_result(self):
        # print(f" error={self.error}, result= {self.result}, info= {self.info}, origin= {self.origin}")
        return ApiResult(
            error=self.error.value if isinstance(self.error, Enum) else self.error,
            result=ToolBoxResultBM(
                data_to=self.result.data_to.value if isinstance(self.result.data_to, Enum) else self.result.data_to,
                data_info=self.result.data_info,
                data=self.result.data,
                data_type=self.result.data_type
            ) if self.result else None,
            info=ToolBoxInfoBM(
                exec_code=self.info.exec_code,  # exec_code umwandel in http resposn codes
                help_text=self.info.help_text
            ) if self.info else None,
            origin=self.origin
        )

    def task(self, task):
        self._task = task
        return self

    @staticmethod
    def result_from_dict(error: str, result: dict, info: dict, origin: list or None or str):
        # print(f" error={self.error}, result= {self.result}, info= {self.info}, origin= {self.origin}")
        return ApiResult(
            error=error if isinstance(error, Enum) else error,
            result=ToolBoxResultBM(
                data_to=result.get('data_to') if isinstance(result.get('data_to'), Enum) else result.get('data_to'),
                data_info=result.get('data_info', '404'),
                data=result.get('data'),
                data_type=result.get('data_type', '404'),
            ) if result else ToolBoxResultBM(
                data_to=ToolBoxInterfaces.cli.value,
                data_info='',
                data='404',
                data_type='404',
            ),
            info=ToolBoxInfoBM(
                exec_code=info.get('exec_code', 404),
                help_text=info.get('help_text', '404')
            ) if info else ToolBoxInfoBM(
                exec_code=404,
                help_text='404'
            ),
            origin=origin
        ).as_result()

    @classmethod
    def stream(cls,
               stream_generator: Any,  # Renamed from source for clarity
               content_type: str = "text/event-stream",  # Default to SSE
               headers: dict | None = None,
               info: str = "OK",
               interface: ToolBoxInterfaces = ToolBoxInterfaces.remote,
               cleanup_func: Callable[[], None] | Callable[[], T] | Callable[[], AsyncGenerator[T, None]] | None = None):
        """
        Create a streaming response Result. Handles SSE and other stream types.

        Args:
            stream_generator: Any stream source (async generator, sync generator, iterable, or single item).
            content_type: Content-Type header (default: text/event-stream for SSE).
            headers: Additional HTTP headers for the response.
            info: Help text for the result.
            interface: Interface to send data to.
            cleanup_func: Optional function for cleanup.

        Returns:
            A Result object configured for streaming.
        """
        error = ToolBoxError.none
        info_obj = ToolBoxInfo(exec_code=0, help_text=info)

        final_generator: AsyncGenerator[str, None]

        if content_type == "text/event-stream":
            # For SSE, always use SSEGenerator.create_sse_stream to wrap the source.
            # SSEGenerator.create_sse_stream handles various types of stream_generator internally.
            final_generator = SSEGenerator.create_sse_stream(source=stream_generator, cleanup_func=cleanup_func)

            # Standard SSE headers for the HTTP response itself
            # These will be stored in the Result object. Rust side decides how to use them.
            standard_sse_headers = {
                "Cache-Control": "no-cache",  # SSE specific
                "Connection": "keep-alive",  # SSE specific
                "X-Accel-Buffering": "no",  # Useful for proxies with SSE
                # Content-Type is implicitly text/event-stream, will be in streaming_data below
            }
            all_response_headers = standard_sse_headers.copy()
            if headers:
                all_response_headers.update(headers)
        else:
            # For non-SSE streams.
            # If stream_generator is sync, wrap it to be async.
            # If already async or single item, it will be handled.
            # Rust's stream_generator in ToolboxClient seems to handle both sync/async Python generators.
            # For consistency with how SSEGenerator does it, we can wrap sync ones.
            if inspect.isgenerator(stream_generator) or \
                (not isinstance(stream_generator, str) and hasattr(stream_generator, '__iter__')):
                final_generator = SSEGenerator.wrap_sync_generator(stream_generator)  # Simple async wrapper
            elif inspect.isasyncgen(stream_generator):
                final_generator = stream_generator
            else:  # Single item or string
                async def _single_item_gen():
                    yield stream_generator

                final_generator = _single_item_gen()
            all_response_headers = headers if headers else {}

        # Prepare streaming data to be stored in the Result object
        streaming_data = {
            "type": "stream",  # Indicator for Rust side
            "generator": final_generator,
            "content_type": content_type,  # Let Rust know the intended content type
            "headers": all_response_headers  # Intended HTTP headers for the overall response
        }

        result_payload = ToolBoxResult(
            data_to=interface,
            data=streaming_data,
            data_info="Streaming response" if content_type != "text/event-stream" else "SSE Event Stream",
            data_type="stream"  # Generic type for Rust to identify it needs to stream from 'generator'
        )

        return cls(error=error, info=info_obj, result=result_payload)

    @classmethod
    def sse(cls,
            stream_generator: Any,
            info: str = "OK",
            interface: ToolBoxInterfaces = ToolBoxInterfaces.remote,
            cleanup_func: Callable[[], None] | Callable[[], T] | Callable[[], AsyncGenerator[T, None]] | None = None,
            # http_headers: Optional[dict] = None # If we want to allow overriding default SSE HTTP headers
            ):
        """
        Create an Server-Sent Events (SSE) streaming response Result.

        Args:
            stream_generator: A source yielding individual data items. This can be an
                              async generator, sync generator, iterable, or a single item.
                              Each item will be formatted as an SSE event.
            info: Optional help text for the Result.
            interface: Optional ToolBoxInterface to target.
            cleanup_func: Optional cleanup function to run when the stream ends or is cancelled.
            #http_headers: Optional dictionary of custom HTTP headers for the SSE response.

        Returns:
            A Result object configured for SSE streaming.
        """
        # Result.stream will handle calling SSEGenerator.create_sse_stream
        # and setting appropriate default headers for SSE when content_type is "text/event-stream".
        return cls.stream(
            stream_generator=stream_generator,
            content_type="text/event-stream",
            # headers=http_headers, # Pass if we add http_headers param
            info=info,
            interface=interface,
            cleanup_func=cleanup_func
        )

    @classmethod
    def default(cls, interface=ToolBoxInterfaces.native):
        error = ToolBoxError.none
        info = ToolBoxInfo(exec_code=-1, help_text="")
        result = ToolBoxResult(data_to=interface)
        return cls(error=error, info=info, result=result)

    @classmethod
    def json(cls, data, info="OK", interface=ToolBoxInterfaces.remote, exec_code=0, status_code=None):
        """Create a JSON response Result."""
        error = ToolBoxError.none
        info_obj = ToolBoxInfo(exec_code=status_code or exec_code, help_text=info)

        result = ToolBoxResult(
            data_to=interface,
            data=data,
            data_info="JSON response",
            data_type="json"
        )

        return cls(error=error, info=info_obj, result=result)

    @classmethod
    def text(cls, text_data, content_type="text/plain",exec_code=None,status=200, info="OK", interface=ToolBoxInterfaces.remote, headers=None):
        """Create a text response Result with specific content type."""
        if headers is not None:
            return cls.html(text_data, status= exec_code or status, info=info, headers=headers)
        error = ToolBoxError.none
        info_obj = ToolBoxInfo(exec_code=exec_code or status, help_text=info)

        result = ToolBoxResult(
            data_to=interface,
            data=text_data,
            data_info="Text response",
            data_type=content_type
        )

        return cls(error=error, info=info_obj, result=result)

    @classmethod
    def binary(cls, data, content_type="application/octet-stream", download_name=None, info="OK",
               interface=ToolBoxInterfaces.remote):
        """Create a binary data response Result."""
        error = ToolBoxError.none
        info_obj = ToolBoxInfo(exec_code=0, help_text=info)

        # Create a dictionary with binary data and metadata
        binary_data = {
            "data": data,
            "content_type": content_type,
            "filename": download_name
        }

        result = ToolBoxResult(
            data_to=interface,
            data=binary_data,
            data_info=f"Binary response: {download_name}" if download_name else "Binary response",
            data_type="binary"
        )

        return cls(error=error, info=info_obj, result=result)

    @classmethod
    def file(cls, data, filename, content_type=None, info="OK", interface=ToolBoxInterfaces.remote):
        """Create a file download response Result.

        Args:
            data: File data as bytes or base64 string
            filename: Name of the file for download
            content_type: MIME type of the file (auto-detected if None)
            info: Response info text
            interface: Target interface

        Returns:
            Result object configured for file download
        """
        import base64
        import mimetypes

        error = ToolBoxError.none
        info_obj = ToolBoxInfo(exec_code=200, help_text=info)

        # Auto-detect content type if not provided
        if content_type is None:
            content_type, _ = mimetypes.guess_type(filename)
            if content_type is None:
                content_type = "application/octet-stream"

        # Ensure data is base64 encoded string (as expected by Rust server)
        if isinstance(data, bytes):
            base64_data = base64.b64encode(data).decode('utf-8')
        elif isinstance(data, str):
            # Assume it's already base64 encoded
            base64_data = data
        else:
            raise ValueError("File data must be bytes or base64 string")

        result = ToolBoxResult(
            data_to=interface,
            data=base64_data,  # Rust expects base64 string for "file" type
            data_info=f"File download: {filename}",
            data_type="file"
        )

        return cls(error=error, info=info_obj, result=result)

    @classmethod
    def file_path(cls, path: str, filename: str = None, content_type: str = None,
                  info: str = "OK", interface=None):
        """
        Erstellt ein Result für File-Download via Streaming.

        Die Datei wird NICHT in Memory geladen, sondern direkt gestreamt.
        Nutzt wsgi.file_wrapper für optimale Performance.

        Args:
            path: Absoluter Pfad zur Datei
            filename: Download-Filename (default: basename von path)
            content_type: MIME Type (default: auto-detect)
            info: Info text
            interface: ToolBoxInterface (default: remote)

        Returns:
            Result mit data_type="file_path"

        Example:
            # In einem Modul:
            @export(mod_name="MyMod", api=True)
            def download_report(request):
                report_path = "/path/to/report.pdf"
                return Result.file_path(report_path, filename="Monthly_Report.pdf")
        """
        import os
        import mimetypes

        # Interface default
        if interface is None:
            from toolboxv2.utils.system.types import ToolBoxInterfaces
            interface = ToolBoxInterfaces.remote

        # Prüfe ob Datei existiert
        if not os.path.isfile(path):
            return cls.error(
                data=f"File not found: {path}",
                info="File not found",
                exec_code=404
            )

        # Filename
        if filename is None:
            filename = os.path.basename(path)

        # Content-Type auto-detect
        if content_type is None:
            content_type, _ = mimetypes.guess_type(path)
            content_type = content_type or "application/octet-stream"

        # Result erstellen
        from toolboxv2.utils.system.types import (
            ToolBoxError, ToolBoxInfo, ToolBoxResult
        )

        error = ToolBoxError.none
        info_obj = ToolBoxInfo(exec_code=200, help_text=info)

        result = ToolBoxResult(
            data_to=interface,
            data=path,  # Pfad, nicht Inhalt!
            data_info=f"filename={filename}",
            data_type="file_path"  # Wird von server_worker erkannt
        )

        return cls(error=error, info=info_obj, result=result)

    @classmethod
    def file_stream(cls, generator, filename: str, content_type: str = None,
                    size: int = None, info: str = "OK", interface=None):
        """
        Erstellt ein Result für File-Download via Generator-Streaming.

        Für Dateien die on-the-fly generiert werden (z.B. aus MinIO).

        Args:
            generator: Generator/Iterator der bytes yielded
            filename: Download-Filename
            content_type: MIME Type (default: application/octet-stream)
            size: Dateigröße wenn bekannt (für Content-Length header)
            info: Info text
            interface: ToolBoxInterface

        Returns:
            Result mit streaming data

        Example:
            # Streaming aus MinIO:
            @export(mod_name="MyMod", api=True)
            async def download_from_minio(request, file_id):
                def stream_minio():
                    response = minio_client.get_object(bucket, file_id)
                    for chunk in response.stream(32*1024):
                        yield chunk
                    response.close()

                return Result.file_stream(
                    stream_minio(),
                    filename="data.csv",
                    content_type="text/csv"
                )
        """
        import mimetypes

        if interface is None:
            from toolboxv2.utils.system.types import ToolBoxInterfaces
            interface = ToolBoxInterfaces.remote

        # Content-Type
        if content_type is None:
            content_type, _ = mimetypes.guess_type(filename)
            content_type = content_type or "application/octet-stream"

        # Headers
        headers = {
            "Content-Type": content_type,
            "Content-Disposition": f'attachment; filename="{filename}"',
        }
        if size:
            headers["Content-Length"] = str(size)

        # Nutze existierende stream() Methode
        return cls.stream(
            stream_generator=generator,
            content_type=content_type,
            headers=headers,
            info=info,
            interface=interface
        )

    @classmethod
    def redirect(cls, url, status_code=302, info="Redirect", interface=ToolBoxInterfaces.remote):
        """Create a redirect response."""
        error = ToolBoxError.none
        info_obj = ToolBoxInfo(exec_code=status_code, help_text=info)

        result = ToolBoxResult(
            data_to=interface,
            data=url,
            data_info="Redirect response",
            data_type="redirect"
        )

        return cls(error=error, info=info_obj, result=result)

    @classmethod
    def ok(cls, data=None, data_info="", info="OK", interface=ToolBoxInterfaces.native):
        error = ToolBoxError.none
        info = ToolBoxInfo(exec_code=0, help_text=info)
        result = ToolBoxResult(data_to=interface, data=data, data_info=data_info, data_type=type(data).__name__)
        return cls(error=error, info=info, result=result)

    @classmethod
    def html(cls, data=None, data_info="", info="OK", interface=ToolBoxInterfaces.remote, data_type="html",status=200, headers=None, row=False):
        error = ToolBoxError.none
        info = ToolBoxInfo(exec_code=status, help_text=info)
        from ...utils.system.getting_and_closing_app import get_app

        if not row and not '"<div class="main-content""' in data:
            data = f'<div class="main-content frosted-glass">{data}<div>'
        if not row and not get_app().web_context() in data:
            data = get_app().web_context() + data

        if isinstance(headers, dict):
            result = ToolBoxResult(data_to=interface, data={'html':data,'headers':headers}, data_info=data_info,
                                   data_type="special_html")
        else:
            result = ToolBoxResult(data_to=interface, data=data, data_info=data_info,
                                   data_type=data_type if data_type is not None else type(data).__name__)
        return cls(error=error, info=info, result=result)

    @classmethod
    def future(cls, data=None, data_info="", info="OK", interface=ToolBoxInterfaces.future):
        error = ToolBoxError.none
        info = ToolBoxInfo(exec_code=0, help_text=info)
        result = ToolBoxResult(data_to=interface, data=data, data_info=data_info, data_type="future")
        return cls(error=error, info=info, result=result)

    @classmethod
    def custom_error(cls, data=None, data_info="", info="", exec_code=-1, interface=ToolBoxInterfaces.native):
        error = ToolBoxError.custom_error
        info = ToolBoxInfo(exec_code=exec_code, help_text=info)
        result = ToolBoxResult(data_to=interface, data=data, data_info=data_info, data_type=type(data).__name__)
        return cls(error=error, info=info, result=result)

    @classmethod
    def error(cls, data=None, data_info="", info="", exec_code=450, interface=ToolBoxInterfaces.remote):
        error = ToolBoxError.custom_error
        info = ToolBoxInfo(exec_code=exec_code, help_text=info)
        result = ToolBoxResult(data_to=interface, data=data, data_info=data_info, data_type=type(data).__name__)
        return cls(error=error, info=info, result=result)

    @classmethod
    def default_user_error(cls, info="", exec_code=-3, interface=ToolBoxInterfaces.native, data=None):
        error = ToolBoxError.input_error
        info = ToolBoxInfo(exec_code, info)
        result = ToolBoxResult(data_to=interface, data=data, data_type=type(data).__name__)
        return cls(error=error, info=info, result=result)

    @classmethod
    def default_internal_error(cls, info="", exec_code=-2, interface=ToolBoxInterfaces.native, data=None):
        error = ToolBoxError.internal_error
        info = ToolBoxInfo(exec_code, info)
        result = ToolBoxResult(data_to=interface, data=data, data_type=type(data).__name__)
        return cls(error=error, info=info, result=result)

    def print(self, show=True, show_data=True, prifix="", full_data=False):
        # frame = inspect.currentframe().f_back
        # info = inspect.getframeinfo(frame)
#
        # cls = None
        # if "self" in frame.f_locals:
        #     cls = frame.f_locals["self"].__class__.__name__
#
        # print(
        #     f"Caller: {cls + '.' if cls else ''}{frame.f_code.co_name} "
        #     f"| File: {info.filename} "
        #     f"| Line: {info.lineno}"
        # )
        data = '\n' + f"{((prifix + f'Data_{self.result.data_type}: ' + str(self.result.data) if self.result.data is not None else 'NO Data') if not isinstance(self.result.data, Result) else self.result.data.print(show=False, show_data=show_data, prifix=prifix + '-')) if show_data else 'Data: private'}"
        origin = '\n' + f"{prifix + 'Origin: ' + str(self.origin) if self.origin is not None else 'NO Origin'}"
        text = (f"Function Exec code: {self.info.exec_code}"
                f"\n{prifix}Info's:"
                f" {self.info.help_text} {'<|> ' + str(self.result.data_info) if self.result.data_info is not None else ''}"
                f"{origin}{((data[:100]+'...') if not full_data else (data)) if not data.endswith('NO Data') else ''}\n")
        if not show:
            return text
        print("\n======== Result ========\n" + text + "------- EndOfD -------")
        return self

    def log(self, show_data=True, prifix=""):
        from toolboxv2 import get_logger
        get_logger().debug(self.print(show=False, show_data=show_data, prifix=prifix).replace("\n", " - "))
        return self

    def __str__(self):
        # frame = inspect.currentframe().f_back
        # info = inspect.getframeinfo(frame)
#
        # cls = None
        # if "self" in frame.f_locals:
        #     cls = frame.f_locals["self"].__class__.__name__
#
        # print(
        #     f"Caller: {cls + '.' if cls else ''}{frame.f_code.co_name} "
        #     f"| File: {info.filename} "
        #     f"| Line: {info.lineno}"
        # )
        return self.print(show=False, show_data=True)

    def get(self, key=None, default=None):
        data = self.result.data
        if isinstance(data, Result) or hasattr(data, 'result'):
            return data.get(key=key, default=default)
        if key is not None and isinstance(data, dict):
            return data.get(key, default)
        return data if data is not None else default

    async def aget(self, key=None, default=None):
        if asyncio.isfuture(self.result.data) or asyncio.iscoroutine(self.result.data) or (
            isinstance(self.result.data_to, Enum) and self.result.data_to.name == ToolBoxInterfaces.future.name):
            data = await self.result.data
        else:
            data = self.get(key=None, default=None)
        if isinstance(data, Result):
            return data.get(key=key, default=default)
        if key is not None and isinstance(data, dict):
            return data.get(key, default)
        return data if data is not None else default

    def lazy_return(self, _=0, data=None, **kwargs):
        flags = ['raise', 'logg', 'user', 'intern']
        flag = flags[_] if isinstance(_, int) else _
        if self.info.exec_code == 0:
            return self if data is None else data if _test_is_result(data) else self.ok(data=data, **kwargs)
        if flag == 'raise':
            raise ValueError(self.print(show=False))
        if flag == 'logg':
            from .. import get_logger
            get_logger().error(self.print(show=False))

        if flag == 'user':
            return self if data is None else data if _test_is_result(data) else self.default_user_error(data=data,
                                                                                                        **kwargs)
        if flag == 'intern':
            return self if data is None else data if _test_is_result(data) else self.default_internal_error(data=data,
                                                                                                            **kwargs)

        return self if data is None else data if _test_is_result(data) else self.custom_error(data=data, **kwargs)

    @property
    def bg_task(self):
        return self._task

binary(data, content_type='application/octet-stream', download_name=None, info='OK', interface=ToolBoxInterfaces.remote) classmethod

Create a binary data response Result.

Source code in toolboxv2/utils/system/types.py

@classmethod
def binary(cls, data, content_type="application/octet-stream", download_name=None, info="OK",
           interface=ToolBoxInterfaces.remote):
    """Create a binary data response Result."""
    error = ToolBoxError.none
    info_obj = ToolBoxInfo(exec_code=0, help_text=info)

    # Create a dictionary with binary data and metadata
    binary_data = {
        "data": data,
        "content_type": content_type,
        "filename": download_name
    }

    result = ToolBoxResult(
        data_to=interface,
        data=binary_data,
        data_info=f"Binary response: {download_name}" if download_name else "Binary response",
        data_type="binary"
    )

    return cls(error=error, info=info_obj, result=result)

cast_to(target_type)

Cast result to different type

Source code in toolboxv2/utils/system/types.py

def cast_to(self, target_type: Type[T]) -> 'Result[T]':
    """Cast result to different type"""
    new_result = Result(
        error=self.error,
        result=self.result,
        info=self.info,
        origin=self.origin,
        generic_type=target_type
    )
    new_result._generic_type = target_type
    return new_result

file(data, filename, content_type=None, info='OK', interface=ToolBoxInterfaces.remote) classmethod

Create a file download response Result.

Parameters:

Name	Type	Description	Default
data		File data as bytes or base64 string	required
filename		Name of the file for download	required
content_type		MIME type of the file (auto-detected if None)	None
info		Response info text	'OK'
interface		Target interface	remote

Returns:

Type	Description
	Result object configured for file download

Source code in toolboxv2/utils/system/types.py

@classmethod
def file(cls, data, filename, content_type=None, info="OK", interface=ToolBoxInterfaces.remote):
    """Create a file download response Result.

    Args:
        data: File data as bytes or base64 string
        filename: Name of the file for download
        content_type: MIME type of the file (auto-detected if None)
        info: Response info text
        interface: Target interface

    Returns:
        Result object configured for file download
    """
    import base64
    import mimetypes

    error = ToolBoxError.none
    info_obj = ToolBoxInfo(exec_code=200, help_text=info)

    # Auto-detect content type if not provided
    if content_type is None:
        content_type, _ = mimetypes.guess_type(filename)
        if content_type is None:
            content_type = "application/octet-stream"

    # Ensure data is base64 encoded string (as expected by Rust server)
    if isinstance(data, bytes):
        base64_data = base64.b64encode(data).decode('utf-8')
    elif isinstance(data, str):
        # Assume it's already base64 encoded
        base64_data = data
    else:
        raise ValueError("File data must be bytes or base64 string")

    result = ToolBoxResult(
        data_to=interface,
        data=base64_data,  # Rust expects base64 string for "file" type
        data_info=f"File download: {filename}",
        data_type="file"
    )

    return cls(error=error, info=info_obj, result=result)

file_path(path, filename=None, content_type=None, info='OK', interface=None) classmethod

Erstellt ein Result für File-Download via Streaming.

Die Datei wird NICHT in Memory geladen, sondern direkt gestreamt. Nutzt wsgi.file_wrapper für optimale Performance.

Parameters:

Name	Type	Description	Default
path	str	Absoluter Pfad zur Datei	required
filename	str	Download-Filename (default: basename von path)	None
content_type	str	MIME Type (default: auto-detect)	None
info	str	Info text	'OK'
interface		ToolBoxInterface (default: remote)	None

Returns:

Type	Description
	Result mit data_type="file_path"

Example

In einem Modul:

@export(mod_name="MyMod", api=True) def download_report(request): report_path = "/path/to/report.pdf" return Result.file_path(report_path, filename="Monthly_Report.pdf")

Source code in toolboxv2/utils/system/types.py

@classmethod
def file_path(cls, path: str, filename: str = None, content_type: str = None,
              info: str = "OK", interface=None):
    """
    Erstellt ein Result für File-Download via Streaming.

    Die Datei wird NICHT in Memory geladen, sondern direkt gestreamt.
    Nutzt wsgi.file_wrapper für optimale Performance.

    Args:
        path: Absoluter Pfad zur Datei
        filename: Download-Filename (default: basename von path)
        content_type: MIME Type (default: auto-detect)
        info: Info text
        interface: ToolBoxInterface (default: remote)

    Returns:
        Result mit data_type="file_path"

    Example:
        # In einem Modul:
        @export(mod_name="MyMod", api=True)
        def download_report(request):
            report_path = "/path/to/report.pdf"
            return Result.file_path(report_path, filename="Monthly_Report.pdf")
    """
    import os
    import mimetypes

    # Interface default
    if interface is None:
        from toolboxv2.utils.system.types import ToolBoxInterfaces
        interface = ToolBoxInterfaces.remote

    # Prüfe ob Datei existiert
    if not os.path.isfile(path):
        return cls.error(
            data=f"File not found: {path}",
            info="File not found",
            exec_code=404
        )

    # Filename
    if filename is None:
        filename = os.path.basename(path)

    # Content-Type auto-detect
    if content_type is None:
        content_type, _ = mimetypes.guess_type(path)
        content_type = content_type or "application/octet-stream"

    # Result erstellen
    from toolboxv2.utils.system.types import (
        ToolBoxError, ToolBoxInfo, ToolBoxResult
    )

    error = ToolBoxError.none
    info_obj = ToolBoxInfo(exec_code=200, help_text=info)

    result = ToolBoxResult(
        data_to=interface,
        data=path,  # Pfad, nicht Inhalt!
        data_info=f"filename={filename}",
        data_type="file_path"  # Wird von server_worker erkannt
    )

    return cls(error=error, info=info_obj, result=result)

file_stream(generator, filename, content_type=None, size=None, info='OK', interface=None) classmethod

Erstellt ein Result für File-Download via Generator-Streaming.

Für Dateien die on-the-fly generiert werden (z.B. aus MinIO).

Parameters:

Name	Type	Description	Default
generator		Generator/Iterator der bytes yielded	required
filename	str	Download-Filename	required
content_type	str	MIME Type (default: application/octet-stream)	None
size	int	Dateigröße wenn bekannt (für Content-Length header)	None
info	str	Info text	'OK'
interface		ToolBoxInterface	None

Returns:

Type	Description
	Result mit streaming data

Example

Streaming aus MinIO:

@export(mod_name="MyMod", api=True) async def download_from_minio(request, file_id): def stream_minio(): response = minio_client.get_object(bucket, file_id) for chunk in response.stream(32*1024): yield chunk response.close()

return Result.file_stream(
    stream_minio(),
    filename="data.csv",
    content_type="text/csv"
)

Source code in toolboxv2/utils/system/types.py

@classmethod
def file_stream(cls, generator, filename: str, content_type: str = None,
                size: int = None, info: str = "OK", interface=None):
    """
    Erstellt ein Result für File-Download via Generator-Streaming.

    Für Dateien die on-the-fly generiert werden (z.B. aus MinIO).

    Args:
        generator: Generator/Iterator der bytes yielded
        filename: Download-Filename
        content_type: MIME Type (default: application/octet-stream)
        size: Dateigröße wenn bekannt (für Content-Length header)
        info: Info text
        interface: ToolBoxInterface

    Returns:
        Result mit streaming data

    Example:
        # Streaming aus MinIO:
        @export(mod_name="MyMod", api=True)
        async def download_from_minio(request, file_id):
            def stream_minio():
                response = minio_client.get_object(bucket, file_id)
                for chunk in response.stream(32*1024):
                    yield chunk
                response.close()

            return Result.file_stream(
                stream_minio(),
                filename="data.csv",
                content_type="text/csv"
            )
    """
    import mimetypes

    if interface is None:
        from toolboxv2.utils.system.types import ToolBoxInterfaces
        interface = ToolBoxInterfaces.remote

    # Content-Type
    if content_type is None:
        content_type, _ = mimetypes.guess_type(filename)
        content_type = content_type or "application/octet-stream"

    # Headers
    headers = {
        "Content-Type": content_type,
        "Content-Disposition": f'attachment; filename="{filename}"',
    }
    if size:
        headers["Content-Length"] = str(size)

    # Nutze existierende stream() Methode
    return cls.stream(
        stream_generator=generator,
        content_type=content_type,
        headers=headers,
        info=info,
        interface=interface
    )

get_type_info()

Get the generic type information

Source code in toolboxv2/utils/system/types.py

def get_type_info(self) -> Optional[Type]:
    """Get the generic type information"""
    return self._generic_type

is_typed()

Check if result has type information

Source code in toolboxv2/utils/system/types.py

def is_typed(self) -> bool:
    """Check if result has type information"""
    return self._generic_type is not None

json(data, info='OK', interface=ToolBoxInterfaces.remote, exec_code=0, status_code=None) classmethod

Create a JSON response Result.

Source code in toolboxv2/utils/system/types.py

@classmethod
def json(cls, data, info="OK", interface=ToolBoxInterfaces.remote, exec_code=0, status_code=None):
    """Create a JSON response Result."""
    error = ToolBoxError.none
    info_obj = ToolBoxInfo(exec_code=status_code or exec_code, help_text=info)

    result = ToolBoxResult(
        data_to=interface,
        data=data,
        data_info="JSON response",
        data_type="json"
    )

    return cls(error=error, info=info_obj, result=result)

redirect(url, status_code=302, info='Redirect', interface=ToolBoxInterfaces.remote) classmethod

Create a redirect response.

Source code in toolboxv2/utils/system/types.py

@classmethod
def redirect(cls, url, status_code=302, info="Redirect", interface=ToolBoxInterfaces.remote):
    """Create a redirect response."""
    error = ToolBoxError.none
    info_obj = ToolBoxInfo(exec_code=status_code, help_text=info)

    result = ToolBoxResult(
        data_to=interface,
        data=url,
        data_info="Redirect response",
        data_type="redirect"
    )

    return cls(error=error, info=info_obj, result=result)

sse(stream_generator, info='OK', interface=ToolBoxInterfaces.remote, cleanup_func=None) classmethod

Create an Server-Sent Events (SSE) streaming response Result.

Parameters:

Name	Type	Description	Default
stream_generator	Any	A source yielding individual data items. This can be an async generator, sync generator, iterable, or a single item. Each item will be formatted as an SSE event.	required
info	str	Optional help text for the Result.	'OK'
interface	ToolBoxInterfaces	Optional ToolBoxInterface to target.	remote
cleanup_func	Callable[[], None] | Callable[[], T] | Callable[[], AsyncGenerator[T, None]] | None	Optional cleanup function to run when the stream ends or is cancelled.	None
#http_headers		Optional dictionary of custom HTTP headers for the SSE response.	required

Returns:

Type	Description
	A Result object configured for SSE streaming.

Source code in toolboxv2/utils/system/types.py

@classmethod
def sse(cls,
        stream_generator: Any,
        info: str = "OK",
        interface: ToolBoxInterfaces = ToolBoxInterfaces.remote,
        cleanup_func: Callable[[], None] | Callable[[], T] | Callable[[], AsyncGenerator[T, None]] | None = None,
        # http_headers: Optional[dict] = None # If we want to allow overriding default SSE HTTP headers
        ):
    """
    Create an Server-Sent Events (SSE) streaming response Result.

    Args:
        stream_generator: A source yielding individual data items. This can be an
                          async generator, sync generator, iterable, or a single item.
                          Each item will be formatted as an SSE event.
        info: Optional help text for the Result.
        interface: Optional ToolBoxInterface to target.
        cleanup_func: Optional cleanup function to run when the stream ends or is cancelled.
        #http_headers: Optional dictionary of custom HTTP headers for the SSE response.

    Returns:
        A Result object configured for SSE streaming.
    """
    # Result.stream will handle calling SSEGenerator.create_sse_stream
    # and setting appropriate default headers for SSE when content_type is "text/event-stream".
    return cls.stream(
        stream_generator=stream_generator,
        content_type="text/event-stream",
        # headers=http_headers, # Pass if we add http_headers param
        info=info,
        interface=interface,
        cleanup_func=cleanup_func
    )

stream(stream_generator, content_type='text/event-stream', headers=None, info='OK', interface=ToolBoxInterfaces.remote, cleanup_func=None) classmethod

Create a streaming response Result. Handles SSE and other stream types.

Parameters:

Name	Type	Description	Default
stream_generator	Any	Any stream source (async generator, sync generator, iterable, or single item).	required
content_type	str	Content-Type header (default: text/event-stream for SSE).	'text/event-stream'
headers	dict | None	Additional HTTP headers for the response.	None
info	str	Help text for the result.	'OK'
interface	ToolBoxInterfaces	Interface to send data to.	remote
cleanup_func	Callable[[], None] | Callable[[], T] | Callable[[], AsyncGenerator[T, None]] | None	Optional function for cleanup.	None

Returns:

Type	Description
	A Result object configured for streaming.

Source code in toolboxv2/utils/system/types.py

@classmethod
def stream(cls,
           stream_generator: Any,  # Renamed from source for clarity
           content_type: str = "text/event-stream",  # Default to SSE
           headers: dict | None = None,
           info: str = "OK",
           interface: ToolBoxInterfaces = ToolBoxInterfaces.remote,
           cleanup_func: Callable[[], None] | Callable[[], T] | Callable[[], AsyncGenerator[T, None]] | None = None):
    """
    Create a streaming response Result. Handles SSE and other stream types.

    Args:
        stream_generator: Any stream source (async generator, sync generator, iterable, or single item).
        content_type: Content-Type header (default: text/event-stream for SSE).
        headers: Additional HTTP headers for the response.
        info: Help text for the result.
        interface: Interface to send data to.
        cleanup_func: Optional function for cleanup.

    Returns:
        A Result object configured for streaming.
    """
    error = ToolBoxError.none
    info_obj = ToolBoxInfo(exec_code=0, help_text=info)

    final_generator: AsyncGenerator[str, None]

    if content_type == "text/event-stream":
        # For SSE, always use SSEGenerator.create_sse_stream to wrap the source.
        # SSEGenerator.create_sse_stream handles various types of stream_generator internally.
        final_generator = SSEGenerator.create_sse_stream(source=stream_generator, cleanup_func=cleanup_func)

        # Standard SSE headers for the HTTP response itself
        # These will be stored in the Result object. Rust side decides how to use them.
        standard_sse_headers = {
            "Cache-Control": "no-cache",  # SSE specific
            "Connection": "keep-alive",  # SSE specific
            "X-Accel-Buffering": "no",  # Useful for proxies with SSE
            # Content-Type is implicitly text/event-stream, will be in streaming_data below
        }
        all_response_headers = standard_sse_headers.copy()
        if headers:
            all_response_headers.update(headers)
    else:
        # For non-SSE streams.
        # If stream_generator is sync, wrap it to be async.
        # If already async or single item, it will be handled.
        # Rust's stream_generator in ToolboxClient seems to handle both sync/async Python generators.
        # For consistency with how SSEGenerator does it, we can wrap sync ones.
        if inspect.isgenerator(stream_generator) or \
            (not isinstance(stream_generator, str) and hasattr(stream_generator, '__iter__')):
            final_generator = SSEGenerator.wrap_sync_generator(stream_generator)  # Simple async wrapper
        elif inspect.isasyncgen(stream_generator):
            final_generator = stream_generator
        else:  # Single item or string
            async def _single_item_gen():
                yield stream_generator

            final_generator = _single_item_gen()
        all_response_headers = headers if headers else {}

    # Prepare streaming data to be stored in the Result object
    streaming_data = {
        "type": "stream",  # Indicator for Rust side
        "generator": final_generator,
        "content_type": content_type,  # Let Rust know the intended content type
        "headers": all_response_headers  # Intended HTTP headers for the overall response
    }

    result_payload = ToolBoxResult(
        data_to=interface,
        data=streaming_data,
        data_info="Streaming response" if content_type != "text/event-stream" else "SSE Event Stream",
        data_type="stream"  # Generic type for Rust to identify it needs to stream from 'generator'
    )

    return cls(error=error, info=info_obj, result=result_payload)

text(text_data, content_type='text/plain', exec_code=None, status=200, info='OK', interface=ToolBoxInterfaces.remote, headers=None) classmethod

Create a text response Result with specific content type.

Source code in toolboxv2/utils/system/types.py

@classmethod
def text(cls, text_data, content_type="text/plain",exec_code=None,status=200, info="OK", interface=ToolBoxInterfaces.remote, headers=None):
    """Create a text response Result with specific content type."""
    if headers is not None:
        return cls.html(text_data, status= exec_code or status, info=info, headers=headers)
    error = ToolBoxError.none
    info_obj = ToolBoxInfo(exec_code=exec_code or status, help_text=info)

    result = ToolBoxResult(
        data_to=interface,
        data=text_data,
        data_info="Text response",
        data_type=content_type
    )

    return cls(error=error, info=info_obj, result=result)

typed_aget(key=None, default=None) async

Async get data with type validation

Source code in toolboxv2/utils/system/types.py

async def typed_aget(self, key=None, default=None) -> T:
    """Async get data with type validation"""
    data = await self.aget(key, default)

    if self._generic_type and data is not None:
        if not self._validate_type(data, self._generic_type):
            from toolboxv2 import get_logger
            get_logger().warning(f"Type mismatch: expected {self._generic_type}, got {type(data)}")

    return data

typed_get(key=None, default=None)

Get data with type validation

Source code in toolboxv2/utils/system/types.py

def typed_get(self, key=None, default=None) -> T:
    """Get data with type validation"""
    data = self.get(key, default)

    if self._generic_type and data is not None:
        # Validate type matches generic parameter
        if not self._validate_type(data, self._generic_type):
            from toolboxv2 import get_logger
            get_logger().warning(f"Type mismatch: expected {self._generic_type}, got {type(data)}")

    return data

typed_json(data, info='OK', interface=ToolBoxInterfaces.remote, exec_code=0, status_code=None) classmethod

Create JSON result with type information

Source code in toolboxv2/utils/system/types.py

@classmethod
def typed_json(cls, data: T, info="OK", interface=ToolBoxInterfaces.remote, exec_code=0,
               status_code=None) -> 'Result[T]':
    """Create JSON result with type information"""
    error = ToolBoxError.none
    info_obj = ToolBoxInfo(exec_code=status_code or exec_code, help_text=info)

    result = ToolBoxResult(
        data_to=interface,
        data=data,
        data_info="JSON response",
        data_type="json"
    )

    instance = cls(error=error, info=info_obj, result=result)
    if hasattr(cls, '_generic_type'):
        instance._generic_type = cls._generic_type

    return instance

typed_ok(data, data_info='', info='OK', interface=ToolBoxInterfaces.native) classmethod

Create OK result with type information

Source code in toolboxv2/utils/system/types.py

@classmethod
def typed_ok(cls, data: T, data_info="", info="OK", interface=ToolBoxInterfaces.native) -> 'Result[T]':
    """Create OK result with type information"""
    error = ToolBoxError.none
    info_obj = ToolBoxInfo(exec_code=0, help_text=info)
    result = ToolBoxResult(data_to=interface, data=data, data_info=data_info, data_type=type(data).__name__)

    instance = cls(error=error, info=info_obj, result=result)
    if hasattr(cls, '_generic_type'):
        instance._generic_type = cls._generic_type

    return instance

toolboxv2.ApiResult

Bases: BaseModel

Source code in toolboxv2/utils/system/types.py

class ApiResult(BaseModel):
    error: None | str= None
    origin: Any | None
    result: ToolBoxResultBM | None = None
    info: ToolBoxInfoBM | None

    def as_result(self):
        return Result(
            error=self.error.value if isinstance(self.error, Enum) else self.error,
            result=ToolBoxResult(
                data_to=self.result.data_to.value if isinstance(self.result.data_to, Enum) else self.result.data_to,
                data_info=self.result.data_info,
                data=self.result.data,
                data_type=self.result.data_type
            ) if self.result else None,
            info=ToolBoxInfo(
                exec_code=self.info.exec_code,
                help_text=self.info.help_text
            ) if self.info else None,
            origin=self.origin
        )

    def to_api_result(self):
        return self

    def print(self, *args, **kwargs):
        res = self.as_result().print(*args, **kwargs)
        if not isinstance(res, str):
            res = res.to_api_result()
        return res

    def __getattr__(self, name):
        # proxy to result
        return getattr(self.as_result(), name)

toolboxv2.RequestData

Main class representing the complete request data structure.

Source code in toolboxv2/utils/system/types.py

@dataclass
class RequestData:
    """Main class representing the complete request data structure."""
    request: Request
    session: Session
    session_id: str

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> 'RequestData':
        """Create a RequestData instance from a dictionary."""
        return cls(
            request=Request.from_dict(data.get('request', {})),
            session=Session.from_dict(data.get('session', {})),
            session_id=data.get('session_id', '')
        )

    def to_dict(self) -> dict[str, Any]:
        """Convert the RequestData object back to a dictionary."""
        return {
            'request': self.request.to_dict(),
            'session': self.session.to_dict(),
            'session_id': self.session_id
        }

    def __getattr__(self, name: str) -> Any:
        """Delegate unknown attributes to the `request` object."""
        # Nur wenn das Attribut nicht direkt in RequestData existiert
        # und auch nicht `session` oder `session_id` ist
        if hasattr(self.request, name):
            return getattr(self.request, name)
        raise AttributeError(f"'RequestData' object has no attribute '{name}'")

    @classmethod
    def moc(cls):
        return cls(
            request=Request.from_dict({
                'content_type': 'application/x-www-form-urlencoded',
                'headers': {
                    'accept': '*/*',
                    'accept-encoding': 'gzip, deflate, br, zstd',
                    'accept-language': 'de-DE,de;q=0.9,en-US;q=0.8,en;q=0.7',
                    'connection': 'keep-alive',
                    'content-length': '107',
                    'content-type': 'application/x-www-form-urlencoded',
                    'cookie': 'session=abc123',
                    'host': 'localhost:8080',
                    'hx-current-url': 'http://localhost:8080/api/TruthSeeker/get_main_ui',
                    'hx-request': 'true',
                    'hx-target': 'estimates-guest_1fc2c9',
                    'hx-trigger': 'config-form-guest_1fc2c9',
                    'origin': 'http://localhost:8080',
                    'referer': 'http://localhost:8080/api/TruthSeeker/get_main_ui',
                    'sec-ch-ua': '"Chromium";v="134", "Not:A-Brand";v="24", "Google Chrome";v="134"',
                    'sec-ch-ua-mobile': '?0',
                    'sec-ch-ua-platform': '"Windows"',
                    'sec-fetch-dest': 'empty',
                    'sec-fetch-mode': 'cors',
                    'sec-fetch-site': 'same-origin',
                    'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
                },
                'method': 'POST',
                'path': '/api/TruthSeeker/update_estimates',
                'query_params': {},
                'form_data': {
                    'param1': 'value1',
                    'param2': 'value2'
                }
            }),
            session=Session.from_dict({
                'SiID': '29a2e258e18252e2afd5ff943523f09c82f1bb9adfe382a6f33fc6a8381de898',
                'level': '1',
                'spec': '74eed1c8de06886842e235486c3c2fd6bcd60586998ac5beb87f13c0d1750e1d',
                'user_name': 'root',
                'custom_field': 'custom_value'
            }),
            session_id='0x29dd1ac0d1e30d3f'
        )

from_dict(data) classmethod

Create a RequestData instance from a dictionary.

Source code in toolboxv2/utils/system/types.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> 'RequestData':
    """Create a RequestData instance from a dictionary."""
    return cls(
        request=Request.from_dict(data.get('request', {})),
        session=Session.from_dict(data.get('session', {})),
        session_id=data.get('session_id', '')
    )

to_dict()

Convert the RequestData object back to a dictionary.

Source code in toolboxv2/utils/system/types.py

def to_dict(self) -> dict[str, Any]:
    """Convert the RequestData object back to a dictionary."""
    return {
        'request': self.request.to_dict(),
        'session': self.session.to_dict(),
        'session_id': self.session_id
    }

Security

toolboxv2.Code

Source code in toolboxv2/utils/security/cryp.py

class Code:

    @staticmethod
    def DK():
        return DEVICE_KEY

    @staticmethod
    def generate_random_string(length: int) -> str:
        """
        Generiert eine zufällige Zeichenkette der angegebenen Länge.

        Args:
            length (int): Die Länge der zu generierenden Zeichenkette.

        Returns:
            str: Die generierte Zeichenkette.
        """
        return secrets.token_urlsafe(length)

    def decode_code(self, encrypted_data, key=None):

        if not isinstance(encrypted_data, str):
            encrypted_data = str(encrypted_data)

        if key is None:
            key = DEVICE_KEY()

        return self.decrypt_symmetric(encrypted_data, key)

    def encode_code(self, data, key=None):

        if not isinstance(data, str):
            data = str(data)

        if key is None:
            key = DEVICE_KEY()

        return self.encrypt_symmetric(data, key)

    @staticmethod
    def generate_seed() -> int:
        """
        Erzeugt eine zufällige Zahl als Seed.

        Returns:
            int: Eine zufällige Zahl.
        """
        return random.randint(2 ** 32 - 1, 2 ** 64 - 1)

    @staticmethod
    def one_way_hash(text: str, salt: str = '', pepper: str = '') -> str:
        """
        Erzeugt einen Hash eines gegebenen Textes mit Salt, Pepper und optional einem Seed.

        Args:
            text (str): Der zu hashende Text.
            salt (str): Der Salt-Wert.
            pepper (str): Der Pepper-Wert.
            seed (int, optional): Ein optionaler Seed-Wert. Standardmäßig None.

        Returns:
            str: Der resultierende Hash-Wert.
        """
        return hashlib.sha256((salt + text + pepper).encode()).hexdigest()

    @staticmethod
    def generate_symmetric_key(as_str=True) -> str or bytes:
        """
        Generiert einen Schlüssel für die symmetrische Verschlüsselung.

        Returns:
            str: Der generierte Schlüssel.
        """
        key = Fernet.generate_key()
        if as_str:
            key = key.decode()
        return key

    @staticmethod
    def encrypt_symmetric(text: str or bytes, key: str) -> str:
        """
        Verschlüsselt einen Text mit einem gegebenen symmetrischen Schlüssel.

        Args:
            text (str): Der zu verschlüsselnde Text.
            key (str): Der symmetrische Schlüssel.

        Returns:
            str: Der verschlüsselte Text.
        """
        if isinstance(text, str):
            text = text.encode()
        if isinstance(key, str):
            key = key.encode()

        fernet = Fernet(key)
        return fernet.encrypt(text).decode()

    @staticmethod
    def decrypt_symmetric(encrypted_text: str, key: str, to_str=True, mute=False) -> str or bytes:
        """
        Entschlüsselt einen Text mit einem gegebenen symmetrischen Schlüssel.

        Args:
            encrypted_text (str): Der zu entschlüsselnde Text.
            key (str): Der symmetrische Schlüssel.
            to_str (bool): default true returns str if false returns bytes
        Returns:
            str: Der entschlüsselte Text.
        """

        if isinstance(key, str):
            key = key.encode()

        #try:
        fernet = Fernet(key)
        text_b = fernet.decrypt(encrypted_text)
        if not to_str:
            return text_b
        return text_b.decode()
        # except Exception as e:
        #     get_logger().error(f"Error decrypt_symmetric {e}")
        #     if not mute:
        #         raise e
        #     if not to_str:
        #         return f"Error decoding".encode()
        #     return f"Error decoding"

    @staticmethod
    def generate_asymmetric_keys() -> (str, str):
        """
        Generiert ein Paar von öffentlichen und privaten Schlüsseln für die asymmetrische Verschlüsselung.

        Args:
            seed (int, optional): Ein optionaler Seed-Wert. Standardmäßig None.

        Returns:
            (str, str): Ein Tupel aus öffentlichem und privatem Schlüssel.
        """
        private_key = rsa.generate_private_key(
            public_exponent=65537,
            key_size=2048 * 3,
        )
        public_key = private_key.public_key()

        # Serialisieren der Schlüssel
        pem_private_key = private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption()
        ).decode()

        pem_public_key = public_key.public_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PublicFormat.SubjectPublicKeyInfo
        ).decode()

        return pem_public_key, pem_private_key

    @staticmethod
    def save_keys_to_files(public_key: str, private_key: str, directory: str = "keys") -> None:
        """
        Speichert die generierten Schlüssel in separate Dateien.
        Der private Schlüssel wird mit dem Device Key verschlüsselt.

        Args:
            public_key (str): Der öffentliche Schlüssel im PEM-Format
            private_key (str): Der private Schlüssel im PEM-Format
            directory (str): Das Verzeichnis, in dem die Schlüssel gespeichert werden sollen
        """
        # Erstelle das Verzeichnis, falls es nicht existiert
        os.makedirs(directory, exist_ok=True)

        # Hole den Device Key
        device_key = DEVICE_KEY()

        # Verschlüssele den privaten Schlüssel mit dem Device Key
        encrypted_private_key = Code.encrypt_symmetric(private_key, device_key)

        # Speichere den öffentlichen Schlüssel
        public_key_path = os.path.join(directory, "public_key.pem")
        with open(public_key_path, "w") as f:
            f.write(public_key)

        # Speichere den verschlüsselten privaten Schlüssel
        private_key_path = os.path.join(directory, "private_key.pem")
        with open(private_key_path, "w") as f:
            f.write(encrypted_private_key)

        print("Saved keys in ", public_key_path)

    @staticmethod
    def load_keys_from_files(directory: str = "keys") -> (str, str):
        """
        Lädt die Schlüssel aus den Dateien.
        Der private Schlüssel wird mit dem Device Key entschlüsselt.

        Args:
            directory (str): Das Verzeichnis, aus dem die Schlüssel geladen werden sollen

        Returns:
            (str, str): Ein Tupel aus öffentlichem und privatem Schlüssel

        Raises:
            FileNotFoundError: Wenn die Schlüsseldateien nicht gefunden werden können
        """
        # Pfade zu den Schlüsseldateien
        public_key_path = os.path.join(directory, "public_key.pem")
        private_key_path = os.path.join(directory, "private_key.pem")

        # Prüfe ob die Dateien existieren
        if not os.path.exists(public_key_path) or not os.path.exists(private_key_path):
            return "", ""

        # Hole den Device Key
        device_key = DEVICE_KEY()

        # Lade den öffentlichen Schlüssel
        with open(public_key_path) as f:
            public_key = f.read()

        # Lade und entschlüssele den privaten Schlüssel
        with open(private_key_path) as f:
            encrypted_private_key = f.read()
            private_key = Code.decrypt_symmetric(encrypted_private_key, device_key)

        return public_key, private_key

    @staticmethod
    def encrypt_asymmetric(text: str, public_key_str: str) -> str:
        """
        Verschlüsselt einen Text mit einem gegebenen öffentlichen Schlüssel.

        Args:
            text (str): Der zu verschlüsselnde Text.
            public_key_str (str): Der öffentliche Schlüssel als String oder im pem format.

        Returns:
            str: Der verschlüsselte Text.
        """
        # try:
        #    public_key: RSAPublicKey = serialization.load_pem_public_key(public_key_str.encode())
        #  except Exception as e:
        #     get_logger().error(f"Error encrypt_asymmetric {e}")
        try:
            public_key: RSAPublicKey = serialization.load_pem_public_key(public_key_str.encode())
            encrypted = public_key.encrypt(
                text.encode(),
                padding.OAEP(
                    mgf=padding.MGF1(algorithm=hashes.SHA512()),
                    algorithm=hashes.SHA512(),
                    label=None
                )
            )
            return encrypted.hex()
        except Exception as e:
            get_logger().error(f"Error encrypt_asymmetric {e}")
            return "Invalid"

    @staticmethod
    def decrypt_asymmetric(encrypted_text_hex: str, private_key_str: str) -> str:
        """
        Entschlüsselt einen Text mit einem gegebenen privaten Schlüssel.

        Args:
            encrypted_text_hex (str): Der verschlüsselte Text als Hex-String.
            private_key_str (str): Der private Schlüssel als String.

        Returns:
            str: Der entschlüsselte Text.
        """
        try:
            private_key = serialization.load_pem_private_key(private_key_str.encode(), password=None)
            decrypted = private_key.decrypt(
                bytes.fromhex(encrypted_text_hex),
                padding.OAEP(
                    mgf=padding.MGF1(algorithm=hashes.SHA512()),
                    algorithm=hashes.SHA512(),
                    label=None
                )
            )
            return decrypted.decode()

        except Exception as e:
            get_logger().error(f"Error decrypt_asymmetric {e}")
        return "Invalid"

    @staticmethod
    def verify_signature(signature: str or bytes, message: str or bytes, public_key_str: str,
                         salt_length=padding.PSS.MAX_LENGTH) -> bool:
        if isinstance(signature, str):
            signature = signature.encode()
        if isinstance(message, str):
            message = message.encode()
        try:
            public_key: RSAPublicKey = serialization.load_pem_public_key(public_key_str.encode())
            public_key.verify(
                signature=signature,
                data=message,
                padding=padding.PSS(
                    mgf=padding.MGF1(hashes.SHA512()),
                    salt_length=salt_length
                ),
                algorithm=hashes.SHA512()
            )
            return True
        except:
            pass
        return False

    @staticmethod
    def verify_signature_web_algo(signature: str or bytes, message: str or bytes, public_key_str: str,
                                  algo: int = -512) -> bool:
        signature_algorithm = ECDSA(hashes.SHA512())
        if algo != -512:
            signature_algorithm = ECDSA(hashes.SHA256())

        if isinstance(signature, str):
            signature = signature.encode()
        if isinstance(message, str):
            message = message.encode()
        try:
            public_key = serialization.load_pem_public_key(public_key_str.encode())
            public_key.verify(
                signature=signature,
                data=message,
                # padding=padding.PSS(
                #    mgf=padding.MGF1(hashes.SHA512()),
                #    salt_length=padding.PSS.MAX_LENGTH
                # ),
                signature_algorithm=signature_algorithm
            )
            return True
        except:
            pass
        return False

    @staticmethod
    def create_signature(message: str, private_key_str: str, salt_length=padding.PSS.MAX_LENGTH,
                         row=False) -> str or bytes:
        try:
            private_key = serialization.load_pem_private_key(private_key_str.encode(), password=None)
            signature = private_key.sign(
                message.encode(),
                padding.PSS(
                    mgf=padding.MGF1(hashes.SHA512()),
                    salt_length=salt_length
                ),
                hashes.SHA512()
            )
            if row:
                return signature
            return base64.b64encode(signature).decode()
        except Exception as e:
            get_logger().error(f"Error create_signature {e}")
            print(e)
        return "Invalid Key"

    @staticmethod
    def pem_to_public_key(pem_key: str):
        """
        Konvertiert einen PEM-kodierten öffentlichen Schlüssel in ein PublicKey-Objekt.

        Args:
            pem_key (str): Der PEM-kodierte öffentliche Schlüssel.

        Returns:
            PublicKey: Das PublicKey-Objekt.
        """
        public_key = serialization.load_pem_public_key(pem_key.encode())
        return public_key

    @staticmethod
    def public_key_to_pem(public_key: RSAPublicKey):
        """
        Konvertiert ein PublicKey-Objekt in einen PEM-kodierten String.

        Args:
            public_key (PublicKey): Das PublicKey-Objekt.

        Returns:
            str: Der PEM-kodierte öffentliche Schlüssel.
        """
        pem = public_key.public_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PublicFormat.SubjectPublicKeyInfo
        )
        return pem.decode()

decrypt_asymmetric(encrypted_text_hex, private_key_str) staticmethod

Entschlüsselt einen Text mit einem gegebenen privaten Schlüssel.

Parameters:

Name	Type	Description	Default
encrypted_text_hex	str	Der verschlüsselte Text als Hex-String.	required
private_key_str	str	Der private Schlüssel als String.	required

Returns:

Name	Type	Description
str	str	Der entschlüsselte Text.

Source code in toolboxv2/utils/security/cryp.py

@staticmethod
def decrypt_asymmetric(encrypted_text_hex: str, private_key_str: str) -> str:
    """
    Entschlüsselt einen Text mit einem gegebenen privaten Schlüssel.

    Args:
        encrypted_text_hex (str): Der verschlüsselte Text als Hex-String.
        private_key_str (str): Der private Schlüssel als String.

    Returns:
        str: Der entschlüsselte Text.
    """
    try:
        private_key = serialization.load_pem_private_key(private_key_str.encode(), password=None)
        decrypted = private_key.decrypt(
            bytes.fromhex(encrypted_text_hex),
            padding.OAEP(
                mgf=padding.MGF1(algorithm=hashes.SHA512()),
                algorithm=hashes.SHA512(),
                label=None
            )
        )
        return decrypted.decode()

    except Exception as e:
        get_logger().error(f"Error decrypt_asymmetric {e}")
    return "Invalid"

decrypt_symmetric(encrypted_text, key, to_str=True, mute=False) staticmethod

Entschlüsselt einen Text mit einem gegebenen symmetrischen Schlüssel.

Parameters:

Name	Type	Description	Default
encrypted_text	str	Der zu entschlüsselnde Text.	required
key	str	Der symmetrische Schlüssel.	required
to_str	bool	default true returns str if false returns bytes	True

Returns: str: Der entschlüsselte Text.

Source code in toolboxv2/utils/security/cryp.py

@staticmethod
def decrypt_symmetric(encrypted_text: str, key: str, to_str=True, mute=False) -> str or bytes:
    """
    Entschlüsselt einen Text mit einem gegebenen symmetrischen Schlüssel.

    Args:
        encrypted_text (str): Der zu entschlüsselnde Text.
        key (str): Der symmetrische Schlüssel.
        to_str (bool): default true returns str if false returns bytes
    Returns:
        str: Der entschlüsselte Text.
    """

    if isinstance(key, str):
        key = key.encode()

    #try:
    fernet = Fernet(key)
    text_b = fernet.decrypt(encrypted_text)
    if not to_str:
        return text_b
    return text_b.decode()

encrypt_asymmetric(text, public_key_str) staticmethod

Verschlüsselt einen Text mit einem gegebenen öffentlichen Schlüssel.

Parameters:

Name	Type	Description	Default
text	str	Der zu verschlüsselnde Text.	required
public_key_str	str	Der öffentliche Schlüssel als String oder im pem format.	required

Returns:

Name	Type	Description
str	str	Der verschlüsselte Text.

Source code in toolboxv2/utils/security/cryp.py

@staticmethod
def encrypt_asymmetric(text: str, public_key_str: str) -> str:
    """
    Verschlüsselt einen Text mit einem gegebenen öffentlichen Schlüssel.

    Args:
        text (str): Der zu verschlüsselnde Text.
        public_key_str (str): Der öffentliche Schlüssel als String oder im pem format.

    Returns:
        str: Der verschlüsselte Text.
    """
    # try:
    #    public_key: RSAPublicKey = serialization.load_pem_public_key(public_key_str.encode())
    #  except Exception as e:
    #     get_logger().error(f"Error encrypt_asymmetric {e}")
    try:
        public_key: RSAPublicKey = serialization.load_pem_public_key(public_key_str.encode())
        encrypted = public_key.encrypt(
            text.encode(),
            padding.OAEP(
                mgf=padding.MGF1(algorithm=hashes.SHA512()),
                algorithm=hashes.SHA512(),
                label=None
            )
        )
        return encrypted.hex()
    except Exception as e:
        get_logger().error(f"Error encrypt_asymmetric {e}")
        return "Invalid"

encrypt_symmetric(text, key) staticmethod

Verschlüsselt einen Text mit einem gegebenen symmetrischen Schlüssel.

Parameters:

Name	Type	Description	Default
text	str	Der zu verschlüsselnde Text.	required
key	str	Der symmetrische Schlüssel.	required

Returns:

Name	Type	Description
str	str	Der verschlüsselte Text.

Source code in toolboxv2/utils/security/cryp.py

@staticmethod
def encrypt_symmetric(text: str or bytes, key: str) -> str:
    """
    Verschlüsselt einen Text mit einem gegebenen symmetrischen Schlüssel.

    Args:
        text (str): Der zu verschlüsselnde Text.
        key (str): Der symmetrische Schlüssel.

    Returns:
        str: Der verschlüsselte Text.
    """
    if isinstance(text, str):
        text = text.encode()
    if isinstance(key, str):
        key = key.encode()

    fernet = Fernet(key)
    return fernet.encrypt(text).decode()

generate_asymmetric_keys() staticmethod

Generiert ein Paar von öffentlichen und privaten Schlüsseln für die asymmetrische Verschlüsselung.

Parameters:

Name	Type	Description	Default
seed	int	Ein optionaler Seed-Wert. Standardmäßig None.	required

Returns:

Type	Description
(str, str)	Ein Tupel aus öffentlichem und privatem Schlüssel.

Source code in toolboxv2/utils/security/cryp.py

@staticmethod
def generate_asymmetric_keys() -> (str, str):
    """
    Generiert ein Paar von öffentlichen und privaten Schlüsseln für die asymmetrische Verschlüsselung.

    Args:
        seed (int, optional): Ein optionaler Seed-Wert. Standardmäßig None.

    Returns:
        (str, str): Ein Tupel aus öffentlichem und privatem Schlüssel.
    """
    private_key = rsa.generate_private_key(
        public_exponent=65537,
        key_size=2048 * 3,
    )
    public_key = private_key.public_key()

    # Serialisieren der Schlüssel
    pem_private_key = private_key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=serialization.NoEncryption()
    ).decode()

    pem_public_key = public_key.public_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PublicFormat.SubjectPublicKeyInfo
    ).decode()

    return pem_public_key, pem_private_key

generate_random_string(length) staticmethod

Generiert eine zufällige Zeichenkette der angegebenen Länge.

Parameters:

Name	Type	Description	Default
length	int	Die Länge der zu generierenden Zeichenkette.	required

Returns:

Name	Type	Description
str	str	Die generierte Zeichenkette.

Source code in toolboxv2/utils/security/cryp.py

@staticmethod
def generate_random_string(length: int) -> str:
    """
    Generiert eine zufällige Zeichenkette der angegebenen Länge.

    Args:
        length (int): Die Länge der zu generierenden Zeichenkette.

    Returns:
        str: Die generierte Zeichenkette.
    """
    return secrets.token_urlsafe(length)

generate_seed() staticmethod

Erzeugt eine zufällige Zahl als Seed.

Returns:

Name	Type	Description
int	int	Eine zufällige Zahl.

Source code in toolboxv2/utils/security/cryp.py

@staticmethod
def generate_seed() -> int:
    """
    Erzeugt eine zufällige Zahl als Seed.

    Returns:
        int: Eine zufällige Zahl.
    """
    return random.randint(2 ** 32 - 1, 2 ** 64 - 1)

generate_symmetric_key(as_str=True) staticmethod

Generiert einen Schlüssel für die symmetrische Verschlüsselung.

Returns:

Name	Type	Description
str	str or bytes	Der generierte Schlüssel.

Source code in toolboxv2/utils/security/cryp.py

@staticmethod
def generate_symmetric_key(as_str=True) -> str or bytes:
    """
    Generiert einen Schlüssel für die symmetrische Verschlüsselung.

    Returns:
        str: Der generierte Schlüssel.
    """
    key = Fernet.generate_key()
    if as_str:
        key = key.decode()
    return key

load_keys_from_files(directory='keys') staticmethod

Lädt die Schlüssel aus den Dateien. Der private Schlüssel wird mit dem Device Key entschlüsselt.

Parameters:

Name	Type	Description	Default
directory	str	Das Verzeichnis, aus dem die Schlüssel geladen werden sollen	'keys'

Returns:

Type	Description
(str, str)	Ein Tupel aus öffentlichem und privatem Schlüssel

Raises:

Type	Description
FileNotFoundError	Wenn die Schlüsseldateien nicht gefunden werden können

Source code in toolboxv2/utils/security/cryp.py

@staticmethod
def load_keys_from_files(directory: str = "keys") -> (str, str):
    """
    Lädt die Schlüssel aus den Dateien.
    Der private Schlüssel wird mit dem Device Key entschlüsselt.

    Args:
        directory (str): Das Verzeichnis, aus dem die Schlüssel geladen werden sollen

    Returns:
        (str, str): Ein Tupel aus öffentlichem und privatem Schlüssel

    Raises:
        FileNotFoundError: Wenn die Schlüsseldateien nicht gefunden werden können
    """
    # Pfade zu den Schlüsseldateien
    public_key_path = os.path.join(directory, "public_key.pem")
    private_key_path = os.path.join(directory, "private_key.pem")

    # Prüfe ob die Dateien existieren
    if not os.path.exists(public_key_path) or not os.path.exists(private_key_path):
        return "", ""

    # Hole den Device Key
    device_key = DEVICE_KEY()

    # Lade den öffentlichen Schlüssel
    with open(public_key_path) as f:
        public_key = f.read()

    # Lade und entschlüssele den privaten Schlüssel
    with open(private_key_path) as f:
        encrypted_private_key = f.read()
        private_key = Code.decrypt_symmetric(encrypted_private_key, device_key)

    return public_key, private_key

one_way_hash(text, salt='', pepper='') staticmethod

Erzeugt einen Hash eines gegebenen Textes mit Salt, Pepper und optional einem Seed.

Parameters:

Name	Type	Description	Default
text	str	Der zu hashende Text.	required
salt	str	Der Salt-Wert.	''
pepper	str	Der Pepper-Wert.	''
seed	int	Ein optionaler Seed-Wert. Standardmäßig None.	required

Returns:

Name	Type	Description
str	str	Der resultierende Hash-Wert.

Source code in toolboxv2/utils/security/cryp.py

@staticmethod
def one_way_hash(text: str, salt: str = '', pepper: str = '') -> str:
    """
    Erzeugt einen Hash eines gegebenen Textes mit Salt, Pepper und optional einem Seed.

    Args:
        text (str): Der zu hashende Text.
        salt (str): Der Salt-Wert.
        pepper (str): Der Pepper-Wert.
        seed (int, optional): Ein optionaler Seed-Wert. Standardmäßig None.

    Returns:
        str: Der resultierende Hash-Wert.
    """
    return hashlib.sha256((salt + text + pepper).encode()).hexdigest()

pem_to_public_key(pem_key) staticmethod

Konvertiert einen PEM-kodierten öffentlichen Schlüssel in ein PublicKey-Objekt.

Parameters:

Name	Type	Description	Default
pem_key	str	Der PEM-kodierte öffentliche Schlüssel.	required

Returns:

Name	Type	Description
PublicKey		Das PublicKey-Objekt.

Source code in toolboxv2/utils/security/cryp.py

@staticmethod
def pem_to_public_key(pem_key: str):
    """
    Konvertiert einen PEM-kodierten öffentlichen Schlüssel in ein PublicKey-Objekt.

    Args:
        pem_key (str): Der PEM-kodierte öffentliche Schlüssel.

    Returns:
        PublicKey: Das PublicKey-Objekt.
    """
    public_key = serialization.load_pem_public_key(pem_key.encode())
    return public_key

public_key_to_pem(public_key) staticmethod

Konvertiert ein PublicKey-Objekt in einen PEM-kodierten String.

Parameters:

Name	Type	Description	Default
public_key	PublicKey	Das PublicKey-Objekt.	required

Returns:

Name	Type	Description
str		Der PEM-kodierte öffentliche Schlüssel.

Source code in toolboxv2/utils/security/cryp.py

@staticmethod
def public_key_to_pem(public_key: RSAPublicKey):
    """
    Konvertiert ein PublicKey-Objekt in einen PEM-kodierten String.

    Args:
        public_key (PublicKey): Das PublicKey-Objekt.

    Returns:
        str: Der PEM-kodierte öffentliche Schlüssel.
    """
    pem = public_key.public_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PublicFormat.SubjectPublicKeyInfo
    )
    return pem.decode()

save_keys_to_files(public_key, private_key, directory='keys') staticmethod

Speichert die generierten Schlüssel in separate Dateien. Der private Schlüssel wird mit dem Device Key verschlüsselt.

Parameters:

Name	Type	Description	Default
public_key	str	Der öffentliche Schlüssel im PEM-Format	required
private_key	str	Der private Schlüssel im PEM-Format	required
directory	str	Das Verzeichnis, in dem die Schlüssel gespeichert werden sollen	'keys'

Source code in toolboxv2/utils/security/cryp.py

@staticmethod
def save_keys_to_files(public_key: str, private_key: str, directory: str = "keys") -> None:
    """
    Speichert die generierten Schlüssel in separate Dateien.
    Der private Schlüssel wird mit dem Device Key verschlüsselt.

    Args:
        public_key (str): Der öffentliche Schlüssel im PEM-Format
        private_key (str): Der private Schlüssel im PEM-Format
        directory (str): Das Verzeichnis, in dem die Schlüssel gespeichert werden sollen
    """
    # Erstelle das Verzeichnis, falls es nicht existiert
    os.makedirs(directory, exist_ok=True)

    # Hole den Device Key
    device_key = DEVICE_KEY()

    # Verschlüssele den privaten Schlüssel mit dem Device Key
    encrypted_private_key = Code.encrypt_symmetric(private_key, device_key)

    # Speichere den öffentlichen Schlüssel
    public_key_path = os.path.join(directory, "public_key.pem")
    with open(public_key_path, "w") as f:
        f.write(public_key)

    # Speichere den verschlüsselten privaten Schlüssel
    private_key_path = os.path.join(directory, "private_key.pem")
    with open(private_key_path, "w") as f:
        f.write(encrypted_private_key)

    print("Saved keys in ", public_key_path)

Modules & Flows

Module (mods) und Flows sind zu groß für eine generierte Sammelseite — siehe die jeweiligen Kapitel in der Navigation (CloudM, ISAA, Runtime).

toolboxv2.flows_dict(s='.py', remote=False, dir_path=None, flows_dict_=None, ui=False)

Source code in toolboxv2/flows/__init__.py

def flows_dict(s='.py', remote=False, dir_path=None, flows_dict_=None, ui=False):
    if s:
        s = normalize_flow_name(s)
    if flows_dict_ is None:
        flows_dict_ = {}
    with Spinner("Loading flows"):
        # Erhalte den Pfad zum aktuellen Verzeichnis
        if dir_path is None:
            for ex_path in os.getenv("EXTERNAL_PATH_RUNNABLE", '').split(','):
                if not ex_path or len(ex_path) == 0:
                    continue
                flows_dict(s,remote,ex_path,flows_dict_, ui)
            dir_path = os.path.dirname(os.path.realpath(__file__))

        # subordner-priorität: erst mini/, dann rekursiv die restlichen
        entries = os.listdir(dir_path)
        root_files = [(None, f) for f in entries
                      if os.path.isfile(os.path.join(dir_path, f))]
        subdirs = [d for d in entries
                   if os.path.isdir(os.path.join(dir_path, d)) and not d.startswith('__')]

        # mini zuerst
        ordered_subdirs = ([d for d in subdirs if d == 'mini']
                           + [d for d in subdirs if d != 'mini'])

        gezielt = s not in ('', '.py')  # konkreter flow gesucht?

        # 1. root-level files
        collect_files(flows_dict_, root_files, remote, s, dir_path, ui)
        if gezielt in flows_dict_:
            return flows_dict_

        # 2. subordner in priorität, früher abbruch bei gezielter suche
        for sub in ordered_subdirs:
            sub_path = os.path.join(dir_path, sub)
            sub_files = [(sub, f) for f in os.listdir(sub_path)
                         if os.path.isfile(os.path.join(sub_path, f))]
            collect_files(flows_dict_, sub_files, remote, s, dir_path, ui)
            if gezielt in flows_dict_:
                break

        return flows_dict_

toolboxv2.TBEF

Automatic generated by ToolBox v = 0.1.22

Other Exposed Items

toolboxv2.ToolBox_over = 'root' module-attribute