Database

class lightning.app.components.database.server.Database(models, db_filename='database.db', store_interval=10, debug=False)

Bases: LightningWork

The Database Component enables to interact with an SQLite database to store some structured information about your application.

The provided models are SQLModel tables

Parameters

• models (Union[Type[object], List[Type[object]]]) – A SQLModel or a list of SQLModels table to be added to the database.

• db_filename (str) – The name of the SQLite database.

• store_interval (int) – Time interval (in seconds) at which the database is periodically synchronized to the Drive. Note that the database is also always synchronized on exit.

• debug (bool) – Whether to run the database in debug mode.

Example:

from typing import List
from sqlmodel import SQLModel, Field
from uuid import uuid4

from lightning.app import LightningFlow, LightningApp
from lightning.app.components.database import Database, DatabaseClient

class CounterModel(SQLModel, table=True):
    __table_args__ = {"extend_existing": True}

    id: int = Field(default=None, primary_key=True)
    count: int


class Flow(LightningFlow):

    def __init__(self):
        super().__init__()
        self._private_token = uuid4().hex
        self.db = Database(models=[CounterModel])
        self._client = None
        self.counter = 0

    def run(self):
        self.db.run(token=self._private_token)

        if not self.db.alive():
            return

        if self.counter == 0:
            self._client = DatabaseClient(
                model=CounterModel,
                db_url=self.db.url,
                token=self._private_token,
            )

        rows = self._client.select_all()

        print(f"{self.counter}: {rows}")

        if not rows:
            self._client.insert(CounterModel(count=0))
        else:
            row: CounterModel = rows[0]
            row.count += 1
            self._client.update(row)

        if self.counter >= 100:
            row: CounterModel = rows[0]
            self._client.delete(row)
            self.stop()

        self.counter += 1

app = LightningApp(Flow())

If you want to use nested SQLModels, we provide a utility to do so as follows:

Example:

from typing import List
from sqlmodel import SQLModel, Field
from sqlalchemy import Column

from lightning.app.components.database.utilities import pydantic_column_type

class KeyValuePair(SQLModel):
    name: str
    value: str

class CounterModel(SQLModel, table=True):
    __table_args__ = {"extend_existing": True}

    name: int = Field(default=None, primary_key=True)

    # RIGHT THERE ! You need to use Field and Column with the `pydantic_column_type` utility.
    kv: List[KeyValuePair] = Field(..., sa_column=Column(pydantic_column_type(List[KeyValuePair])))

alive()

Hack: Returns whether the server is alive.

Return type

bool

on_exit()

Override this hook to add your logic when the work is exiting.

Note: This hook is not guaranteed to be called when running in the cloud.

run(token=None)

Parameters

token (Optional[str]) – Token used to protect the database access. Ensure you don’t expose it through the App State.

Return type

None