erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

tools (dynamic_import): add an argument to select bridge type:

Browse source 
If a frontend is using Twisted it must use the `TwistedBridge` instead of `AIOBridge`. A
new argument has been added to let do that.
This commit is contained in:
Goffi 2026-03-13 11:26:57 +01:00
commit e82a946123
1 changed files with 22 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

33
libervia/backend/tools/common/dynamic_import.py
View file

@ -20,7 +20,7 @@
from importlib import import_module
from types import ModuleType
from typing import Any, Literal
from typing import Any, Literal, LiteralString
from libervia.backend.core import exceptions
from libervia.backend.core.log import getLogger
@ -53,7 +53,7 @@ def import_bridge(name: str) -> ModuleType | None:
return bridge_module
def _load_bridge(bridge_settings: BridgeSettings, verbose: bool = True):
def _load_bridge(bridge_settings: BridgeSettings, bridge_type, verbose: bool):
"""Load requested bridge.
@raise BridgeInitError: Bridge can't be initialised.
@ -69,7 +69,7 @@ def _load_bridge(bridge_settings: BridgeSettings, verbose: bool = True):
# FIXME: temporary class selection, bridges need to be refactored to always use async
# classes (and to not use a class starting with a lower case)
try:
bridge_cls = bridge_module.AIOBridge
bridge_cls = getattr(bridge_module, bridge_type)
except AttributeError:
bridge_cls = bridge_module.bridge
try:
@ -81,7 +81,9 @@ def _load_bridge(bridge_settings: BridgeSettings, verbose: bool = True):
raise exceptions.BridgeInitError from e
def load_bridge(config: Settings, verbose: bool = True) -> tuple[Any, BridgeSettings]:
def load_bridge(
config: Settings, verbose: bool = True, bridge_type="AIOBridge"
) -> tuple[Any, BridgeSettings]:
"""Load and initialise the bridge.
"auto" bridge is handled
@ -99,7 +101,7 @@ def load_bridge(config: Settings, verbose: bool = True) -> tuple[Any, BridgeSett
try:
bridge_settings = DBusBridgeSettings()
config.bridges = [bridge_settings]
bridge = _load_bridge(bridge_settings, verbose=verbose)
bridge = _load_bridge(bridge_settings, bridge_type, verbose=verbose)
except exceptions.BridgeInitError as e:
if verbose:
log.warning(f"Can't load D-Bus bridge: {e}")
@ -108,13 +110,13 @@ def load_bridge(config: Settings, verbose: bool = True) -> tuple[Any, BridgeSett
config.bridges = [bridge_settings]
if bridge is None:
bridge = _load_bridge(bridge_settings, verbose=verbose)
bridge = _load_bridge(bridge_settings, bridge_type, verbose=verbose)
return bridge, bridge_settings
async def _load_and_init_bridge(
bridge_settings: BridgeSettings, verbose: bool, is_backend: bool
bridge_settings: BridgeSettings, bridge_type: str, verbose: bool, is_backend: bool
):
"""Load and set requested bridge.
@ -122,7 +124,7 @@ async def _load_and_init_bridge(
@raise BridgeExceptionNoService: Bridge is initialised, but the backend can't be
found.
"""
bridge = _load_bridge(bridge_settings, verbose)
bridge = _load_bridge(bridge_settings, bridge_type, verbose)
try:
bridge_post_init = bridge.post_init
except AttributeError:
@ -146,13 +148,18 @@ async def _load_and_init_bridge(
async def load_and_init_bridge(
config: Settings, verbose: bool = True, is_backend=False
config: Settings,
verbose: bool = True,
bridge_type: Literal["AIOBridge", "TwistedBridge"] = "AIOBridge",
is_backend=False,
) -> tuple[Any, BridgeSettings]:
"""Load and initialise the bridge.
"auto" bridge is handled
@param config: Libervia Settings
@param verbose: Logs when a Bridge can't be initialised if True.
@param bridge_type: Bridge class to use.
@param is_backend: Set to True if calling from backend.
@return: initialised bridge and bridge settings
@raise BridgeInitError: Bridge can't be initialised.
@ -172,7 +179,9 @@ async def load_and_init_bridge(
try:
bridge_settings = DBusBridgeSettings()
config.bridges = [bridge_settings]
bridge = await _load_and_init_bridge(bridge_settings, verbose, is_backend)
bridge = await _load_and_init_bridge(
bridge_settings, bridge_type, verbose, is_backend
)
except exceptions.BridgeInitError as e:
if verbose:
log.warning(f"Can't load D-Bus bridge: {e}")
@ -181,6 +190,8 @@ async def load_and_init_bridge(
config.bridges = [bridge_settings]
if bridge is None:
bridge = await _load_and_init_bridge(bridge_settings, verbose, is_backend)
bridge = await _load_and_init_bridge(
bridge_settings, bridge_type, verbose, is_backend
)
return bridge, bridge_settings