class lightning_app.core.LightningFlow[source]

Bases: object

The LightningFlow is used by the LightningApp to coordinate and manage long- running jobs contained, the LightningWork.

A LightningFlow is characterized by:

  • A set of state variables.

  • Long-running jobs (LightningWork).

  • Its children LightningFlow or LightningWork with their state variables.

State variables

The LightningFlow are special classes whose attributes require to be json-serializable (e.g., int, float, bool, list, dict, …).

They also may not reach into global variables unless they are constant.

The attributes need to be all defined in __init__ method, and eventually assigned to different values throughout the lifetime of the object. However, defining new attributes outside of __init__ is not allowed.

Attributes taken together represent the state of the component. Components are capable of retrieving their state and that of their children recursively at any time. They are also capable of setting an externally provided state recursively to its children.

Execution model and work

The entry point for execution is the run method at the root component. The run method of the root component may call the run method of its children, and the children may call the run methods of their children and so on.

The run method of the root component is called repeatedly in a while loop forever until the app gets terminated. In this programming model (reminiscent of React, Vue or Streamlit from the JavaScript world), the values of the state variables, or their changes, are translated into actions throughout the component hierarchy. This means the flow of execution will only be affected by state changes in a component or one of its children, and otherwise remain idempotent.

The actions themselves are self-contained within LightningWork. The LightningWork are typically used for long-running jobs, like downloading a dataset, performing a query, starting a computationally heavy script. While one may access any state variable in a LightningWork from a LightningFlow, one may not directly call methods of other components from within a LightningWork as LightningWork can’t have any children. This limitation allows applications to be distributed at scale.

Component hierarchy and App

Given the above characteristics, a root LightningFlow, potentially containing children components, can be passed to an App object and its execution can be distributed (each LightningWork will be run within its own process or different arrangements).


>>> from lightning_app import LightningFlow
>>> class RootFlow(LightningFlow):
...     def __init__(self):
...         super().__init__()
...         self.counter = 0
...     def run(self):
...         self.counter += 1
>>> flow = RootFlow()
>>> assert flow.counter == 1
>>> assert flow.state["vars"]["counter"] == 1
static _attach_backend(flow, backend)[source]

Attach the backend to all flows and its children.


Used to exit the application.

Private method.

Deprecated since version 1.9.0: This function is deprecated and will be removed in 2.0.0. Use stop() instead.

Return type


static _is_state_attribute(name)[source]

Every public attribute is part of the state by default and all protected (prefixed by ‘_’) or private (prefixed by ‘__’) attributes are not.

Exceptions are listed in the _INTERNAL_STATE_VARS class variable.

Return type



Configure the API routes of the LightningFlow.

Returns a list of HttpMethod such as Post or Get.

from lightning_app import LightningFlow
from lightning_app.api import Post

from pydantic import BaseModel

class HandlerModel(BaseModel):
    name: str

class Flow(L.LightningFlow):
    def __init__(self):
        self.names = []

    def handler(self, config: HandlerModel) -> None:

    def configure_api(self):
        return [Post("/v1/api/request", self.handler)]

Once the app is running, you can access the Swagger UI of the app under the /docs route.


Configure the commands of this LightningFlow.

Returns a list of dictionaries mapping a command name to a flow method.

class Flow(LightningFlow):
    def __init__(self):
        self.names = []

    def configure_commands(self):
        return {"my_command_name": self.my_remote_method}

    def my_remote_method(self, name):

Once the app is running with the following command:

lightning run app
lightning my_command_name --args name=my_own_name

Configure the UI layout of this LightningFlow.

You can either

  1. Return a single Frontend object to serve a user interface for this Flow.

  2. Return a single dictionary to expose the UI of a child flow.

  3. Return a list of dictionaries to arrange the children of this flow in one or multiple tabs.

Example: Serve a static directory (with at least a file index.html inside).

from lightning_app.frontend import StaticWebFrontend

class Flow(LightningFlow):

    def configure_layout(self):
        return StaticWebFrontend("path/to/folder/to/serve")

Example: Serve a streamlit UI (needs the streamlit package to be installed).

from lightning_app.frontend import StaticWebFrontend

class Flow(LightningFlow):

    def configure_layout(self):
        return StreamlitFrontend(render_fn=my_streamlit_ui)

def my_streamlit_ui(state):
    # add your streamlit code here!
    import streamlit as st

Example: Arrange the UI of my children in tabs (default UI by Lightning).

class Flow(LightningFlow):
    def configure_layout(self):
        return [
            dict(name="First Tab", content=self.child0),
            dict(name="Second Tab", content=self.child1),
            dict(name="Lightning", content=""),

If you don’t implement configure_layout, Lightning will collect all children and display their UI in a tab (if they have their own configure_layout implemented).


This hook gets called at the time of app creation and then again as part of the loop. If desired, the returned layout configuration can depend on the state. The only exception are the flows that return a Frontend. These need to be provided at the time of app creation in order for the runtime to start the server.

Learn more about adding UI

Return type

Union[Dict[str, Any], List[Dict[str, Any]], Frontend]

experimental_iterate(iterable, run_once=True, user_key='')[source]

This method should always be used with any kind of iterable to ensure its fault tolerant.

If you want your iterable to always be consumed from scratch, you shouldn’t use this method.

  • iterable (Iterable) – Iterable to iterate over. The iterable shouldn’t have side effects or be random.

  • run_once (bool) – Whether to run the entire iteration only once. Otherwise, it would restart from the beginning.

  • user_key (str) – Key to be used to track the caching mechanism.

Return type


load_state_dict(flow_state, children_states, strict=True)[source]

Reloads the state of this flow and its children.

import lightning as L

class Work(L.LightningWork):
    def __init__(self):
        self.counter = 0

    def run(self):
        self.counter += 1

class Flow(L.LightningFlow):
    def run(self):
        # dynamically create a work.
        if not getattr(self, "w", None):
            self.w = WorkReload()

    def load_state_dict(self, flow_state, children_states, strict) -> None:
        # 1: Re-instantiate the dynamic work
        self.w = Work()

        # 2: Make any states modification / migration.

        # 3: Call the parent ``load_state_dict`` to
        # recursively reload the states.
  • flow_state (Dict[str, Any]) – The state of the current flow.

  • children_states (Dict[str, Any]) – The state of the dynamic children of this flow.

  • strict (bool) – Whether to raise an exception if a dynamic children hasn’t been re-created.

Return type



Return its LightningWork with their names.

Return type

List[Tuple[str, LightningWork]]

run(*args, **kwargs)[source]

Override with your own logic.

Return type


schedule(cron_pattern, start_time=None, user_key=None)[source]

The schedule method is used to run a part of the flow logic on timely manner.

from lightning_app import LightningFlow

class Flow(LightningFlow):
    def run(self):
        if self.schedule("hourly"):
            print("run some code every hour")

A best practice is to avoid running a dynamic flow or work under the self.schedule method. Instead, instantiate them within the condition, but run them outside.

from lightning_app import LightningFlow
from lightning_app.structures import List

class SchedulerDAG(LightningFlow):
    def __init__(self):
        self.dags = List()

    def run(self):
        if self.schedule("hourly"):

        for dag in self.dags:
            payload =

Learn more about Scheduling

Return type


set_state(provided_state, recurse=True)[source]

Method to set the state to this LightningFlow, its children and LightningWork.

  • provided_state (Dict) – The state to be reloaded

  • recurse (bool) – Whether to apply the state down children.

Return type



Returns the current flow state but not its children.


Method used to exit the application.

Return type



Return its LightningWork.

Return type


property flows: Dict[str, lightning_app.core.flow.LightningFlow][source]

Return its children LightningFlow.

Return type

Dict[str, LightningFlow]

property lightningignore: Tuple[str, ...][source]

Programmatic equivalent of the .lightningignore file.

Return type

Tuple[str, ...]

property name[source]

Return the current LightningFlow name.

property ready: bool[source]

Override to customize when your App should be ready.

Return type


property state[source]

Returns the current flow state along its children.