Skip to content

import_from_string

faststream.cli.utils.imports.import_from_string #

import_from_string(
    import_str: str,
) -> Tuple[Path, FastStream]

Import FastStream application from module specified by a string.

PARAMETER DESCRIPTION
import_str

A string in the format ":" specifying the module and faststream application to import.

TYPE: str

RETURNS DESCRIPTION
Tuple[Path, FastStream]

Tuple[ModuleType, FastStream]: A tuple containing the imported module and the faststream application.

RAISES DESCRIPTION
BadParameter

Raised if the given value is not of type string, if the import string is not in the format ":", if the module is not found, or if the faststream appliation is not found in the module.

Source code in faststream/cli/utils/imports.py
def import_from_string(import_str: str) -> Tuple[Path, FastStream]:
    """
    Import FastStream application from module specified by a string.

    Parameters:
        import_str (str): A string in the format "<module>:<attribute>" specifying the module and faststream application to import.

    Returns:
        Tuple[ModuleType, FastStream]: A tuple containing the imported module and the faststream application.

    Raises:
        typer.BadParameter: Raised if the given value is not of type string, if the import string is not in the format
            "<module>:<attribute>", if the module is not found, or if the faststream appliation is not found in the module.
    """
    if not isinstance(import_str, str):
        raise typer.BadParameter("Given value is not of type string")

    module_str, _, attrs_str = import_str.partition(":")
    if not module_str or not attrs_str:
        raise typer.BadParameter(
            f'Import string "{import_str}" must be in format "<module>:<attribute>"'
        )

    try:
        module = importlib.import_module(  # nosemgrep: python.lang.security.audit.non-literal-import.non-literal-import
            module_str
        )

    except ModuleNotFoundError:
        module_path, app_name = get_app_path(import_str)
        instance = try_import_app(module_path, app_name)

    else:
        attr = module
        try:
            for attr_str in attrs_str.split("."):
                attr = getattr(attr, attr_str)
            instance = attr  # type: ignore[assignment]

        except AttributeError as e:
            typer.echo(e, err=True)
            raise typer.BadParameter(
                f'Attribute "{attrs_str}" not found in module "{module_str}".'
            ) from e

        if module.__file__:
            module_path = Path(module.__file__).resolve().parent
        else:
            module_path = Path.cwd()

    return module_path, instance

Last update: 2023-11-16