Class-Based NiceGUI Pages and Integrations: A Practical Guide¶

Objective: Master building structured NiceGUI applications with class-based pages, dependency injection, and clean architecture. When you need to build real applications—not prototypes—with auth, dashboards, API integrations, and maintainable code, this tutorial becomes your weapon of choice.

Most NiceGUI examples are single-file fever dreams: everything in main.py, global state drifting around like cosmic dust. That's fine for a quick prototype, but the moment you want a real app—auth, dashboards, API integrations—you need structure.

This tutorial shows how to build class-based NiceGUI pages with:

A clean project layout
Reusable page classes and components
Centralized routing
Integration points for external services (APIs, DBs, background workers)

Prerequisites¶

Python >= 3.10
NiceGUI >= 2.x
Basic understanding of Python classes and async/await

1. Project Layout¶

Let's start with a small but scalable layout:

nicegui-app/
├── app.py                # entrypoint
├── core/
│   ├── __init__.py
│   ├── app_factory.py    # create_app() lives here
│   └── services.py       # "business logic", API clients, db, etc
├── pages/
│   ├── __init__.py
│   ├── base.py           # BasePage class
│   ├── home.py           # HomePage
│   ├── dashboard.py      # DashboardPage
│   └── form_demo.py      # FormPage example
└── requirements.txt

requirements.txt¶

nicegui>=2.0.0
httpx>=0.27.0

2. The Base Page Pattern¶

We want each page to be a class that:

Knows its own route path
Encapsulates UI construction (build(), layout(), etc.)
Optionally injects shared services (API clients, repositories)
Can be registered in a central router

2.1 pages/base.py¶

# pages/base.py
from __future__ import annotations

from abc import ABC, abstractmethod
from typing import Optional, Any

from nicegui import ui


class BasePage(ABC):
    """Abstract base for all pages.

    Responsibilities:
      - Define `route` (e.g., "/dashboard")
      - Provide `register(app)` to hook into NiceGUI routing
      - Implement `build()` to construct content
    """

    # Route path, to be overridden
    route: str = "/"

    # Optional title; can be used to set browser title, header, etc.
    title: str = "Untitled"

    def __init__(self, services: Optional[dict[str, Any]] = None) -> None:
        self.services = services or {}

    @classmethod
    def register(cls, app, services: Optional[dict[str, Any]] = None) -> None:
        """Attach this page to the NiceGUI app router."""
        page_instance = cls(services=services)

        @ui.page(cls.route)
        def _() -> None:
            page_instance.render()

    def render(self) -> None:
        """Default render flow: layout + build."""
        with ui.column().classes("w-full items-stretch gap-4 p-4"):
            self._render_header()
            self.build()

    def _render_header(self) -> None:
        with ui.row().classes("w-full justify-between items-center"):
            ui.label(self.title).classes("text-2xl font-bold")
            # Hook for page-level actions (buttons, filters, etc.)
            self.render_actions()

    def render_actions(self) -> None:
        """Override in subclasses for header actions."""
        ...

    @abstractmethod
    def build(self) -> None:
        """Subclasses must implement to build the page contents."""
        raise NotImplementedError

This gives us a template:

route: where it lives
title: display / browser / header
render(): calls _render_header() + build()
register(app, services): binds it to @ui.page

3. Creating a Simple Home Page¶

3.1 pages/home.py¶

# pages/home.py
from __future__ import annotations

from nicegui import ui
from .base import BasePage


class HomePage(BasePage):
    route = "/"
    title = "Welcome to the NiceGUI App"

    def render_actions(self) -> None:
        # Example header actions
        with ui.row().classes("gap-2"):
            ui.button("Dashboard", on_click=lambda: ui.open("/dashboard"))
            ui.button("Form Demo", on_click=lambda: ui.open("/form-demo"))

    def build(self) -> None:
        ui.label("This is the home page, built from a class.").classes("text-lg")
        ui.markdown(
            """
            - Navigate using the header buttons
            - Each page is defined as a Python class
            - Shared services are injected via the BasePage constructor
            """
        )

        with ui.row().classes("gap-4 mt-4"):
            with ui.card():
                ui.label("Quick Card")
                ui.label("You can put any UI components in build().")

            with ui.card():
                ui.label("Another Card")
                ui.button("Click me", on_click=lambda: ui.notify("Clicked on home page!"))

4. Injecting Services: API / DB / Whatever¶

We don't want each page to instantiate its own HTTP client or database connection. Instead, we define services in one place and pass them down.

4.1 core/services.py¶

# core/services.py
from __future__ import annotations

from dataclasses import dataclass
from typing import Any
import httpx


@dataclass
class ApiClient:
    base_url: str

    async def get_status(self) -> dict[str, Any]:
        async with httpx.AsyncClient(base_url=self.base_url, timeout=5.0) as client:
            r = await client.get("/status")
            r.raise_for_status()
            return r.json()


def create_services() -> dict[str, Any]:
    """Construct and return the service container."""
    api_client = ApiClient(base_url="https://httpbin.org")
    # Extend here with db clients, redis, etc.
    return {
        "api": api_client,
    }

This gives us a simple "service container" dictionary.

5. A Dashboard Page Using Services¶

Now let's build a page that calls an API and renders a table / cards.

5.1 pages/dashboard.py¶

# pages/dashboard.py
from __future__ import annotations

from typing import Any
from nicegui import ui
from .base import BasePage


class DashboardPage(BasePage):
    route = "/dashboard"
    title = "Dashboard"

    async def _refresh_status(self, label: ui.label) -> None:
        api = self.services["api"]
        label.text = "Loading..."
        try:
            data: dict[str, Any] = await api.get_status()
            label.text = f"API status: {data.get('url', 'unknown')}"
            ui.notify("Status refreshed", type="positive")
        except Exception as e:  # noqa: BLE001
            label.text = "Error while fetching status"
            ui.notify(f"Error: {e}", type="negative")

    def build(self) -> None:
        ui.label("Dashboard overview").classes("text-lg mb-2")

        with ui.card().classes("w-full max-w-xl"):
            status_label = ui.label("API status: unknown").classes("mb-2")
            ui.button(
                "Refresh status",
                on_click=lambda: self._refresh_status(status_label),
            )

        with ui.row().classes("gap-4 mt-4"):
            with ui.card():
                ui.label("Metric A")
                ui.label("42")

            with ui.card():
                ui.label("Metric B")
                ui.label("1337")

Note: - We access self.services["api"] - We use an async helper _refresh_status and hook it with on_click - NiceGUI is fine with async callables

6. A Form Page with Validation¶

Let's add another class-based page that showcases forms, validation, and interactions.

6.1 pages/form_demo.py¶

# pages/form_demo.py
from __future__ import annotations

from nicegui import ui
from .base import BasePage


class FormDemoPage(BasePage):
    route = "/form-demo"
    title = "Form Demo"

    def build(self) -> None:
        ui.label("User Input Form").classes("text-lg mb-2")

        with ui.card().classes("w-full max-w-lg"):
            name = ui.input("Name").props("outlined").classes("mb-2")
            email = ui.input("Email").props("outlined").classes("mb-2")
            agree = ui.checkbox("I agree to the terms").classes("mb-2")

            result = ui.label("").classes("mt-2 text-sm text-gray-500")

            def submit() -> None:
                errors = []
                if not name.value:
                    errors.append("Name is required.")
                if not email.value or "@" not in email.value:
                    errors.append("Valid email is required.")
                if not agree.value:
                    errors.append("You must agree to the terms.")

                if errors:
                    ui.notify("\n".join(errors), type="negative")
                    result.text = "Errors: " + "; ".join(errors)
                else:
                    ui.notify("Form submitted!", type="positive")
                    result.text = f"Submitted: {name.value} <{email.value}>"

            ui.button("Submit", on_click=submit)

7. Wiring It All Together: App Factory¶

Now we need a single app factory that:

Builds the NiceGUI app
Creates the shared services
Registers all page classes

7.1 core/app_factory.py¶

# core/app_factory.py
from __future__ import annotations

from nicegui import ui
from typing import Type

from pages.base import BasePage
from pages.home import HomePage
from pages.dashboard import DashboardPage
from pages.form_demo import FormDemoPage
from core.services import create_services


def create_app():
    """Create and configure the NiceGUI app with class-based pages."""
    services = create_services()

    # List of page classes to register
    page_classes: list[Type[BasePage]] = [
        HomePage,
        DashboardPage,
        FormDemoPage,
    ]

    # Register each page
    for page_cls in page_classes:
        page_cls.register(ui, services=services)

    # Optional: global header/footer, default layouts, etc.
    # For example, a simple top-level menu on every page via NiceGUI's `ui.header`:
    @ui.header()
    def header():
        with ui.row().classes("w-full justify-between items-center px-4 py-2"):
            ui.label("NiceGUI Class-Based App").classes("text-lg font-semibold")
            with ui.row().classes("gap-2"):
                ui.button("Home", on_click=lambda: ui.open("/"))
                ui.button("Dashboard", on_click=lambda: ui.open("/dashboard"))
                ui.button("Form", on_click=lambda: ui.open("/form-demo"))

    return ui

8. Entry Point¶

8.1 app.py¶

# app.py
from __future__ import annotations

from core.app_factory import create_app

app = create_app()

if __name__ in {"__main__", "__mp_main__"}:
    # Default NiceGUI run:
    app.run(
        title="NiceGUI Class-Based Pages",
        reload=False,  # set True in dev if desired
        host="0.0.0.0",
        port=8080,
    )

Run it:

# Option 1: Direct Python
python app.py

# Option 2: Uvicorn (if using FastAPI integration)
uvicorn app:app --reload

9. Integration Patterns¶

Now that the skeleton is alive, here are some patterns for more serious integration.

9.1: FastAPI + NiceGUI in One Process¶

NiceGUI exposes a FastAPI app via ui.run_with / app.fastapi. A common pattern:

Use FastAPI for REST/webhook endpoints
Use NiceGUI for UI
Share service container between them

Skeleton:

# app.py
from nicegui import ui
from fastapi import FastAPI

from core.app_factory import create_app
from core.services import create_services

services = create_services()
ng_app = create_app()  # registers pages, etc.

fastapi_app = FastAPI()

# Example API endpoint
@fastapi_app.get("/api/ping")
async def ping():
    api = services["api"]
    data = await api.get_status()
    return {"status": "ok", "upstream": data}

ui.run_with(
    fastapi_app,
    title="NiceGUI + FastAPI",
    host="0.0.0.0",
    port=8080,
)

Now:

http://localhost:8080/ → NiceGUI UI
http://localhost:8080/api/ping → FastAPI JSON

Pages keep using the same service container.

9.2: Page-Level Dependencies¶

Sometimes you don't want to pass the entire service container; you want typed attributes.

You can wrap the base to provide typed accessors:

# pages/base.py (alt pattern)
from dataclasses import dataclass
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from core.services import ApiClient

@dataclass
class TypedServices:
    api: "ApiClient"

class BasePage(ABC):
    # ...
    def __init__(self, services: dict[str, Any]) -> None:
        self.services_raw = services
        self.s = TypedServices(api=services["api"])

Then in page:

# pages/dashboard.py
class DashboardPage(BasePage):
    # ...
    async def _refresh_status(self, label: ui.label) -> None:
        data = await self.s.api.get_status()
        # ...

9.3: Background Tasks & Schedulers¶

For periodic tasks (polling metrics, refreshing dashboards), you can:

Use asyncio.create_task from within a page
Or integrate external schedulers (APScheduler, Prefect, etc.) and push updates via NiceGUI

Example: simple background update:

# in a Page.build()
import asyncio

counter_label = ui.label("Counter: 0")
self.counter = 0  # attach to instance

async def updater():
    while True:
        await asyncio.sleep(1)
        self.counter += 1
        counter_label.text = f"Counter: {self.counter}"

asyncio.create_task(updater())

Just be aware you're now dancing with concurrency and lifetime—good for small toys, but for production you'll want a more disciplined approach.

10. Integration with Prefect¶

Let's build a real-world example: a NiceGUI dashboard that displays Prefect flow runs and allows triggering new runs.

10.1 core/services.py (Extended)¶

# core/services.py (extended)
from __future__ import annotations

from dataclasses import dataclass
from typing import Any
import httpx


@dataclass
class ApiClient:
    base_url: str

    async def get_status(self) -> dict[str, Any]:
        async with httpx.AsyncClient(base_url=self.base_url, timeout=5.0) as client:
            r = await client.get("/status")
            r.raise_for_status()
            return r.json()


@dataclass
class PrefectClient:
    api_url: str
    api_key: str | None = None

    async def get_flow_runs(
        self, limit: int = 10, flow_id: str | None = None
    ) -> list[dict[str, Any]]:
        """Fetch recent flow runs from Prefect API."""
        headers = {}
        if self.api_key:
            headers["Authorization"] = f"Bearer {self.api_key}"

        params = {"limit": limit}
        if flow_id:
            params["flow_id"] = flow_id

        async with httpx.AsyncClient(
            base_url=self.api_url, headers=headers, timeout=10.0
        ) as client:
            r = await client.get("/flow_runs", params=params)
            r.raise_for_status()
            return r.json()

    async def trigger_flow_run(self, deployment_id: str) -> dict[str, Any]:
        """Trigger a new flow run."""
        headers = {}
        if self.api_key:
            headers["Authorization"] = f"Bearer {self.api_key}"

        async with httpx.AsyncClient(
            base_url=self.api_url, headers=headers, timeout=10.0
        ) as client:
            r = await client.post(
                f"/deployments/{deployment_id}/create_flow_run",
                json={},
            )
            r.raise_for_status()
            return r.json()


def create_services() -> dict[str, Any]:
    """Construct and return the service container."""
    import os

    api_client = ApiClient(base_url="https://httpbin.org")
    prefect_client = PrefectClient(
        api_url=os.getenv("PREFECT_API_URL", "http://localhost:4200/api"),
        api_key=os.getenv("PREFECT_API_KEY"),
    )

    return {
        "api": api_client,
        "prefect": prefect_client,
    }

10.2 pages/prefect_dashboard.py¶

# pages/prefect_dashboard.py
from __future__ import annotations

from typing import Any
from nicegui import ui
from .base import BasePage


class PrefectDashboardPage(BasePage):
    route = "/prefect"
    title = "Prefect Dashboard"

    def __init__(self, services: dict[str, Any] | None = None) -> None:
        super().__init__(services)
        self.flow_runs: list[dict[str, Any]] = []
        self.runs_table: ui.table | None = None

    async def _refresh_runs(self) -> None:
        """Fetch and display Prefect flow runs."""
        prefect = self.services["prefect"]
        try:
            self.flow_runs = await prefect.get_flow_runs(limit=20)
            if self.runs_table:
                self.runs_table.rows = [
                    {
                        "id": run.get("id", "unknown"),
                        "name": run.get("name", "unnamed"),
                        "state": run.get("state", {}).get("type", "unknown"),
                        "start_time": run.get("start_time", "N/A"),
                    }
                    for run in self.flow_runs
                ]
            ui.notify(f"Loaded {len(self.flow_runs)} flow runs", type="positive")
        except Exception as e:  # noqa: BLE001
            ui.notify(f"Error loading runs: {e}", type="negative")

    async def _trigger_run(self, deployment_id: str) -> None:
        """Trigger a new flow run."""
        prefect = self.services["prefect"]
        try:
            result = await prefect.trigger_flow_run(deployment_id)
            ui.notify(f"Flow run created: {result.get('id')}", type="positive")
            await self._refresh_runs()
        except Exception as e:  # noqa: BLE001
            ui.notify(f"Error triggering run: {e}", type="negative")

    def build(self) -> None:
        ui.label("Prefect Flow Runs").classes("text-lg mb-2")

        with ui.row().classes("gap-2 mb-4"):
            ui.button("Refresh", on_click=self._refresh_runs)
            ui.button("Auto-refresh (10s)", on_click=self._start_auto_refresh)

        columns = [
            {"name": "id", "label": "Run ID", "field": "id"},
            {"name": "name", "label": "Name", "field": "name"},
            {"name": "state", "label": "State", "field": "state"},
            {"name": "start_time", "label": "Start Time", "field": "start_time"},
        ]

        self.runs_table = ui.table(
            columns=columns,
            rows=[],
            row_key="id",
        ).classes("w-full")

        # Initial load
        self._refresh_runs()

    def _start_auto_refresh(self) -> None:
        """Start auto-refreshing the table every 10 seconds."""
        import asyncio

        async def refresh_loop():
            while True:
                await asyncio.sleep(10)
                await self._refresh_runs()

        asyncio.create_task(refresh_loop())
        ui.notify("Auto-refresh started", type="info")

10.3 Update app_factory.py¶

# core/app_factory.py (add PrefectDashboardPage)
from pages.prefect_dashboard import PrefectDashboardPage

# In create_app():
page_classes: list[Type[BasePage]] = [
    HomePage,
    DashboardPage,
    FormDemoPage,
    PrefectDashboardPage,  # Add this
]

11. Integration with PostgreSQL¶

Let's add a page that queries PostgreSQL and displays data in real-time.

11.1 core/services.py (Add Postgres)¶

# core/services.py (add Postgres client)
import os
from typing import Any
import asyncpg


@dataclass
class PostgresClient:
    connection_string: str
    pool: asyncpg.Pool | None = None

    async def get_pool(self) -> asyncpg.Pool:
        """Get or create connection pool."""
        if self.pool is None:
            self.pool = await asyncpg.create_pool(self.connection_string)
        return self.pool

    async def query(self, sql: str, *args: Any) -> list[dict[str, Any]]:
        """Execute query and return rows as dicts."""
        pool = await self.get_pool()
        async with pool.acquire() as conn:
            rows = await conn.fetch(sql, *args)
            return [dict(row) for row in rows]

    async def execute(self, sql: str, *args: Any) -> str:
        """Execute statement and return result."""
        pool = await self.get_pool()
        async with pool.acquire() as conn:
            return await conn.execute(sql, *args)


def create_services() -> dict[str, Any]:
    """Construct and return the service container."""
    api_client = ApiClient(base_url="https://httpbin.org")
    prefect_client = PrefectClient(
        api_url=os.getenv("PREFECT_API_URL", "http://localhost:4200/api"),
        api_key=os.getenv("PREFECT_API_KEY"),
    )
    postgres_client = PostgresClient(
        connection_string=os.getenv(
            "DATABASE_URL", "postgresql://user:pass@localhost/dbname"
        )
    )

    return {
        "api": api_client,
        "prefect": prefect_client,
        "postgres": postgres_client,
    }

11.2 pages/postgres_dashboard.py¶

# pages/postgres_dashboard.py
from __future__ import annotations

from typing import Any
from nicegui import ui
from .base import BasePage


class PostgresDashboardPage(BasePage):
    route = "/postgres"
    title = "PostgreSQL Dashboard"

    def __init__(self, services: dict[str, Any] | None = None) -> None:
        super().__init__(services)
        self.query_result: list[dict[str, Any]] = []
        self.result_table: ui.table | None = None

    async def _execute_query(self, sql_input: ui.textarea) -> None:
        """Execute SQL query and display results."""
        postgres = self.services["postgres"]
        sql = sql_input.value.strip()

        if not sql:
            ui.notify("Please enter a SQL query", type="warning")
            return

        try:
            self.query_result = await postgres.query(sql)
            if self.result_table:
                columns = (
                    [{"name": k, "label": k, "field": k} for k in self.query_result[0].keys()]
                    if self.query_result
                    else []
                )
                self.result_table.columns = columns
                self.result_table.rows = self.query_result
            ui.notify(f"Query returned {len(self.query_result)} rows", type="positive")
        except Exception as e:  # noqa: BLE001
            ui.notify(f"Query error: {e}", type="negative")

    def build(self) -> None:
        ui.label("PostgreSQL Query Interface").classes("text-lg mb-2")

        with ui.card().classes("w-full"):
            sql_input = ui.textarea(
                label="SQL Query",
                placeholder="SELECT * FROM table LIMIT 10;",
            ).classes("w-full mb-2")
            sql_input.value = "SELECT version();"

            ui.button(
                "Execute",
                on_click=lambda: self._execute_query(sql_input),
            ).classes("mb-4")

            columns = [{"name": "result", "label": "Result", "field": "result"}]
            self.result_table = ui.table(
                columns=columns,
                rows=[],
            ).classes("w-full")

11.3 Update requirements.txt¶

nicegui>=2.0.0
httpx>=0.27.0
asyncpg>=0.29.0

12. Integration with Redis¶

For caching and real-time updates, let's add Redis support.

12.1 core/services.py (Add Redis)¶

# core/services.py (add Redis)
import redis.asyncio as redis


@dataclass
class RedisClient:
    url: str
    client: redis.Redis | None = None

    async def get_client(self) -> redis.Redis:
        """Get or create Redis client."""
        if self.client is None:
            self.client = redis.from_url(self.url, decode_responses=True)
        return self.client

    async def get(self, key: str) -> str | None:
        """Get value from Redis."""
        client = await self.get_client()
        return await client.get(key)

    async def set(self, key: str, value: str, ex: int | None = None) -> bool:
        """Set value in Redis."""
        client = await self.get_client()
        return await client.set(key, value, ex=ex)

    async def keys(self, pattern: str = "*") -> list[str]:
        """Get keys matching pattern."""
        client = await self.get_client()
        return await client.keys(pattern)


def create_services() -> dict[str, Any]:
    """Construct and return the service container."""
    import os

    # ... existing services ...
    redis_client = RedisClient(url=os.getenv("REDIS_URL", "redis://localhost:6379"))

    return {
        "api": api_client,
        "prefect": prefect_client,
        "postgres": postgres_client,
        "redis": redis_client,
    }

12.2 pages/redis_dashboard.py¶

# pages/redis_dashboard.py
from __future__ import annotations

from typing import Any
from nicegui import ui
from .base import BasePage


class RedisDashboardPage(BasePage):
    route = "/redis"
    title = "Redis Dashboard"

    async def _get_value(self, key_input: ui.input, value_label: ui.label) -> None:
        """Get value from Redis."""
        redis_client = self.services["redis"]
        key = key_input.value

        if not key:
            ui.notify("Please enter a key", type="warning")
            return

        try:
            value = await redis_client.get(key)
            value_label.text = f"Value: {value if value else '(nil)'}"
            ui.notify("Value retrieved", type="positive")
        except Exception as e:  # noqa: BLE001
            ui.notify(f"Error: {e}", type="negative")

    async def _set_value(
        self, key_input: ui.input, value_input: ui.input
    ) -> None:
        """Set value in Redis."""
        redis_client = self.services["redis"]
        key = key_input.value
        value = value_input.value

        if not key:
            ui.notify("Please enter a key", type="warning")
            return

        try:
            await redis_client.set(key, value)
            ui.notify(f"Set {key} = {value}", type="positive")
        except Exception as e:  # noqa: BLE001
            ui.notify(f"Error: {e}", type="negative")

    async def _list_keys(self, keys_list: ui.column) -> None:
        """List all keys in Redis."""
        redis_client = self.services["redis"]
        try:
            keys = await redis_client.keys()
            keys_list.clear()
            with keys_list:
                for key in keys[:100]:  # Limit to 100 keys
                    ui.label(key).classes("text-sm")
            ui.notify(f"Found {len(keys)} keys", type="positive")
        except Exception as e:  # noqa: BLE001
            ui.notify(f"Error: {e}", type="negative")

    def build(self) -> None:
        ui.label("Redis Operations").classes("text-lg mb-2")

        with ui.card().classes("w-full max-w-2xl"):
            key_input = ui.input("Key").classes("mb-2")
            value_input = ui.input("Value").classes("mb-2")
            value_label = ui.label("Value: (not loaded)").classes("mb-2")

            with ui.row().classes("gap-2 mb-4"):
                ui.button(
                    "Get",
                    on_click=lambda: self._get_value(key_input, value_label),
                )
                ui.button(
                    "Set",
                    on_click=lambda: self._set_value(key_input, value_input),
                )

            ui.separator()

            ui.label("Keys").classes("text-md font-semibold mb-2")
            keys_list = ui.column().classes("w-full max-h-64 overflow-auto")
            ui.button(
                "List Keys",
                on_click=lambda: self._list_keys(keys_list),
            )

12.3 Update requirements.txt¶

nicegui>=2.0.0
httpx>=0.27.0
asyncpg>=0.29.0
redis>=5.0.0

13. Data Flow Diagram¶

flowchart LR
    subgraph Services["Services / Backend"]
        API["ApiClient (httpx)"]
        Prefect["PrefectClient"]
        Postgres["PostgresClient (asyncpg)"]
        Redis["RedisClient"]
    end

    subgraph UI["NiceGUI App"]
        Home["HomePage"]
        Dash["DashboardPage"]
        Form["FormDemoPage"]
        PrefDash["PrefectDashboardPage"]
        PgDash["PostgresDashboardPage"]
        RedisDash["RedisDashboardPage"]
    end

    AppFactory["create_app()"] --> UI
    AppFactory --> Services

    Home --> Services
    Dash --> API
    PrefDash --> Prefect
    PgDash --> Postgres
    RedisDash --> Redis

    Prefect -->|"HTTP / JSON"| PrefAPI["Prefect API"]
    Postgres -->|"SQL"| PgDB["PostgreSQL"]
    Redis -->|"Commands"| RedisServer["Redis Server"]

The core idea:

App factory builds services + registers pages
Pages take services as dependencies
Everything stays in tidy modules instead of a global ball of state

14. Complete Example: All Together¶

Here's a complete app.py that ties everything together:

# app.py
from __future__ import annotations

import os
from nicegui import ui
from fastapi import FastAPI

from core.app_factory import create_app
from core.services import create_services

# Create services
services = create_services()

# Create NiceGUI app with all pages
ng_app = create_app()

# Create FastAPI app for API endpoints
fastapi_app = FastAPI()


@fastapi_app.get("/api/health")
async def health():
    """Health check endpoint."""
    return {"status": "ok", "services": list(services.keys())}


@fastapi_app.get("/api/services")
async def list_services():
    """List available services."""
    return {"services": list(services.keys())}


# Run NiceGUI with FastAPI
if __name__ in {"__main__", "__mp_main__"}:
    ui.run_with(
        fastapi_app,
        title="NiceGUI Class-Based App",
        reload=os.getenv("NICEGUI_RELOAD", "false").lower() == "true",
        host="0.0.0.0",
        port=int(os.getenv("PORT", "8080")),
    )

15. Checklist for Extending This¶

If you're going to hand this to an agent and tell it to mutate reality:

✅ Add more page classes under pages/ for specific domains (metrics, admin, logs)
✅ Implement a shared BaseLayout with sidebar navigation and theming
✅ Integrate authentication (e.g., JWT, OIDC) in an AuthPage + guards around routes
✅ Add a SettingsPage that persists config to a DB or file
✅ Add tests for service layer and simple smoke tests for NiceGUI pages (headless)
✅ Add error boundaries and global error handling
✅ Implement real-time updates via WebSockets or Server-Sent Events
✅ Add logging and monitoring integration
✅ Create reusable UI components (cards, tables, forms) in a components/ directory

16. Production Considerations¶

16.1 Error Handling¶

Add global error handling:

# core/app_factory.py
@ui.exception_handler
def handle_exception(e: Exception):
    ui.notify(f"Error: {e}", type="negative")
    # Log to your logging service

16.2 Authentication Guards¶

Add route protection:

# pages/base.py (extended)
class BasePage(ABC):
    # ...
    requires_auth: bool = False

    def render(self) -> None:
        if self.requires_auth and not self._is_authenticated():
            ui.open("/login")
            return
        # ... rest of render

16.3 Environment Configuration¶

Use environment variables for all configuration:

# core/config.py
import os
from dataclasses import dataclass

@dataclass
class Config:
    prefect_api_url: str = os.getenv("PREFECT_API_URL", "http://localhost:4200/api")
    prefect_api_key: str | None = os.getenv("PREFECT_API_KEY")
    database_url: str = os.getenv("DATABASE_URL", "postgresql://localhost/db")
    redis_url: str = os.getenv("REDIS_URL", "redis://localhost:6379")