(
    config: Union[dict, RootConfig] = None,
    *,
    config_file: str = None,
    service_name: str = None,
    process_name: str = None,
    features: List[Union[Feature, str]] = ALL_FEATURES,
    client_class: Type[BusClient] = BusClient,
    node_class: Type[BusPath] = BusPath,
    plugins=None,
    flask: bool = False,
    hook_registry: Optional[HookRegistry] = None,
    api_registry: Optional[ApiRegistry] = None,
    **kwargs,
)
return client
Create a new bus instance which can be used to access the bus.

    Typically this will be used as follows:

        import lightbus

        bus = lightbus.create()

    This will be a `BusPath` instance. If you wish to access the lower
    level `BusClient` you can do so via `bus.client`.

    Args:
        config (dict, Config): The config object or dictionary to load
        config_file (str): The path to a config file to load (should end in .json or .yaml)
        service_name (str): The name of this service - will be used when creating event consumer groups
        process_name (str): The unique name of this process - used when retrieving unprocessed events following a crash
        client_class (Type[BusClient]): The class from which the bus client will be instantiated
        node_class (BusPath): The class from which the bus path will be instantiated
        plugins (list): A list of plugin instances to load
        flask (bool): Are we using flask? If so we will make sure we don't start lightbus in the reloader process
        **kwargs (): Any additional instantiation arguments to be passed to `client_class`.

    Returns: BusPath
Create a new bus instance which can be used to access the bus.
Create a new bus instance which can be used to access the bus .
def create_client(
    config: Union[dict, RootConfig] = None,
    *,
    config_file: str = None,
    service_name: str = None,
    process_name: str = None,
    features: List[Union[Feature, str]] = ALL_FEATURES,
    client_class: Type[BusClient] = BusClient,
    node_class: Type[BusPath] = BusPath,
    plugins=None,
    flask: bool = False,
    hook_registry: Optional[HookRegistry] = None,
    api_registry: Optional[ApiRegistry] = None,
    **kwargs,
) -> BusClient:
    """
    Create a new bus instance which can be used to access the bus.

    Typically this will be used as follows:

        import lightbus

        bus = lightbus.create()

    This will be a `BusPath` instance. If you wish to access the lower
    level `BusClient` you can do so via `bus.client`.

    Args:
        config (dict, Config): The config object or dictionary to load
        config_file (str): The path to a config file to load (should end in .json or .yaml)
        service_name (str): The name of this service - will be used when creating event consumer groups
        process_name (str): The unique name of this process - used when retrieving unprocessed events following a crash
        client_class (Type[BusClient]): The class from which the bus client will be instantiated
        node_class (BusPath): The class from which the bus path will be instantiated
        plugins (list): A list of plugin instances to load
        flask (bool): Are we using flask? If so we will make sure we don't start lightbus in the reloader process
        **kwargs (): Any additional instantiation arguments to be passed to `client_class`.

    Returns: BusPath

    """
    if flask:
        in_flask_server = sys.argv[0].endswith("flask") and "run" in sys.argv
        if in_flask_server and os.environ.get("WERKZEUG_RUN_MAIN", "").lower() != "true":
            # Flask has a reloader process that shouldn't start a lightbus client
            return

    # Ensure an event loop exists, as creating InternalQueue
    # objects requires that we have one.
    get_event_loop()

    # If were are running via the Lightbus CLI then we may have
    # some command line arguments we need to apply.
    # pylint: disable=cyclic-import,import-outside-toplevel
    from lightbus.commands import COMMAND_PARSED_ARGS

    config_file = COMMAND_PARSED_ARGS.get("config_file", None) or config_file
    service_name = COMMAND_PARSED_ARGS.get("service_name", None) or service_name
    process_name = COMMAND_PARSED_ARGS.get("process_name", None) or process_name

    if config is None:
        config = load_config(
            from_file=config_file, service_name=service_name, process_name=process_name
        )

    if isinstance(config, Mapping):
        config = Config.load_dict(config or {})
    elif isinstance(config, RootConfig):
        config = Config(config)

    transport_registry = kwargs.pop("transport_registry", None) or TransportRegistry().load_config(
        config
    )

    schema = Schema(
        schema_transport=transport_registry.get_schema_transport(),
        max_age_seconds=config.bus().schema.ttl,
        human_readable=config.bus().schema.human_readable,
    )

    error_queue: ErrorQueueType = InternalQueue()

    # Plugin registry

    plugin_registry = PluginRegistry()
    if plugins is None:
        logger.debug("Auto-loading any installed Lightbus plugins...")
        plugin_registry.autoload_plugins(config)
    else:
        logger.debug("Loading explicitly specified Lightbus plugins....")
        plugin_registry.set_plugins(plugins)

    # Hook registry
    if not hook_registry:
        hook_registry = HookRegistry(
            error_queue=error_queue, execute_plugin_hooks=plugin_registry.execute_hook
        )

    # API registry
    if not api_registry:
        api_registry = ApiRegistry()
        api_registry.add(LightbusStateApi())
        api_registry.add(LightbusMetricsApi())

    events_queue_client_to_dock = InternalQueue()
    events_queue_dock_to_client = InternalQueue()

    event_client = EventClient(
        api_registry=api_registry,
        hook_registry=hook_registry,
        config=config,
        schema=schema,
        error_queue=error_queue,
        consume_from=events_queue_dock_to_client,
        produce_to=events_queue_client_to_dock,
    )

    event_dock = EventDock(
        transport_registry=transport_registry,
        api_registry=api_registry,
        config=config,
        error_queue=error_queue,
        consume_from=events_queue_client_to_dock,
        produce_to=events_queue_dock_to_client,
    )

    rpcs_queue_client_to_dock = InternalQueue()
    rpcs_queue_dock_to_client = InternalQueue()

    rpc_result_client = RpcResultClient(
        api_registry=api_registry,
        hook_registry=hook_registry,
        config=config,
        schema=schema,
        error_queue=error_queue,
        consume_from=rpcs_queue_dock_to_client,
        produce_to=rpcs_queue_client_to_dock,
    )

    rpc_result_dock = RpcResultDock(
        transport_registry=transport_registry,
        api_registry=api_registry,
        config=config,
        error_queue=error_queue,
        consume_from=rpcs_queue_client_to_dock,
        produce_to=rpcs_queue_dock_to_client,
    )

    client = client_class(
        config=config,
        hook_registry=hook_registry,
        plugin_registry=plugin_registry,
        features=features,
        schema=schema,
        api_registry=api_registry,
        event_client=event_client,
        rpc_result_client=rpc_result_client,
        error_queue=error_queue,
        transport_registry=transport_registry,
        **kwargs,
    )

    # Pass the client to any hooks
    # (use a weakref to prevent circular references)
    hook_registry.set_extra_parameter("client", weakref.proxy(client))

    # We don't do this normally as the docks do not need to be
    # accessed directly, but this is useful in testing
    # TODO: Testing flag removed, but these are only needed in testing.
    #       Perhaps wrap them up in a way that makes this obvious
    client.event_dock = event_dock
    client.rpc_result_dock = rpc_result_dock

    # Wait until the docks are ready to go
    if not get_event_loop().is_running():
        # If the event loop is running then we can assume that this setup
        # will happen momentarily anyway. Plus, if an event loop is running
        # then we cannot block. This is mainly to make sure everything
        # starts up in a sensible fashion when we are in non-worker-mode.
        block(event_dock.wait_until_ready(), timeout=2)
        block(rpc_result_dock.wait_until_ready(), timeout=2)

    return client
def create_client ( config : Union [ dict , RootConfig ] = None , * , config_file : str = None , service_name : str = None , process_name : str = None , features : List [ Union [ Feature , str ] ] = ALL_FEATURES , client_class : Type [ BusClient ] = BusClient , node_class : Type [ BusPath ] = BusPath , plugins = None , flask : bool = False , hook_registry : Optional [ HookRegistry ] = None , api_registry : Optional [ ApiRegistry ] = None , * * kwargs , ) -> BusClient : if flask : in_flask_server = sys . argv [ 0 ] . endswith ( "flask" ) and "run" in sys . argv if in_flask_server and os . environ . get ( "WERKZEUG_RUN_MAIN" , "" ) . lower ( ) != "true" : # Flask has a reloader process that shouldn't start a lightbus client return # Ensure an event loop exists, as creating InternalQueue # objects requires that we have one. get_event_loop ( ) # If were are running via the Lightbus CLI then we may have # some command line arguments we need to apply. # pylint: disable=cyclic-import,import-outside-toplevel from lightbus . commands import COMMAND_PARSED_ARGS config_file = COMMAND_PARSED_ARGS . get ( "config_file" , None ) or config_file service_name = COMMAND_PARSED_ARGS . get ( "service_name" , None ) or service_name process_name = COMMAND_PARSED_ARGS . get ( "process_name" , None ) or process_name if config is None : config = load_config ( from_file = config_file , service_name = service_name , process_name = process_name ) if isinstance ( config , Mapping ) : config = Config . load_dict ( config or { } ) elif isinstance ( config , RootConfig ) : config = Config ( config ) transport_registry = kwargs . pop ( "transport_registry" , None ) or TransportRegistry ( ) . load_config ( config ) schema = Schema ( schema_transport = transport_registry . get_schema_transport ( ) , max_age_seconds = config . bus ( ) . schema . ttl , human_readable = config . bus ( ) . schema . human_readable , ) error_queue : ErrorQueueType = InternalQueue ( ) # Plugin registry plugin_registry = PluginRegistry ( ) if plugins is None : logger . debug ( "Auto-loading any installed Lightbus plugins..." ) plugin_registry . autoload_plugins ( config ) else : logger . debug ( "Loading explicitly specified Lightbus plugins...." ) plugin_registry . set_plugins ( plugins ) # Hook registry if not hook_registry : hook_registry = HookRegistry ( error_queue = error_queue , execute_plugin_hooks = plugin_registry . execute_hook ) # API registry if not api_registry : api_registry = ApiRegistry ( ) api_registry . add ( LightbusStateApi ( ) ) api_registry . add ( LightbusMetricsApi ( ) ) events_queue_client_to_dock = InternalQueue ( ) events_queue_dock_to_client = InternalQueue ( ) event_client = EventClient ( api_registry = api_registry , hook_registry = hook_registry , config = config , schema = schema , error_queue = error_queue , consume_from = events_queue_dock_to_client , produce_to = events_queue_client_to_dock , ) event_dock = EventDock ( transport_registry = transport_registry , api_registry = api_registry , config = config , error_queue = error_queue , consume_from = events_queue_client_to_dock , produce_to = events_queue_dock_to_client , ) rpcs_queue_client_to_dock = InternalQueue ( ) rpcs_queue_dock_to_client = InternalQueue ( ) rpc_result_client = RpcResultClient ( api_registry = api_registry , hook_registry = hook_registry , config = config , schema = schema , error_queue = error_queue , consume_from = rpcs_queue_dock_to_client , produce_to = rpcs_queue_client_to_dock , ) rpc_result_dock = RpcResultDock ( transport_registry = transport_registry , api_registry = api_registry , config = config , error_queue = error_queue , consume_from = rpcs_queue_client_to_dock , produce_to = rpcs_queue_dock_to_client , ) client = client_class ( config = config , hook_registry = hook_registry , plugin_registry = plugin_registry , features = features , schema = schema , api_registry = api_registry , event_client = event_client , rpc_result_client = rpc_result_client , error_queue = error_queue , transport_registry = transport_registry , * * kwargs , ) # Pass the client to any hooks # (use a weakref to prevent circular references) hook_registry . set_extra_parameter ( "client" , weakref . proxy ( client ) ) # We don't do this normally as the docks do not need to be # accessed directly, but this is useful in testing # TODO: Testing flag removed, but these are only needed in testing. #       Perhaps wrap them up in a way that makes this obvious client . event_dock = event_dock client . rpc_result_dock = rpc_result_dock # Wait until the docks are ready to go if not get_event_loop ( ) . is_running ( ) : # If the event loop is running then we can assume that this setup # will happen momentarily anyway. Plus, if an event loop is running # then we cannot block. This is mainly to make sure everything # starts up in a sensible fashion when we are in non-worker-mode. block ( event_dock . wait_until_ready ( ) , timeout = 2 ) block ( rpc_result_dock . wait_until_ready ( ) , timeout = 2 ) return client