Skip to content

API Reference

panel_flowdash

panel-flowdash: Dataflow and draggable grid based dashboard editor for Panel.

__all__ = ['AuthConfig', 'ComponentSpec', 'ConfigField', 'DashboardEdge', 'DashboardItem', 'DashboardModel', 'DashboardStore', 'DataflowGraph', 'Identity', 'InputPort', 'OutputPort', 'PanelAppMetadata', 'Permission', 'RegistryEntry', '__version__', 'build_component_spec', 'build_component_specs', 'build_node_state_class', 'build_session_state_class', 'can_administer', 'check_requirements', 'is_authorized', 'panel_app', 'register', 'resolve_identity'] module-attribute

__version__ = importlib.metadata.version(__name__) module-attribute

panel_app = register module-attribute

AuthConfig dataclass

Project-level authorization configuration.

Loaded from the project's __init__.py by the serve command. All fields are optional; the defaults reproduce the pre-auth behavior (allow by default, groups read from the standard claim keys, no admin groups).

admin_groups = field(default_factory=frozenset) class-attribute instance-attribute

default_allow = True class-attribute instance-attribute

group_claims = DEFAULT_GROUP_CLAIMS class-attribute instance-attribute

resolve_groups = None class-attribute instance-attribute

user_groups = field(default_factory=dict) class-attribute instance-attribute

from_module(module) classmethod

Build an :class:AuthConfig from names on a project __init__.

Reads group_claims, user_groups, resolve_groups, admin_groups and default_allow if present, falling back to the defaults otherwise. Missing module or names yield a default config.

ComponentSpec dataclass

Full specification of a component's ports and metadata.

component_id instance-attribute

config = field(default_factory=list) class-attribute instance-attribute

config_editor = None class-attribute instance-attribute

config_state_class = None class-attribute instance-attribute

default_size instance-attribute

description instance-attribute

icon instance-attribute

inputs instance-attribute

outputs instance-attribute

tags instance-attribute

title instance-attribute

ConfigField dataclass

Describes a single design-time configuration option on a component.

default = None class-attribute instance-attribute

label = None class-attribute instance-attribute

name instance-attribute

type = None class-attribute instance-attribute

DashboardEdge dataclass

A connection between two component ports.

source instance-attribute

source_port instance-attribute

target instance-attribute

target_port instance-attribute

from_dict(data) classmethod

to_dict()

DashboardItem dataclass

A component instance on the dashboard.

x, y store the ReactFlow node canvas position. Grid layout (widths, heights, visibility) lives in DashboardModel.tile_layout.

component_id instance-attribute

config = field(default_factory=dict) class-attribute instance-attribute

instance_id instance-attribute

x = 0 class-attribute instance-attribute

y = 0 class-attribute instance-attribute

from_dict(data) classmethod

to_dict()

DashboardModel dataclass

A persisted dashboard: nodes + edges + tile layout.

breakpoints = field(default_factory=list) class-attribute instance-attribute

dashboard_id instance-attribute

edges = field(default_factory=list) class-attribute instance-attribute

items = field(default_factory=list) class-attribute instance-attribute

owner property

The immutable owner principal (the creating user).

permission = field(default_factory=Permission) class-attribute instance-attribute

responsive_layouts = field(default_factory=dict) class-attribute instance-attribute

tile_layout = field(default_factory=list) class-attribute instance-attribute

title instance-attribute

user_id instance-attribute

version = 3 class-attribute instance-attribute

from_dict(data) classmethod

to_dict()

DashboardStore

SQLite-backed store for dashboard models.

can_administer(identity, dashboard_id, admin_groups=frozenset())

Whether identity may administer (edit/delete/share) the dashboard.

create_dashboard(user_id, title)

delete_dashboard(user_id, dashboard_id)

find_by_id_or_title(ref)

Resolve a dashboard by its id first, then by title.

Titles are only unique per user, so a title match returns the most recently updated dashboard. Used to resolve the operator-configured home dashboard, which may be given as either an id or a title.

get_owner(dashboard_id)

Return the owner (user_id) of a dashboard, or None if missing.

list_accessible(identity, *, default_allow=True)

List dashboards the identity owns or has been granted access to.

Owned dashboards sort first (both groups by recency), so a user's own dashboards stay visually grouped ahead of ones shared with them.

list_dashboards(user_id)

load_dashboard(user_id, dashboard_id)

load_for_access(identity, dashboard_id, *, default_allow=True)

Load a dashboard if identity is authorized, else None.

Returns None both when the dashboard does not exist and when access is denied, so callers render a single "not found / denied" view.

rename_dashboard(user_id, dashboard_id, new_title)

save_dashboard(dashboard)

set_permission(dashboard_id, permission)

Persist a new permission set on a dashboard. Returns success.

title_exists(user_id, title, exclude_id=None)

Check if a dashboard with the given title already exists for this user.

DataflowGraph

Manages node state instances and wires edges with runtime validation.

edges property

All current edges.

node_ids property

All current node instance IDs.

add_edge(source_id, source_port, target_id, target_port)

Wire an edge between two ports.

Returns True on success, or an error message string on failure.

add_node(instance_id, component_id)

Create a new node state instance.

get_config_state(instance_id)

Get the config state instance for a node, if it has config.

get_state(instance_id)

Get the state instance for a node.

remove_edge(source_id, source_port, target_id, target_port)

Remove an edge, unsubscribe the watcher, and reset target to default.

remove_node(instance_id)

Remove a node and any edges connected to it.

Identity dataclass

The resolved principal for a session.

groups = field(default_factory=frozenset) class-attribute instance-attribute

oauth_user = None class-attribute instance-attribute

system_user = None class-attribute instance-attribute

user instance-attribute

user_info = field(default_factory=dict) class-attribute instance-attribute

user_names property

All names this identity may be referenced by in a rule.

in_groups(groups)

Whether the identity belongs to any of groups.

is_user(users)

Whether the identity matches any of users (OAuth or system name).

InputPort dataclass

Describes a single input port on a component node.

blocking = True class-attribute instance-attribute

default = None class-attribute instance-attribute

label = None class-attribute instance-attribute

name instance-attribute

required = True class-attribute instance-attribute

type = None class-attribute instance-attribute

OutputPort dataclass

Describes a single output port on a component node.

label = None class-attribute instance-attribute

name instance-attribute

type = None class-attribute instance-attribute

PanelAppMetadata dataclass

Metadata attached to a component or page by the @register decorator.

allow_groups = field(default_factory=list) class-attribute instance-attribute

allow_users = field(default_factory=list) class-attribute instance-attribute

authorize = None class-attribute instance-attribute

component = False class-attribute instance-attribute

config = field(default_factory=list) class-attribute instance-attribute

config_editor = None class-attribute instance-attribute

config_schema = None class-attribute instance-attribute

default_size = None class-attribute instance-attribute

deny_groups = field(default_factory=list) class-attribute instance-attribute

deny_users = field(default_factory=list) class-attribute instance-attribute

description = None class-attribute instance-attribute

icon = None class-attribute instance-attribute

max_size = None class-attribute instance-attribute

min_size = None class-attribute instance-attribute

page = True class-attribute instance-attribute

permission property

Build a :class:~panel_flowdash.auth.Permission from the declared rules.

provides = field(default_factory=list) class-attribute instance-attribute

requires = field(default_factory=list) class-attribute instance-attribute

sidebar = False class-attribute instance-attribute

singleton = False class-attribute instance-attribute

tags = field(default_factory=list) class-attribute instance-attribute

title = None class-attribute instance-attribute

from_app(app) classmethod

Extract metadata from an app object.

Permission dataclass

An allow/deny rule set evaluated against an :class:Identity.

All four fields match either the resolved user (OAuth login or system user) or one of the identity's groups. An empty Permission declares no constraints and defers entirely to the caller's default policy.

allow_groups = field(default_factory=frozenset) class-attribute instance-attribute

allow_users = field(default_factory=frozenset) class-attribute instance-attribute

deny_groups = field(default_factory=frozenset) class-attribute instance-attribute

deny_users = field(default_factory=frozenset) class-attribute instance-attribute

is_empty property

Whether the permission declares no allow or deny rules.

from_dict(data) classmethod

Deserialize from a (possibly None or partial) mapping.

from_spec(*, allow_users=None, allow_groups=None, deny_users=None, deny_groups=None) classmethod

Build a :class:Permission from loosely-typed iterables.

to_dict()

Serialize to sorted lists for JSON persistence.

RegistryEntry dataclass

A registered component/page with its metadata.

app = None class-attribute instance-attribute

app_id instance-attribute

metadata instance-attribute

module_name instance-attribute

module_path = None class-attribute instance-attribute

name instance-attribute

page_path instance-attribute

section instance-attribute

title property

Human-readable title.

load()

Import the module and return the app object.

Caches the result on self.app. Raises on import failure.

build_component_spec(entry)

Build a ComponentSpec from a registry entry.

build_component_specs(registry)

Build specs for all component-enabled entries in a registry.

build_node_state_class(spec)

Create a Parameterized subclass with one param per port (inputs + outputs).

build_session_state_class(registry)

Build a Parameterized subclass with one param per declared state key.

Scans the registry for all provides and requires keys and creates a dynamic class whose parameters represent shared session state.

can_administer(identity, owner, admin_groups)

Whether identity may administer a resource owned by owner.

When no admin_groups are configured administration is unrestricted: the running user (however resolved) may administer any resource. This keeps the default, auth-less deployment fully editable. Once admin_groups are set, only the owner and members of those groups may administer a resource.

check_requirements(state, requires)

Check which required keys are unsatisfied on the given state instance.

Returns a list of dicts describing each unsatisfied requirement. An empty list means all requirements are met.

is_authorized(permission, identity, *, default_allow=True, owner=None)

Evaluate permission against identity.

Order of precedence:

  1. A matching deny_users/deny_groups rule denies access (deny always wins, even for the owner).
  2. The owner, if given and matching, is allowed.
  3. Any allow_* rule present: allowed iff the identity matches at least one of them.
  4. No allow/deny rules at all: fall back to default_allow.

register(*, page=True, component=False, sidebar=False, title=None, icon=None, description=None, tags=None, default_size=None, min_size=None, max_size=None, singleton=False, provides=None, requires=None, config_schema=None, config=None, config_editor=None, allow_users=None, allow_groups=None, deny_users=None, deny_groups=None, authorize=None)

Metadata-only decorator for app exports.

Annotates an app object/callable without altering runtime behavior.

The config_schema, config and config_editor arguments declare design-time configuration options that appear in the node editor. Use config_schema (a param.Parameterized subclass, a Pydantic model, or a JSON Schema dict) to define config explicitly, or config to name which of a Viewer's own params are configuration rather than input ports. Pass config_editor to supply a custom editor callable instead of the auto-generated form.

The allow_users, allow_groups, deny_users and deny_groups arguments declare page-level authorization rules. Users are matched against either the OAuth login or the system user; groups against the identity's resolved group membership. Deny rules always win; when only allow rules are present the identity must match at least one; with no rules the project's default policy applies. Pass authorize for a custom callable taking the resolved Identity and returning a bool (resolved on import).

resolve_identity(auth_config=None)

Resolve the current session's :class:Identity.

Prefers the OAuth login (pn.state.user) when a real provider populated it; otherwise falls back to the system user, then to "anonymous". Groups are the union of claim-derived groups, the static user_groups mapping and any dynamic resolve_groups callback.

__main__

Allow running as python -m panel_flowdash serve.

app

Application builder: scans a project directory and constructs the Panel app.

COMPONENTS_ROUTE = '/components' module-attribute

DASH_ROUTE_PREFIX = '/dash/' module-attribute

logger = logging.getLogger('panel_flowdash') module-attribute

FlowDashApp

Bases: Viewer

FlowDash application: scans a project directory and serves its pages and components.

auth_config = param.ClassSelector(class_=AuthConfig, doc='\n Project-level authorization configuration controlling group discovery,\n the admin groups and the default access policy.') class-attribute instance-attribute
breakpoints = param.List(default=[768, 1200], doc='Responsive breakpoints for the tile grid.') class-attribute instance-attribute
configure_layout = param.Callable(default=None, doc='\n Optional callback invoked on every navigation with\n (app, content, route). Use it to set `app.sidebar` and\n `app.contextbar` for the page currently being served.') class-attribute instance-attribute
contextbar = Children(default=[], doc='Items prepended to the contextbar.') class-attribute instance-attribute
contextbar_open = param.Boolean(default=False, doc='Whether the contextbar is open.') class-attribute instance-attribute
home_dashboard = param.String(default=None, doc="\n Dashboard shown on the homepage ('/'). Accepts a dashboard id or\n title. When unset, the homepage shows the dashboard grid launcher.") class-attribute instance-attribute
nav_variant = param.Selector(default='drawer', objects=['drawer', 'menubar'], doc="\n Where the navigation menu is rendered. 'drawer' docks a MenuList in a\n right-hand drawer; 'menubar' places a MenuBar in the page header with\n quick-action icons alongside it.") class-attribute instance-attribute
page_options = param.Dict(default={}, doc="\n Extra keyword arguments passed through to the underlying\n `panel_material_ui.Page`, overriding the app's own defaults.") class-attribute instance-attribute
project_dir = param.Path(doc='Path to the project directory.') class-attribute instance-attribute
sidebar = Children(default=[], doc='Items prepended to the sidebar.') class-attribute instance-attribute
store = param.ClassSelector(class_=DashboardStore, doc='DashboardStore instance for persistence.') class-attribute instance-attribute
title = param.String(default='FlowDash', doc='Application title shown in the browser tab.') class-attribute instance-attribute
build_routes(project_dir, registry=None, **params) classmethod

Generate route mapping for pn.serve.

auth

Per-page and per-dashboard authorization.

The module defines a small, self-contained authorization model shared by both code-authored pages (permissions declared on the @register decorator) and runtime-created dashboards (permissions persisted in the store):

  • :class:Identity — the resolved principal for a session (OAuth user, system user, resolved user and the set of groups it belongs to).
  • :class:Permission — an allow/deny rule set keyed on users and groups.
  • :class:AuthConfig — project-level configuration controlling how groups are discovered and which groups may administer any dashboard.
  • :func:resolve_identity — builds an :class:Identity from the live Panel session plus the AuthConfig.
  • :func:is_authorized — evaluates a :class:Permission against an :class:Identity.

The same :func:is_authorized evaluator serves pages, dashboards and (in the future) components, so the semantics only ever live in one place.

ANONYMOUS_USER = 'anonymous' module-attribute

DEFAULT_GROUP_CLAIMS = ('groups', 'roles') module-attribute

logger = logging.getLogger('panel_flowdash') module-attribute

AuthConfig dataclass

Project-level authorization configuration.

Loaded from the project's __init__.py by the serve command. All fields are optional; the defaults reproduce the pre-auth behavior (allow by default, groups read from the standard claim keys, no admin groups).

admin_groups = field(default_factory=frozenset) class-attribute instance-attribute
default_allow = True class-attribute instance-attribute
group_claims = DEFAULT_GROUP_CLAIMS class-attribute instance-attribute
resolve_groups = None class-attribute instance-attribute
user_groups = field(default_factory=dict) class-attribute instance-attribute
from_module(module) classmethod

Build an :class:AuthConfig from names on a project __init__.

Reads group_claims, user_groups, resolve_groups, admin_groups and default_allow if present, falling back to the defaults otherwise. Missing module or names yield a default config.

Identity dataclass

The resolved principal for a session.

groups = field(default_factory=frozenset) class-attribute instance-attribute
oauth_user = None class-attribute instance-attribute
system_user = None class-attribute instance-attribute
user instance-attribute
user_info = field(default_factory=dict) class-attribute instance-attribute
user_names property

All names this identity may be referenced by in a rule.

in_groups(groups)

Whether the identity belongs to any of groups.

is_user(users)

Whether the identity matches any of users (OAuth or system name).

Permission dataclass

An allow/deny rule set evaluated against an :class:Identity.

All four fields match either the resolved user (OAuth login or system user) or one of the identity's groups. An empty Permission declares no constraints and defers entirely to the caller's default policy.

allow_groups = field(default_factory=frozenset) class-attribute instance-attribute
allow_users = field(default_factory=frozenset) class-attribute instance-attribute
deny_groups = field(default_factory=frozenset) class-attribute instance-attribute
deny_users = field(default_factory=frozenset) class-attribute instance-attribute
is_empty property

Whether the permission declares no allow or deny rules.

from_dict(data) classmethod

Deserialize from a (possibly None or partial) mapping.

from_spec(*, allow_users=None, allow_groups=None, deny_users=None, deny_groups=None) classmethod

Build a :class:Permission from loosely-typed iterables.

to_dict()

Serialize to sorted lists for JSON persistence.

can_administer(identity, owner, admin_groups)

Whether identity may administer a resource owned by owner.

When no admin_groups are configured administration is unrestricted: the running user (however resolved) may administer any resource. This keeps the default, auth-less deployment fully editable. Once admin_groups are set, only the owner and members of those groups may administer a resource.

is_authorized(permission, identity, *, default_allow=True, owner=None)

Evaluate permission against identity.

Order of precedence:

  1. A matching deny_users/deny_groups rule denies access (deny always wins, even for the owner).
  2. The owner, if given and matching, is allowed.
  3. Any allow_* rule present: allowed iff the identity matches at least one of them.
  4. No allow/deny rules at all: fall back to default_allow.

make_authorize_callback(registry, auth_config, *, url_prefix='')

Build a Panel config.authorize_callback gating page routes.

The returned callback authorizes direct HTTP access to code-authored page URLs against the resolved identity. Non-page routes (/components, /dash/... and unknown paths) always pass here and are gated in-app, where per-dashboard permissions and richer denied views live.

url_prefix strips a server route prefix from the request path before matching against registry page paths.

path_permission_lookup(registry)

Map each page's URL path to its declared :class:Permission.

Used by the HTTP-boundary authorize callback to gate direct URL access to code-authored pages. Only page entries are included; component and SPA-only routes are gated in-app.

resolve_identity(auth_config=None)

Resolve the current session's :class:Identity.

Prefers the OAuth login (pn.state.user) when a real provider populated it; otherwise falls back to the system user, then to "anonymous". Groups are the union of claim-derived groups, the static user_groups mapping and any dynamic resolve_groups callback.

command

Command-line interface for panel-flowdash.

main(args=None)

Entry point for the panel-flowdash CLI.

serve

The flowdash serve subcommand.

log = logging.getLogger(__name__) module-attribute
Serve

Bases: Serve

Serve a flowdash dashboard application from a project directory.

args = (('directory', Argument(metavar='DIRECTORY', help='Path to the project directory containing page/component modules.')), ('--db-path', Argument(action='store', type=str, default=None, help='Path to the SQLite database file. Defaults to <directory>/dashboards.db.')), ('--title', Argument(action='store', type=str, default='FlowDash', help='Application title shown in the browser tab.')), ('--home-dashboard', Argument(action='store', type=str, default=None, help='Dashboard (id or title) to show on the homepage. When unset, the homepage shows the dashboard grid.')), ('--nav-variant', Argument(action='store', type=str, default='drawer', choices=('drawer', 'menubar'), help="Where to render the navigation menu: 'drawer' (docked right-hand drawer) or 'menubar' (in the page header).")), *((name, arg) for name, arg in (_PanelServe.args) if name not in _EXCLUDED_ARGS)) class-attribute instance-attribute
help = 'Launch the FlowDash dashboard server from a project directory.' class-attribute instance-attribute
name = 'serve' class-attribute instance-attribute
customize_applications(args, applications)
customize_kwargs(args, server_kwargs)
invoke(args)

component_spec

Component specification with typed ports for the dataflow editor.

ComponentSpec dataclass

Full specification of a component's ports and metadata.

component_id instance-attribute
config = field(default_factory=list) class-attribute instance-attribute
config_editor = None class-attribute instance-attribute
config_state_class = None class-attribute instance-attribute
default_size instance-attribute
description instance-attribute
icon instance-attribute
inputs instance-attribute
outputs instance-attribute
tags instance-attribute
title instance-attribute

ConfigField dataclass

Describes a single design-time configuration option on a component.

default = None class-attribute instance-attribute
label = None class-attribute instance-attribute
name instance-attribute
type = None class-attribute instance-attribute

InputPort dataclass

Describes a single input port on a component node.

blocking = True class-attribute instance-attribute
default = None class-attribute instance-attribute
label = None class-attribute instance-attribute
name instance-attribute
required = True class-attribute instance-attribute
type = None class-attribute instance-attribute

OutputPort dataclass

Describes a single output port on a component node.

label = None class-attribute instance-attribute
name instance-attribute
type = None class-attribute instance-attribute

build_component_spec(entry)

Build a ComponentSpec from a registry entry.

build_component_specs(registry)

Build specs for all component-enabled entries in a registry.

dashboard_store

SQLite-backed persistence for dashboard graphs.

DashboardEdge dataclass

A connection between two component ports.

source instance-attribute
source_port instance-attribute
target instance-attribute
target_port instance-attribute
from_dict(data) classmethod
to_dict()

DashboardItem dataclass

A component instance on the dashboard.

x, y store the ReactFlow node canvas position. Grid layout (widths, heights, visibility) lives in DashboardModel.tile_layout.

component_id instance-attribute
config = field(default_factory=dict) class-attribute instance-attribute
instance_id instance-attribute
x = 0 class-attribute instance-attribute
y = 0 class-attribute instance-attribute
from_dict(data) classmethod
to_dict()

DashboardModel dataclass

A persisted dashboard: nodes + edges + tile layout.

breakpoints = field(default_factory=list) class-attribute instance-attribute
dashboard_id instance-attribute
edges = field(default_factory=list) class-attribute instance-attribute
items = field(default_factory=list) class-attribute instance-attribute
owner property

The immutable owner principal (the creating user).

permission = field(default_factory=Permission) class-attribute instance-attribute
responsive_layouts = field(default_factory=dict) class-attribute instance-attribute
tile_layout = field(default_factory=list) class-attribute instance-attribute
title instance-attribute
user_id instance-attribute
version = 3 class-attribute instance-attribute
from_dict(data) classmethod
to_dict()

DashboardStore

SQLite-backed store for dashboard models.

can_administer(identity, dashboard_id, admin_groups=frozenset())

Whether identity may administer (edit/delete/share) the dashboard.

create_dashboard(user_id, title)
delete_dashboard(user_id, dashboard_id)
find_by_id_or_title(ref)

Resolve a dashboard by its id first, then by title.

Titles are only unique per user, so a title match returns the most recently updated dashboard. Used to resolve the operator-configured home dashboard, which may be given as either an id or a title.

get_owner(dashboard_id)

Return the owner (user_id) of a dashboard, or None if missing.

list_accessible(identity, *, default_allow=True)

List dashboards the identity owns or has been granted access to.

Owned dashboards sort first (both groups by recency), so a user's own dashboards stay visually grouped ahead of ones shared with them.

list_dashboards(user_id)
load_dashboard(user_id, dashboard_id)
load_for_access(identity, dashboard_id, *, default_allow=True)

Load a dashboard if identity is authorized, else None.

Returns None both when the dashboard does not exist and when access is denied, so callers render a single "not found / denied" view.

rename_dashboard(user_id, dashboard_id, new_title)
save_dashboard(dashboard)
set_permission(dashboard_id, permission)

Persist a new permission set on a dashboard. Returns success.

title_exists(user_id, title, exclude_id=None)

Check if a dashboard with the given title already exists for this user.

dataflow_engine

Dataflow wiring engine with runtime validation.

Each node in the graph gets a NodeState (a dynamic Parameterized subclass) whose parameters correspond to the node's declared input and output ports. Edges are wired via param.watch: when a source port changes, the value is assigned to the target port inside a try/except so that runtime type errors (e.g. param validation failures) are caught and reported via an error callback.

DataflowGraph

Manages node state instances and wires edges with runtime validation.

edges property

All current edges.

node_ids property

All current node instance IDs.

add_edge(source_id, source_port, target_id, target_port)

Wire an edge between two ports.

Returns True on success, or an error message string on failure.

add_node(instance_id, component_id)

Create a new node state instance.

get_config_state(instance_id)

Get the config state instance for a node, if it has config.

get_state(instance_id)

Get the state instance for a node.

remove_edge(source_id, source_port, target_id, target_port)

Remove an edge, unsubscribe the watcher, and reset target to default.

remove_node(instance_id)

Remove a node and any edges connected to it.

build_node_state_class(spec)

Create a Parameterized subclass with one param per port (inputs + outputs).

registry

Component registry: the register decorator and metadata model.

panel_app = register module-attribute

PanelAppMetadata dataclass

Metadata attached to a component or page by the @register decorator.

allow_groups = field(default_factory=list) class-attribute instance-attribute
allow_users = field(default_factory=list) class-attribute instance-attribute
authorize = None class-attribute instance-attribute
component = False class-attribute instance-attribute
config = field(default_factory=list) class-attribute instance-attribute
config_editor = None class-attribute instance-attribute
config_schema = None class-attribute instance-attribute
default_size = None class-attribute instance-attribute
deny_groups = field(default_factory=list) class-attribute instance-attribute
deny_users = field(default_factory=list) class-attribute instance-attribute
description = None class-attribute instance-attribute
icon = None class-attribute instance-attribute
max_size = None class-attribute instance-attribute
min_size = None class-attribute instance-attribute
page = True class-attribute instance-attribute
permission property

Build a :class:~panel_flowdash.auth.Permission from the declared rules.

provides = field(default_factory=list) class-attribute instance-attribute
requires = field(default_factory=list) class-attribute instance-attribute
sidebar = False class-attribute instance-attribute
singleton = False class-attribute instance-attribute
tags = field(default_factory=list) class-attribute instance-attribute
title = None class-attribute instance-attribute
from_app(app) classmethod

Extract metadata from an app object.

RegistryEntry dataclass

A registered component/page with its metadata.

app = None class-attribute instance-attribute
app_id instance-attribute
metadata instance-attribute
module_name instance-attribute
module_path = None class-attribute instance-attribute
name instance-attribute
page_path instance-attribute
section instance-attribute
title property

Human-readable title.

load()

Import the module and return the app object.

Caches the result on self.app. Raises on import failure.

build_registry(project_dir)

Scan project_dir for page/component modules without importing them.

Reads each .py file with the AST to extract @register metadata. Modules are not imported at this stage; each RegistryEntry.app is None until RegistryEntry.load() is called.

register(*, page=True, component=False, sidebar=False, title=None, icon=None, description=None, tags=None, default_size=None, min_size=None, max_size=None, singleton=False, provides=None, requires=None, config_schema=None, config=None, config_editor=None, allow_users=None, allow_groups=None, deny_users=None, deny_groups=None, authorize=None)

Metadata-only decorator for app exports.

Annotates an app object/callable without altering runtime behavior.

The config_schema, config and config_editor arguments declare design-time configuration options that appear in the node editor. Use config_schema (a param.Parameterized subclass, a Pydantic model, or a JSON Schema dict) to define config explicitly, or config to name which of a Viewer's own params are configuration rather than input ports. Pass config_editor to supply a custom editor callable instead of the auto-generated form.

The allow_users, allow_groups, deny_users and deny_groups arguments declare page-level authorization rules. Users are matched against either the OAuth login or the system user; groups against the identity's resolved group membership. Deny rules always win; when only allow rules are present the identity must match at least one; with no rules the project's default policy applies. Pass authorize for a custom callable taking the resolved Identity and returning a bool (resolved on import).

session_state

Per-session shared state built from registry provides/requires declarations.

build_session_state_class(registry)

Build a Parameterized subclass with one param per declared state key.

Scans the registry for all provides and requires keys and creates a dynamic class whose parameters represent shared session state.

check_requirements(state, requires)

Check which required keys are unsatisfied on the given state instance.

Returns a list of dicts describing each unsatisfied requirement. An empty list means all requirements are met.

Modules

Registry

panel_flowdash.registry

Component registry: the register decorator and metadata model.

_APP_METADATA_BY_ID = {} module-attribute

_LITERAL_KEYS = {'page', 'component', 'sidebar', 'title', 'icon', 'description', 'singleton', 'provides', 'requires', 'config', 'tags', 'default_size', 'min_size', 'max_size', 'allow_users', 'allow_groups', 'deny_users', 'deny_groups'} module-attribute

_REGISTER_NAMES = {'register', 'panel_app'} module-attribute

panel_app = register module-attribute

PanelAppMetadata dataclass

Metadata attached to a component or page by the @register decorator.

Source code in src/panel_flowdash/registry.py
@dataclass(frozen=True)
class PanelAppMetadata:
    """Metadata attached to a component or page by the @register decorator."""

    page: bool = True
    component: bool = False
    sidebar: bool = False
    title: str | None = None
    icon: str | None = None
    description: str | None = None
    tags: list[str] = field(default_factory=list)
    default_size: dict[str, Any] | None = None
    min_size: dict[str, Any] | None = None
    max_size: dict[str, Any] | None = None
    singleton: bool = False
    provides: list[str] = field(default_factory=list)
    requires: list[Any] = field(default_factory=list)
    config_schema: Any = None
    config: list[str] = field(default_factory=list)
    config_editor: Callable | None = None
    allow_users: list[str] = field(default_factory=list)
    allow_groups: list[str] = field(default_factory=list)
    deny_users: list[str] = field(default_factory=list)
    deny_groups: list[str] = field(default_factory=list)
    authorize: Callable | None = None

    @property
    def permission(self) -> Permission:
        """Build a :class:`~panel_flowdash.auth.Permission` from the declared rules."""
        return Permission.from_spec(
            allow_users=self.allow_users,
            allow_groups=self.allow_groups,
            deny_users=self.deny_users,
            deny_groups=self.deny_groups,
        )

    @classmethod
    def from_app(cls, app: Any) -> PanelAppMetadata:
        """Extract metadata from an app object."""
        metadata = getattr(app, "__panel_app_metadata__", None)
        if metadata is None:
            metadata = _APP_METADATA_BY_ID.get(id(app))
        if metadata is None:
            return cls()
        if isinstance(metadata, cls):
            return metadata
        if isinstance(metadata, dict):
            return cls(**metadata)
        raise TypeError("Unsupported panel app metadata type.")

allow_groups = field(default_factory=list) class-attribute instance-attribute

allow_users = field(default_factory=list) class-attribute instance-attribute

authorize = None class-attribute instance-attribute

component = False class-attribute instance-attribute

config = field(default_factory=list) class-attribute instance-attribute

config_editor = None class-attribute instance-attribute

config_schema = None class-attribute instance-attribute

default_size = None class-attribute instance-attribute

deny_groups = field(default_factory=list) class-attribute instance-attribute

deny_users = field(default_factory=list) class-attribute instance-attribute

description = None class-attribute instance-attribute

icon = None class-attribute instance-attribute

max_size = None class-attribute instance-attribute

min_size = None class-attribute instance-attribute

page = True class-attribute instance-attribute

permission property

Build a :class:~panel_flowdash.auth.Permission from the declared rules.

provides = field(default_factory=list) class-attribute instance-attribute

requires = field(default_factory=list) class-attribute instance-attribute

sidebar = False class-attribute instance-attribute

singleton = False class-attribute instance-attribute

tags = field(default_factory=list) class-attribute instance-attribute

title = None class-attribute instance-attribute

from_app(app) classmethod

Extract metadata from an app object.

Source code in src/panel_flowdash/registry.py
@classmethod
def from_app(cls, app: Any) -> PanelAppMetadata:
    """Extract metadata from an app object."""
    metadata = getattr(app, "__panel_app_metadata__", None)
    if metadata is None:
        metadata = _APP_METADATA_BY_ID.get(id(app))
    if metadata is None:
        return cls()
    if isinstance(metadata, cls):
        return metadata
    if isinstance(metadata, dict):
        return cls(**metadata)
    raise TypeError("Unsupported panel app metadata type.")

Permission dataclass

An allow/deny rule set evaluated against an :class:Identity.

All four fields match either the resolved user (OAuth login or system user) or one of the identity's groups. An empty Permission declares no constraints and defers entirely to the caller's default policy.

Source code in src/panel_flowdash/auth.py
@dataclass(frozen=True)
class Permission:
    """An allow/deny rule set evaluated against an :class:`Identity`.

    All four fields match either the resolved ``user`` (OAuth login *or* system
    user) or one of the identity's ``groups``. An empty ``Permission`` declares
    no constraints and defers entirely to the caller's default policy.
    """

    allow_users: frozenset[str] = field(default_factory=frozenset)
    allow_groups: frozenset[str] = field(default_factory=frozenset)
    deny_users: frozenset[str] = field(default_factory=frozenset)
    deny_groups: frozenset[str] = field(default_factory=frozenset)

    @property
    def is_empty(self) -> bool:
        """Whether the permission declares no allow or deny rules."""
        return not (self.allow_users or self.allow_groups or self.deny_users or self.deny_groups)

    @classmethod
    def from_spec(
        cls,
        *,
        allow_users: Iterable[str] | None = None,
        allow_groups: Iterable[str] | None = None,
        deny_users: Iterable[str] | None = None,
        deny_groups: Iterable[str] | None = None,
    ) -> Permission:
        """Build a :class:`Permission` from loosely-typed iterables."""
        return cls(
            allow_users=frozenset(allow_users or ()),
            allow_groups=frozenset(allow_groups or ()),
            deny_users=frozenset(deny_users or ()),
            deny_groups=frozenset(deny_groups or ()),
        )

    def to_dict(self) -> dict[str, list[str]]:
        """Serialize to sorted lists for JSON persistence."""
        return {
            "allow_users": sorted(self.allow_users),
            "allow_groups": sorted(self.allow_groups),
            "deny_users": sorted(self.deny_users),
            "deny_groups": sorted(self.deny_groups),
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any] | None) -> Permission:
        """Deserialize from a (possibly ``None`` or partial) mapping."""
        data = data or {}
        return cls.from_spec(
            allow_users=data.get("allow_users"),
            allow_groups=data.get("allow_groups"),
            deny_users=data.get("deny_users"),
            deny_groups=data.get("deny_groups"),
        )

allow_groups = field(default_factory=frozenset) class-attribute instance-attribute

allow_users = field(default_factory=frozenset) class-attribute instance-attribute

deny_groups = field(default_factory=frozenset) class-attribute instance-attribute

deny_users = field(default_factory=frozenset) class-attribute instance-attribute

is_empty property

Whether the permission declares no allow or deny rules.

from_dict(data) classmethod

Deserialize from a (possibly None or partial) mapping.

Source code in src/panel_flowdash/auth.py
@classmethod
def from_dict(cls, data: dict[str, Any] | None) -> Permission:
    """Deserialize from a (possibly ``None`` or partial) mapping."""
    data = data or {}
    return cls.from_spec(
        allow_users=data.get("allow_users"),
        allow_groups=data.get("allow_groups"),
        deny_users=data.get("deny_users"),
        deny_groups=data.get("deny_groups"),
    )

from_spec(*, allow_users=None, allow_groups=None, deny_users=None, deny_groups=None) classmethod

Build a :class:Permission from loosely-typed iterables.

Source code in src/panel_flowdash/auth.py
@classmethod
def from_spec(
    cls,
    *,
    allow_users: Iterable[str] | None = None,
    allow_groups: Iterable[str] | None = None,
    deny_users: Iterable[str] | None = None,
    deny_groups: Iterable[str] | None = None,
) -> Permission:
    """Build a :class:`Permission` from loosely-typed iterables."""
    return cls(
        allow_users=frozenset(allow_users or ()),
        allow_groups=frozenset(allow_groups or ()),
        deny_users=frozenset(deny_users or ()),
        deny_groups=frozenset(deny_groups or ()),
    )

to_dict()

Serialize to sorted lists for JSON persistence.

Source code in src/panel_flowdash/auth.py
def to_dict(self) -> dict[str, list[str]]:
    """Serialize to sorted lists for JSON persistence."""
    return {
        "allow_users": sorted(self.allow_users),
        "allow_groups": sorted(self.allow_groups),
        "deny_users": sorted(self.deny_users),
        "deny_groups": sorted(self.deny_groups),
    }

RegistryEntry dataclass

A registered component/page with its metadata.

Source code in src/panel_flowdash/registry.py
@dataclass
class RegistryEntry:
    """A registered component/page with its metadata."""

    app_id: str
    section: str
    name: str
    page_path: str
    module_name: str
    metadata: PanelAppMetadata
    module_path: pathlib.Path | None = None
    app: Any = None

    @property
    def title(self) -> str:
        """Human-readable title."""
        return self.metadata.title or self.name.replace("_", " ")

    def load(self) -> Any:
        """Import the module and return the app object.

        Caches the result on ``self.app``.  Raises on import failure.
        """
        if self.app is not None:
            return self.app
        module = importlib.import_module(self.module_name)
        app = getattr(module, "app", None)
        if app is None:
            raise ImportError(f"Module '{self.module_name}' has no 'app' export.")
        # Refresh metadata from the live object (decorators may carry richer info
        # e.g. config_schema / config_editor that AST cannot capture).
        object.__setattr__(self, "app", app)
        live_metadata = PanelAppMetadata.from_app(app)
        # Only replace if the live decorator actually produced a non-default result
        # (guards against bare Viewer classes with no @register decorator).
        if live_metadata != PanelAppMetadata():
            object.__setattr__(self, "metadata", live_metadata)
        return app

app = None class-attribute instance-attribute

app_id instance-attribute

metadata instance-attribute

module_name instance-attribute

module_path = None class-attribute instance-attribute

name instance-attribute

page_path instance-attribute

section instance-attribute

title property

Human-readable title.

load()

Import the module and return the app object.

Caches the result on self.app. Raises on import failure.

Source code in src/panel_flowdash/registry.py
def load(self) -> Any:
    """Import the module and return the app object.

    Caches the result on ``self.app``.  Raises on import failure.
    """
    if self.app is not None:
        return self.app
    module = importlib.import_module(self.module_name)
    app = getattr(module, "app", None)
    if app is None:
        raise ImportError(f"Module '{self.module_name}' has no 'app' export.")
    # Refresh metadata from the live object (decorators may carry richer info
    # e.g. config_schema / config_editor that AST cannot capture).
    object.__setattr__(self, "app", app)
    live_metadata = PanelAppMetadata.from_app(app)
    # Only replace if the live decorator actually produced a non-default result
    # (guards against bare Viewer classes with no @register decorator).
    if live_metadata != PanelAppMetadata():
        object.__setattr__(self, "metadata", live_metadata)
    return app

_eval_literal(node)

Safely evaluate an AST literal node; return None on failure.

Source code in src/panel_flowdash/registry.py
def _eval_literal(node: ast.expr) -> Any:
    """Safely evaluate an AST literal node; return None on failure."""
    try:
        return ast.literal_eval(node)
    except Exception:
        return None

_extract_register_kwargs(source)

Parse source and return the kwargs of the first @register/@panel_app call.

Only literal-evaluable arguments are captured; runtime expressions (e.g. config_schema=MyParamClass) are silently skipped — they are picked up later when the module is actually imported.

Returns None if no @register call is found.

Source code in src/panel_flowdash/registry.py
def _extract_register_kwargs(source: str) -> dict[str, Any] | None:
    """Parse *source* and return the kwargs of the first @register/@panel_app call.

    Only literal-evaluable arguments are captured; runtime expressions (e.g.
    ``config_schema=MyParamClass``) are silently skipped — they are picked up
    later when the module is actually imported.

    Returns ``None`` if no @register call is found.
    """
    try:
        tree = ast.parse(source)
    except SyntaxError:
        return None

    for node in ast.walk(tree):
        if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
            continue
        for decorator in node.decorator_list:
            call = decorator if isinstance(decorator, ast.Call) else None
            if call is None:
                continue
            func = call.func
            name = (
                func.id
                if isinstance(func, ast.Name)
                else func.attr
                if isinstance(func, ast.Attribute)
                else None
            )
            if name not in _REGISTER_NAMES:
                continue
            kwargs: dict[str, Any] = {}
            for kw in call.keywords:
                if kw.arg in _LITERAL_KEYS:
                    val = _eval_literal(kw.value)
                    if val is not None or isinstance(kw.value, ast.Constant):
                        kwargs[kw.arg] = val
            return kwargs

    return None

build_registry(project_dir)

Scan project_dir for page/component modules without importing them.

Reads each .py file with the AST to extract @register metadata. Modules are not imported at this stage; each RegistryEntry.app is None until RegistryEntry.load() is called.

Source code in src/panel_flowdash/registry.py
def build_registry(project_dir: Path) -> dict[str, RegistryEntry]:
    """Scan *project_dir* for page/component modules without importing them.

    Reads each ``.py`` file with the AST to extract ``@register`` metadata.
    Modules are **not** imported at this stage; each ``RegistryEntry.app`` is
    ``None`` until ``RegistryEntry.load()`` is called.
    """
    registry: dict[str, RegistryEntry] = {}

    for section_dir in sorted(project_dir.glob("*")):
        if not section_dir.is_dir() or section_dir.name.startswith(("_", ".")):
            continue
        section = section_dir.name
        for module_path in sorted(section_dir.glob("*.py")):
            if module_path.name.startswith("_"):
                continue

            source = module_path.read_text(encoding="utf-8")
            kwargs = _extract_register_kwargs(source)
            if kwargs is None:
                # No @register call found — skip silently (same as before).
                continue

            # Defaults that match PanelAppMetadata
            page = kwargs.get("page", True)
            component = kwargs.get("component", False)
            if not page and not component:
                continue

            metadata = PanelAppMetadata(
                page=bool(page),
                component=bool(component),
                sidebar=bool(kwargs.get("sidebar", False)),
                title=kwargs.get("title"),
                icon=kwargs.get("icon"),
                description=kwargs.get("description"),
                tags=list(kwargs.get("tags") or []),
                default_size=kwargs.get("default_size"),
                min_size=kwargs.get("min_size"),
                max_size=kwargs.get("max_size"),
                singleton=bool(kwargs.get("singleton", False)),
                provides=list(kwargs.get("provides") or []),
                requires=list(kwargs.get("requires") or []),
                config=list(kwargs.get("config") or []),
                allow_users=list(kwargs.get("allow_users") or []),
                allow_groups=list(kwargs.get("allow_groups") or []),
                deny_users=list(kwargs.get("deny_users") or []),
                deny_groups=list(kwargs.get("deny_groups") or []),
            )

            module_name = ".".join(module_path.relative_to(project_dir).with_suffix("").parts)
            app_id = f"{section}/{module_path.stem}"
            registry[app_id] = RegistryEntry(
                app_id=app_id,
                section=section,
                name=module_path.stem,
                page_path=f"/{app_id}",
                module_name=module_name,
                metadata=metadata,
                module_path=module_path,
                app=None,
            )

    return registry

register(*, page=True, component=False, sidebar=False, title=None, icon=None, description=None, tags=None, default_size=None, min_size=None, max_size=None, singleton=False, provides=None, requires=None, config_schema=None, config=None, config_editor=None, allow_users=None, allow_groups=None, deny_users=None, deny_groups=None, authorize=None)

Metadata-only decorator for app exports.

Annotates an app object/callable without altering runtime behavior.

The config_schema, config and config_editor arguments declare design-time configuration options that appear in the node editor. Use config_schema (a param.Parameterized subclass, a Pydantic model, or a JSON Schema dict) to define config explicitly, or config to name which of a Viewer's own params are configuration rather than input ports. Pass config_editor to supply a custom editor callable instead of the auto-generated form.

The allow_users, allow_groups, deny_users and deny_groups arguments declare page-level authorization rules. Users are matched against either the OAuth login or the system user; groups against the identity's resolved group membership. Deny rules always win; when only allow rules are present the identity must match at least one; with no rules the project's default policy applies. Pass authorize for a custom callable taking the resolved Identity and returning a bool (resolved on import).

Source code in src/panel_flowdash/registry.py
def register(
    *,
    page: bool = True,
    component: bool = False,
    sidebar: bool = False,
    title: str | None = None,
    icon: str | None = None,
    description: str | None = None,
    tags: list[str] | None = None,
    default_size: dict[str, Any] | None = None,
    min_size: dict[str, Any] | None = None,
    max_size: dict[str, Any] | None = None,
    singleton: bool = False,
    provides: list[str] | None = None,
    requires: list[Any] | None = None,
    config_schema: Any = None,
    config: list[str] | None = None,
    config_editor: Callable | None = None,
    allow_users: list[str] | None = None,
    allow_groups: list[str] | None = None,
    deny_users: list[str] | None = None,
    deny_groups: list[str] | None = None,
    authorize: Callable | None = None,
):
    """Metadata-only decorator for app exports.

    Annotates an app object/callable without altering runtime behavior.

    The ``config_schema``, ``config`` and ``config_editor`` arguments declare
    design-time configuration options that appear in the node editor. Use
    ``config_schema`` (a ``param.Parameterized`` subclass, a Pydantic model, or
    a JSON Schema dict) to define config explicitly, or ``config`` to name which
    of a Viewer's own params are configuration rather than input ports. Pass
    ``config_editor`` to supply a custom editor callable instead of the
    auto-generated form.

    The ``allow_users``, ``allow_groups``, ``deny_users`` and ``deny_groups``
    arguments declare page-level authorization rules. Users are matched against
    either the OAuth login or the system user; groups against the identity's
    resolved group membership. Deny rules always win; when only allow rules are
    present the identity must match at least one; with no rules the project's
    default policy applies. Pass ``authorize`` for a custom callable taking the
    resolved ``Identity`` and returning a bool (resolved on import).
    """
    metadata = PanelAppMetadata(
        page=page,
        component=component,
        sidebar=sidebar,
        title=title,
        icon=icon,
        description=description,
        tags=list(tags or []),
        default_size=default_size,
        min_size=min_size,
        max_size=max_size,
        singleton=singleton,
        provides=list(provides or []),
        requires=list(requires or []),
        config_schema=config_schema,
        config=list(config or []),
        config_editor=config_editor,
        allow_users=list(allow_users or []),
        allow_groups=list(allow_groups or []),
        deny_users=list(deny_users or []),
        deny_groups=list(deny_groups or []),
        authorize=authorize,
    )

    def _decorator(app):
        _APP_METADATA_BY_ID[id(app)] = metadata
        try:
            app.__panel_app_metadata__ = metadata
        except Exception:
            pass
        return app

    return _decorator

Component Spec

panel_flowdash.component_spec

Component specification with typed ports for the dataflow editor.

_BASE_PARAMS = set(param.Parameterized.param) module-attribute

ComponentSpec dataclass

Full specification of a component's ports and metadata.

Source code in src/panel_flowdash/component_spec.py
@dataclass(frozen=True)
class ComponentSpec:
    """Full specification of a component's ports and metadata."""

    component_id: str
    title: str
    description: str | None
    icon: str | None
    tags: list[str]
    outputs: list[OutputPort]
    inputs: list[InputPort]
    default_size: dict[str, Any] | None
    config: list[ConfigField] = field(default_factory=list)
    config_state_class: type[param.Parameterized] | None = None
    config_editor: Callable | None = None

component_id instance-attribute

config = field(default_factory=list) class-attribute instance-attribute

config_editor = None class-attribute instance-attribute

config_state_class = None class-attribute instance-attribute

default_size instance-attribute

description instance-attribute

icon instance-attribute

inputs instance-attribute

outputs instance-attribute

tags instance-attribute

title instance-attribute

ConfigField dataclass

Describes a single design-time configuration option on a component.

Source code in src/panel_flowdash/component_spec.py
@dataclass(frozen=True)
class ConfigField:
    """Describes a single design-time configuration option on a component."""

    name: str
    type: str | None = None
    label: str | None = None
    default: Any = None

default = None class-attribute instance-attribute

label = None class-attribute instance-attribute

name instance-attribute

type = None class-attribute instance-attribute

InputPort dataclass

Describes a single input port on a component node.

Source code in src/panel_flowdash/component_spec.py
@dataclass(frozen=True)
class InputPort:
    """Describes a single input port on a component node."""

    name: str
    type: str | None = None
    label: str | None = None
    required: bool = True
    blocking: bool = True
    default: Any = None

blocking = True class-attribute instance-attribute

default = None class-attribute instance-attribute

label = None class-attribute instance-attribute

name instance-attribute

required = True class-attribute instance-attribute

type = None class-attribute instance-attribute

OutputPort dataclass

Describes a single output port on a component node.

Source code in src/panel_flowdash/component_spec.py
@dataclass(frozen=True)
class OutputPort:
    """Describes a single output port on a component node."""

    name: str
    type: str | None = None
    label: str | None = None

label = None class-attribute instance-attribute

name instance-attribute

type = None class-attribute instance-attribute

PanelAppMetadata dataclass

Metadata attached to a component or page by the @register decorator.

Source code in src/panel_flowdash/registry.py
@dataclass(frozen=True)
class PanelAppMetadata:
    """Metadata attached to a component or page by the @register decorator."""

    page: bool = True
    component: bool = False
    sidebar: bool = False
    title: str | None = None
    icon: str | None = None
    description: str | None = None
    tags: list[str] = field(default_factory=list)
    default_size: dict[str, Any] | None = None
    min_size: dict[str, Any] | None = None
    max_size: dict[str, Any] | None = None
    singleton: bool = False
    provides: list[str] = field(default_factory=list)
    requires: list[Any] = field(default_factory=list)
    config_schema: Any = None
    config: list[str] = field(default_factory=list)
    config_editor: Callable | None = None
    allow_users: list[str] = field(default_factory=list)
    allow_groups: list[str] = field(default_factory=list)
    deny_users: list[str] = field(default_factory=list)
    deny_groups: list[str] = field(default_factory=list)
    authorize: Callable | None = None

    @property
    def permission(self) -> Permission:
        """Build a :class:`~panel_flowdash.auth.Permission` from the declared rules."""
        return Permission.from_spec(
            allow_users=self.allow_users,
            allow_groups=self.allow_groups,
            deny_users=self.deny_users,
            deny_groups=self.deny_groups,
        )

    @classmethod
    def from_app(cls, app: Any) -> PanelAppMetadata:
        """Extract metadata from an app object."""
        metadata = getattr(app, "__panel_app_metadata__", None)
        if metadata is None:
            metadata = _APP_METADATA_BY_ID.get(id(app))
        if metadata is None:
            return cls()
        if isinstance(metadata, cls):
            return metadata
        if isinstance(metadata, dict):
            return cls(**metadata)
        raise TypeError("Unsupported panel app metadata type.")

allow_groups = field(default_factory=list) class-attribute instance-attribute

allow_users = field(default_factory=list) class-attribute instance-attribute

authorize = None class-attribute instance-attribute

component = False class-attribute instance-attribute

config = field(default_factory=list) class-attribute instance-attribute

config_editor = None class-attribute instance-attribute

config_schema = None class-attribute instance-attribute

default_size = None class-attribute instance-attribute

deny_groups = field(default_factory=list) class-attribute instance-attribute

deny_users = field(default_factory=list) class-attribute instance-attribute

description = None class-attribute instance-attribute

icon = None class-attribute instance-attribute

max_size = None class-attribute instance-attribute

min_size = None class-attribute instance-attribute

page = True class-attribute instance-attribute

permission property

Build a :class:~panel_flowdash.auth.Permission from the declared rules.

provides = field(default_factory=list) class-attribute instance-attribute

requires = field(default_factory=list) class-attribute instance-attribute

sidebar = False class-attribute instance-attribute

singleton = False class-attribute instance-attribute

tags = field(default_factory=list) class-attribute instance-attribute

title = None class-attribute instance-attribute

from_app(app) classmethod

Extract metadata from an app object.

Source code in src/panel_flowdash/registry.py
@classmethod
def from_app(cls, app: Any) -> PanelAppMetadata:
    """Extract metadata from an app object."""
    metadata = getattr(app, "__panel_app_metadata__", None)
    if metadata is None:
        metadata = _APP_METADATA_BY_ID.get(id(app))
    if metadata is None:
        return cls()
    if isinstance(metadata, cls):
        return metadata
    if isinstance(metadata, dict):
        return cls(**metadata)
    raise TypeError("Unsupported panel app metadata type.")

RegistryEntry dataclass

A registered component/page with its metadata.

Source code in src/panel_flowdash/registry.py
@dataclass
class RegistryEntry:
    """A registered component/page with its metadata."""

    app_id: str
    section: str
    name: str
    page_path: str
    module_name: str
    metadata: PanelAppMetadata
    module_path: pathlib.Path | None = None
    app: Any = None

    @property
    def title(self) -> str:
        """Human-readable title."""
        return self.metadata.title or self.name.replace("_", " ")

    def load(self) -> Any:
        """Import the module and return the app object.

        Caches the result on ``self.app``.  Raises on import failure.
        """
        if self.app is not None:
            return self.app
        module = importlib.import_module(self.module_name)
        app = getattr(module, "app", None)
        if app is None:
            raise ImportError(f"Module '{self.module_name}' has no 'app' export.")
        # Refresh metadata from the live object (decorators may carry richer info
        # e.g. config_schema / config_editor that AST cannot capture).
        object.__setattr__(self, "app", app)
        live_metadata = PanelAppMetadata.from_app(app)
        # Only replace if the live decorator actually produced a non-default result
        # (guards against bare Viewer classes with no @register decorator).
        if live_metadata != PanelAppMetadata():
            object.__setattr__(self, "metadata", live_metadata)
        return app

app = None class-attribute instance-attribute

app_id instance-attribute

metadata instance-attribute

module_name instance-attribute

module_path = None class-attribute instance-attribute

name instance-attribute

page_path instance-attribute

section instance-attribute

title property

Human-readable title.

load()

Import the module and return the app object.

Caches the result on self.app. Raises on import failure.

Source code in src/panel_flowdash/registry.py
def load(self) -> Any:
    """Import the module and return the app object.

    Caches the result on ``self.app``.  Raises on import failure.
    """
    if self.app is not None:
        return self.app
    module = importlib.import_module(self.module_name)
    app = getattr(module, "app", None)
    if app is None:
        raise ImportError(f"Module '{self.module_name}' has no 'app' export.")
    # Refresh metadata from the live object (decorators may carry richer info
    # e.g. config_schema / config_editor that AST cannot capture).
    object.__setattr__(self, "app", app)
    live_metadata = PanelAppMetadata.from_app(app)
    # Only replace if the live decorator actually produced a non-default result
    # (guards against bare Viewer classes with no @register decorator).
    if live_metadata != PanelAppMetadata():
        object.__setattr__(self, "metadata", live_metadata)
    return app

_config_from_mapping(schema)

Build a config state class from a JSON-Schema-like properties dict.

Source code in src/panel_flowdash/component_spec.py
def _config_from_mapping(
    schema: dict[str, Any],
) -> tuple[type[param.Parameterized], list[ConfigField]]:
    """Build a config state class from a JSON-Schema-like properties dict."""
    properties = schema.get("properties", schema)
    fields: list[ConfigField] = []
    params: dict[str, param.Parameter] = {}
    for name, prop in properties.items():
        prop = prop if isinstance(prop, dict) else {}
        default = prop.get("default")
        enum = prop.get("enum")
        if enum is not None:
            p = param.Selector(default=default, objects=list(enum))
        else:
            p = param.Parameter(default=default, allow_None=True)
        params[name] = p
        fields.append(
            ConfigField(
                name=name,
                type=prop.get("type"),
                label=prop.get("title", name),
                default=default,
            )
        )
    state_cls = type("ConfigState_schema", (param.Parameterized,), params)
    return state_cls, fields

_config_from_metadata(metadata, viewer_cls=None)

Resolve config into a state class and field list from component metadata.

Source code in src/panel_flowdash/component_spec.py
def _config_from_metadata(
    metadata: PanelAppMetadata,
    viewer_cls: type | None = None,
) -> tuple[type[param.Parameterized] | None, list[ConfigField]]:
    """Resolve config into a state class and field list from component metadata."""
    schema = metadata.config_schema
    if isinstance(schema, type) and issubclass(schema, param.Parameterized):
        return _config_from_param_class(schema)
    if isinstance(schema, type) and _is_pydantic_model(schema):
        return _config_from_mapping(_pydantic_to_properties(schema))
    if isinstance(schema, dict):
        return _config_from_mapping(schema)
    if metadata.config and viewer_cls is not None:
        return _config_from_param_class(viewer_cls, names=metadata.config)
    return None, []

_config_from_param_class(cls, names=None)

Build a config state class and field list from a Parameterized subclass.

When names is given, only those params are treated as config; otherwise every non-base param on the class is used.

Source code in src/panel_flowdash/component_spec.py
def _config_from_param_class(
    cls: type[param.Parameterized],
    names: list[str] | None = None,
) -> tuple[type[param.Parameterized], list[ConfigField]]:
    """Build a config state class and field list from a Parameterized subclass.

    When ``names`` is given, only those params are treated as config; otherwise
    every non-base param on the class is used.
    """
    fields: list[ConfigField] = []
    selected: dict[str, param.Parameter] = {}
    for pname, p in cls.param.objects("existing").items():
        if pname in _BASE_PARAMS or pname.startswith("_"):
            continue
        if names is not None and pname not in names:
            continue
        selected[pname] = p
        fields.append(
            ConfigField(
                name=pname,
                type=_param_type_name(p),
                label=p.label or pname,
                default=p.default,
            )
        )

    params = {name: copy.copy(p) for name, p in selected.items()}
    state_cls = type(f"ConfigState_{cls.__name__}", (param.Parameterized,), params)
    return state_cls, fields

_is_pydantic_model(cls)

Source code in src/panel_flowdash/component_spec.py
def _is_pydantic_model(cls: type) -> bool:
    return hasattr(cls, "model_fields") and hasattr(cls, "model_validate")

_param_type_name(p)

Source code in src/panel_flowdash/component_spec.py
def _param_type_name(p: param.Parameter) -> str:
    return type(p).__name__

_ports_from_metadata(metadata)

Source code in src/panel_flowdash/component_spec.py
def _ports_from_metadata(
    metadata: PanelAppMetadata,
) -> tuple[list[OutputPort], list[InputPort]]:
    outputs = []
    for item in metadata.provides:
        if isinstance(item, str):
            outputs.append(OutputPort(name=item))
        elif isinstance(item, dict):
            outputs.append(
                OutputPort(
                    name=item["key"],
                    type=item.get("type"),
                    label=item.get("label"),
                )
            )

    inputs = []
    for item in metadata.requires:
        if isinstance(item, str):
            inputs.append(InputPort(name=item))
        elif isinstance(item, dict):
            inputs.append(
                InputPort(
                    name=item.get("key", ""),
                    type=item.get("type"),
                    label=item.get("label"),
                    required=item.get("required", True),
                    blocking=item.get("blocking", True),
                    default=item.get("fallback"),
                )
            )

    return outputs, inputs

_ports_from_viewer_class(viewer_cls, exclude=None)

Source code in src/panel_flowdash/component_spec.py
def _ports_from_viewer_class(
    viewer_cls: type,
    exclude: set[str] | None = None,
) -> tuple[list[OutputPort], list[InputPort]]:
    exclude = exclude or set()
    instance = viewer_cls()
    output_info = instance.param.outputs()
    mro_dicts = [cls.__dict__ for cls in viewer_cls.__mro__]

    outputs = []
    for name, (ptype, _method, _index) in output_info.items():
        if not any(name in d for d in mro_dicts):
            continue
        if ptype is None:
            type_str = None
        elif isinstance(ptype, type):
            type_str = ptype.__name__
        else:
            type_str = type(ptype).__name__
        outputs.append(OutputPort(name=name, type=type_str))

    inputs = []
    for pname, p in viewer_cls.param.objects("existing").items():
        if pname in _BASE_PARAMS or pname.startswith("_") or pname in exclude:
            continue
        type_str = type(p).__name__ if p else None
        inputs.append(
            InputPort(
                name=pname,
                type=type_str,
                required=False,
                blocking=False,
            )
        )

    return outputs, inputs

_pydantic_to_properties(cls)

Source code in src/panel_flowdash/component_spec.py
def _pydantic_to_properties(cls: type) -> dict[str, Any]:
    properties = {}
    for name, fieldinfo in cls.model_fields.items():
        default = getattr(fieldinfo, "default", None)
        if default is ... or repr(default) == "PydanticUndefined":
            default = None
        properties[name] = {"default": default, "title": name}
    return {"properties": properties}

build_component_spec(entry)

Build a ComponentSpec from a registry entry.

Source code in src/panel_flowdash/component_spec.py
def build_component_spec(entry: RegistryEntry) -> ComponentSpec:
    """Build a ComponentSpec from a registry entry."""
    app = entry.app
    metadata = entry.metadata

    is_viewer = isinstance(app, type) and issubclass(app, Viewer)
    config_state_class, config = _config_from_metadata(
        metadata, viewer_cls=app if is_viewer else None
    )
    config_names = {f.name for f in config}

    if is_viewer:
        outputs, inputs = _ports_from_viewer_class(app, exclude=config_names)
        dec_outputs, dec_inputs = _ports_from_metadata(metadata)
        if dec_outputs:
            outputs = dec_outputs
        if dec_inputs:
            inputs = [p for p in dec_inputs if p.name not in config_names]
    else:
        outputs, inputs = _ports_from_metadata(metadata)
        inputs = [p for p in inputs if p.name not in config_names]

    return ComponentSpec(
        component_id=entry.app_id,
        title=entry.title,
        description=metadata.description,
        icon=metadata.icon,
        tags=metadata.tags,
        outputs=outputs,
        inputs=inputs,
        default_size=metadata.default_size,
        config=config,
        config_state_class=config_state_class,
        config_editor=metadata.config_editor,
    )

build_component_specs(registry)

Build specs for all component-enabled entries in a registry.

Source code in src/panel_flowdash/component_spec.py
def build_component_specs(
    registry: dict[str, RegistryEntry],
) -> dict[str, ComponentSpec]:
    """Build specs for all component-enabled entries in a registry."""
    specs = {}
    for app_id, entry in registry.items():
        if entry.metadata.component:
            specs[app_id] = build_component_spec(entry)
    return specs

Dataflow Engine

panel_flowdash.dataflow_engine

Dataflow wiring engine with runtime validation.

Each node in the graph gets a NodeState (a dynamic Parameterized subclass) whose parameters correspond to the node's declared input and output ports. Edges are wired via param.watch: when a source port changes, the value is assigned to the target port inside a try/except so that runtime type errors (e.g. param validation failures) are caught and reported via an error callback.

_RESERVED_PARAMS = set(param.Parameterized.param) module-attribute

ComponentSpec dataclass

Full specification of a component's ports and metadata.

Source code in src/panel_flowdash/component_spec.py
@dataclass(frozen=True)
class ComponentSpec:
    """Full specification of a component's ports and metadata."""

    component_id: str
    title: str
    description: str | None
    icon: str | None
    tags: list[str]
    outputs: list[OutputPort]
    inputs: list[InputPort]
    default_size: dict[str, Any] | None
    config: list[ConfigField] = field(default_factory=list)
    config_state_class: type[param.Parameterized] | None = None
    config_editor: Callable | None = None

component_id instance-attribute

config = field(default_factory=list) class-attribute instance-attribute

config_editor = None class-attribute instance-attribute

config_state_class = None class-attribute instance-attribute

default_size instance-attribute

description instance-attribute

icon instance-attribute

inputs instance-attribute

outputs instance-attribute

tags instance-attribute

title instance-attribute

DataflowGraph

Manages node state instances and wires edges with runtime validation.

Source code in src/panel_flowdash/dataflow_engine.py
class DataflowGraph:
    """Manages node state instances and wires edges with runtime validation."""

    def __init__(
        self,
        specs: dict[str, ComponentSpec],
        on_error: Callable[[str, str, str, str, Exception], None] | None = None,
    ):
        """Initialize the dataflow graph.

        Parameters
        ----------
        specs
            Component specs keyed by component_id.
        on_error
            Callback invoked when a runtime value assignment fails.
            Signature: (source_id, source_port, target_id, target_port, exception)
        """
        self._specs = specs
        self._state_classes: dict[str, type[param.Parameterized]] = {}
        self._nodes: dict[str, param.Parameterized] = {}
        self._config_states: dict[str, param.Parameterized] = {}
        self._node_specs: dict[str, ComponentSpec] = {}
        self._edges: list[dict[str, str]] = []
        self._watchers: dict[tuple, param.parameterized.Watcher] = {}
        self._on_error = on_error

        for comp_id, spec in specs.items():
            self._state_classes[comp_id] = build_node_state_class(spec)

    def add_node(self, instance_id: str, component_id: str) -> param.Parameterized:
        """Create a new node state instance."""
        spec = self._specs[component_id]
        cls = self._state_classes[component_id]
        state = cls(name=instance_id)
        self._nodes[instance_id] = state
        self._node_specs[instance_id] = spec
        if spec.config_state_class is not None:
            self._config_states[instance_id] = spec.config_state_class(
                name=f"{instance_id}_config"
            )
        return state

    def remove_node(self, instance_id: str):
        """Remove a node and any edges connected to it."""
        keys_to_remove = [k for k in self._watchers if k[0] == instance_id or k[2] == instance_id]
        for key in keys_to_remove:
            watcher = self._watchers.pop(key)
            src = self._nodes.get(key[0])
            if src is not None:
                src.param.unwatch(watcher)

        self._edges = [
            e for e in self._edges if e["source"] != instance_id and e["target"] != instance_id
        ]
        self._nodes.pop(instance_id, None)
        self._config_states.pop(instance_id, None)
        self._node_specs.pop(instance_id, None)

    def _is_list_input(self, target_id: str, target_port: str) -> bool:
        """Return True if the target port is a List-typed multi-connection input."""
        spec = self._node_specs.get(target_id)
        if not spec:
            return False
        for port in spec.inputs:
            if port.name == target_port:
                return _is_list_port(port)
        return False

    def _rebuild_list_port(self, target_id: str, target_port: str):
        """Recompute the list value for a multi-connection port from all sources."""
        target_state = self._nodes.get(target_id)
        if target_state is None:
            return
        values = []
        for e in self._edges:
            if e["target"] == target_id and e["target_port"] == target_port:
                src_state = self._nodes.get(e["source"])
                if src_state is not None:
                    val = getattr(src_state, e["source_port"], None)
                    if val is not None:
                        values.append(val)
        try:
            setattr(target_state, target_port, values)
        except Exception as exc:
            if self._on_error:
                self._on_error("", "", target_id, target_port, exc)

    def add_edge(
        self,
        source_id: str,
        source_port: str,
        target_id: str,
        target_port: str,
    ) -> bool | str:
        """Wire an edge between two ports.

        Returns True on success, or an error message string on failure.
        """
        source_state = self._nodes.get(source_id)
        target_state = self._nodes.get(target_id)
        if source_state is None or target_state is None:
            return "Source or target node not found."
        if not hasattr(source_state.param, source_port):
            return f"Output port '{source_port}' does not exist on source node."
        if not hasattr(target_state.param, target_port):
            return f"Input port '{target_port}' does not exist on target node."

        is_list = self._is_list_input(target_id, target_port)

        if not is_list:
            for e in self._edges:
                if e["target"] == target_id and e["target_port"] == target_port:
                    return f"Input '{target_port}' already has a connection. Disconnect it first."

        if self._would_create_cycle(source_id, target_id):
            return "Connection rejected: would create a cycle."

        source_spec = self._node_specs.get(source_id)
        target_spec = self._node_specs.get(target_id)
        if source_spec and target_spec and not is_list:
            error = self._check_type_compatibility(
                source_spec, source_port, target_spec, target_port
            )
            if error:
                return error

        if is_list:

            def _propagate(
                event,
                _src_id=source_id,
                _src_port=source_port,
                _tgt_id=target_id,
                _tgt_port=target_port,
            ):
                try:
                    self._rebuild_list_port(_tgt_id, _tgt_port)
                except Exception as exc:
                    if self._on_error:
                        self._on_error(_src_id, _src_port, _tgt_id, _tgt_port, exc)
        else:

            def _propagate(
                event,
                _src_id=source_id,
                _src_port=source_port,
                _tgt_id=target_id,
                _tgt_port=target_port,
                _target=target_state,
            ):
                try:
                    setattr(_target, _tgt_port, event.new)
                except Exception as exc:
                    if self._on_error:
                        self._on_error(_src_id, _src_port, _tgt_id, _tgt_port, exc)

        watcher = source_state.param.watch(_propagate, source_port)
        edge_key = (source_id, source_port, target_id, target_port)
        self._watchers[edge_key] = watcher

        self._edges.append(
            {
                "source": source_id,
                "source_port": source_port,
                "target": target_id,
                "target_port": target_port,
            }
        )

        if is_list:
            self._rebuild_list_port(target_id, target_port)
        else:
            current = getattr(source_state, source_port)
            if current is not None:
                try:
                    setattr(target_state, target_port, current)
                except Exception as exc:
                    if self._on_error:
                        self._on_error(source_id, source_port, target_id, target_port, exc)

        return True

    def _would_create_cycle(self, source_id: str, target_id: str) -> bool:
        """Return True if adding an edge from source to target would create a cycle."""
        if source_id == target_id:
            return True
        visited = set()
        queue = [target_id]
        while queue:
            node = queue.pop(0)
            if node == source_id:
                return True
            if node in visited:
                continue
            visited.add(node)
            for e in self._edges:
                if e["source"] == node:
                    queue.append(e["target"])
        return False

    def _check_type_compatibility(
        self,
        source_spec: ComponentSpec,
        source_port: str,
        target_spec: ComponentSpec,
        target_port: str,
    ) -> str | None:
        """Return an error message if types are incompatible, None if OK."""
        source_type = None
        for port in source_spec.outputs:
            if port.name == source_port:
                source_type = port.type
                break

        target_type = None
        for port in target_spec.inputs:
            if port.name == target_port:
                target_type = port.type
                break

        if source_type is None or target_type is None:
            return None

        if source_type.lower() == target_type.lower():
            return None

        return (
            f"Type mismatch: output '{source_port}' produces '{source_type}' "
            f"but input '{target_port}' expects '{target_type}'."
        )

    def remove_edge(self, source_id: str, source_port: str, target_id: str, target_port: str):
        """Remove an edge, unsubscribe the watcher, and reset target to default."""
        is_list = self._is_list_input(target_id, target_port)

        self._edges = [
            e
            for e in self._edges
            if not (
                e["source"] == source_id
                and e["source_port"] == source_port
                and e["target"] == target_id
                and e["target_port"] == target_port
            )
        ]

        edge_key = (source_id, source_port, target_id, target_port)
        watcher = self._watchers.pop(edge_key, None)
        if watcher is not None:
            source_state = self._nodes.get(source_id)
            if source_state is not None:
                source_state.param.unwatch(watcher)

        if is_list:
            self._rebuild_list_port(target_id, target_port)
        else:
            target_state = self._nodes.get(target_id)
            if target_state is not None and hasattr(target_state, target_port):
                spec = self._node_specs.get(target_id)
                default = None
                if spec:
                    for port in spec.inputs:
                        if port.name == target_port:
                            default = port.default
                            break
                setattr(target_state, target_port, default)

    def get_state(self, instance_id: str) -> param.Parameterized | None:
        """Get the state instance for a node."""
        return self._nodes.get(instance_id)

    def get_config_state(self, instance_id: str) -> param.Parameterized | None:
        """Get the config state instance for a node, if it has config."""
        return self._config_states.get(instance_id)

    @property
    def edges(self) -> list[dict[str, str]]:
        """All current edges."""
        return list(self._edges)

    @property
    def node_ids(self) -> list[str]:
        """All current node instance IDs."""
        return list(self._nodes.keys())

edges property

All current edges.

node_ids property

All current node instance IDs.

add_edge(source_id, source_port, target_id, target_port)

Wire an edge between two ports.

Returns True on success, or an error message string on failure.

Source code in src/panel_flowdash/dataflow_engine.py
def add_edge(
    self,
    source_id: str,
    source_port: str,
    target_id: str,
    target_port: str,
) -> bool | str:
    """Wire an edge between two ports.

    Returns True on success, or an error message string on failure.
    """
    source_state = self._nodes.get(source_id)
    target_state = self._nodes.get(target_id)
    if source_state is None or target_state is None:
        return "Source or target node not found."
    if not hasattr(source_state.param, source_port):
        return f"Output port '{source_port}' does not exist on source node."
    if not hasattr(target_state.param, target_port):
        return f"Input port '{target_port}' does not exist on target node."

    is_list = self._is_list_input(target_id, target_port)

    if not is_list:
        for e in self._edges:
            if e["target"] == target_id and e["target_port"] == target_port:
                return f"Input '{target_port}' already has a connection. Disconnect it first."

    if self._would_create_cycle(source_id, target_id):
        return "Connection rejected: would create a cycle."

    source_spec = self._node_specs.get(source_id)
    target_spec = self._node_specs.get(target_id)
    if source_spec and target_spec and not is_list:
        error = self._check_type_compatibility(
            source_spec, source_port, target_spec, target_port
        )
        if error:
            return error

    if is_list:

        def _propagate(
            event,
            _src_id=source_id,
            _src_port=source_port,
            _tgt_id=target_id,
            _tgt_port=target_port,
        ):
            try:
                self._rebuild_list_port(_tgt_id, _tgt_port)
            except Exception as exc:
                if self._on_error:
                    self._on_error(_src_id, _src_port, _tgt_id, _tgt_port, exc)
    else:

        def _propagate(
            event,
            _src_id=source_id,
            _src_port=source_port,
            _tgt_id=target_id,
            _tgt_port=target_port,
            _target=target_state,
        ):
            try:
                setattr(_target, _tgt_port, event.new)
            except Exception as exc:
                if self._on_error:
                    self._on_error(_src_id, _src_port, _tgt_id, _tgt_port, exc)

    watcher = source_state.param.watch(_propagate, source_port)
    edge_key = (source_id, source_port, target_id, target_port)
    self._watchers[edge_key] = watcher

    self._edges.append(
        {
            "source": source_id,
            "source_port": source_port,
            "target": target_id,
            "target_port": target_port,
        }
    )

    if is_list:
        self._rebuild_list_port(target_id, target_port)
    else:
        current = getattr(source_state, source_port)
        if current is not None:
            try:
                setattr(target_state, target_port, current)
            except Exception as exc:
                if self._on_error:
                    self._on_error(source_id, source_port, target_id, target_port, exc)

    return True

add_node(instance_id, component_id)

Create a new node state instance.

Source code in src/panel_flowdash/dataflow_engine.py
def add_node(self, instance_id: str, component_id: str) -> param.Parameterized:
    """Create a new node state instance."""
    spec = self._specs[component_id]
    cls = self._state_classes[component_id]
    state = cls(name=instance_id)
    self._nodes[instance_id] = state
    self._node_specs[instance_id] = spec
    if spec.config_state_class is not None:
        self._config_states[instance_id] = spec.config_state_class(
            name=f"{instance_id}_config"
        )
    return state

get_config_state(instance_id)

Get the config state instance for a node, if it has config.

Source code in src/panel_flowdash/dataflow_engine.py
def get_config_state(self, instance_id: str) -> param.Parameterized | None:
    """Get the config state instance for a node, if it has config."""
    return self._config_states.get(instance_id)

get_state(instance_id)

Get the state instance for a node.

Source code in src/panel_flowdash/dataflow_engine.py
def get_state(self, instance_id: str) -> param.Parameterized | None:
    """Get the state instance for a node."""
    return self._nodes.get(instance_id)

remove_edge(source_id, source_port, target_id, target_port)

Remove an edge, unsubscribe the watcher, and reset target to default.

Source code in src/panel_flowdash/dataflow_engine.py
def remove_edge(self, source_id: str, source_port: str, target_id: str, target_port: str):
    """Remove an edge, unsubscribe the watcher, and reset target to default."""
    is_list = self._is_list_input(target_id, target_port)

    self._edges = [
        e
        for e in self._edges
        if not (
            e["source"] == source_id
            and e["source_port"] == source_port
            and e["target"] == target_id
            and e["target_port"] == target_port
        )
    ]

    edge_key = (source_id, source_port, target_id, target_port)
    watcher = self._watchers.pop(edge_key, None)
    if watcher is not None:
        source_state = self._nodes.get(source_id)
        if source_state is not None:
            source_state.param.unwatch(watcher)

    if is_list:
        self._rebuild_list_port(target_id, target_port)
    else:
        target_state = self._nodes.get(target_id)
        if target_state is not None and hasattr(target_state, target_port):
            spec = self._node_specs.get(target_id)
            default = None
            if spec:
                for port in spec.inputs:
                    if port.name == target_port:
                        default = port.default
                        break
            setattr(target_state, target_port, default)

remove_node(instance_id)

Remove a node and any edges connected to it.

Source code in src/panel_flowdash/dataflow_engine.py
def remove_node(self, instance_id: str):
    """Remove a node and any edges connected to it."""
    keys_to_remove = [k for k in self._watchers if k[0] == instance_id or k[2] == instance_id]
    for key in keys_to_remove:
        watcher = self._watchers.pop(key)
        src = self._nodes.get(key[0])
        if src is not None:
            src.param.unwatch(watcher)

    self._edges = [
        e for e in self._edges if e["source"] != instance_id and e["target"] != instance_id
    ]
    self._nodes.pop(instance_id, None)
    self._config_states.pop(instance_id, None)
    self._node_specs.pop(instance_id, None)

_is_list_port(port)

Return True if a port's declared type indicates a list/multi-connection input.

Source code in src/panel_flowdash/dataflow_engine.py
def _is_list_port(port) -> bool:
    """Return True if a port's declared type indicates a list/multi-connection input."""
    if port.type is None:
        return False
    return port.type.lower() in ("list", "List")

build_node_state_class(spec)

Create a Parameterized subclass with one param per port (inputs + outputs).

Source code in src/panel_flowdash/dataflow_engine.py
def build_node_state_class(spec: ComponentSpec) -> type[param.Parameterized]:
    """Create a Parameterized subclass with one param per port (inputs + outputs)."""
    params: dict[str, param.Parameter] = {}

    for port in spec.inputs:
        if port.name in _RESERVED_PARAMS:
            continue
        if _is_list_port(port):
            params[port.name] = param.List(default=port.default or [], allow_refs=True)
        else:
            params[port.name] = param.Parameter(
                default=port.default, allow_None=True, allow_refs=True
            )

    for port in spec.outputs:
        if port.name in _RESERVED_PARAMS:
            continue
        if port.name not in params:
            params[port.name] = param.Parameter(
                default=None,
                allow_None=True,
                allow_refs=True,
            )

    class_name = f"NodeState_{spec.component_id.replace('/', '_')}"
    return type(class_name, (param.Parameterized,), params)

Dashboard Store

panel_flowdash.dashboard_store

SQLite-backed persistence for dashboard graphs.

DashboardEdge dataclass

A connection between two component ports.

Source code in src/panel_flowdash/dashboard_store.py
@dataclass
class DashboardEdge:
    """A connection between two component ports."""

    source: str
    source_port: str
    target: str
    target_port: str

    def to_dict(self) -> dict[str, str]:
        return {
            "source": self.source,
            "source_port": self.source_port,
            "target": self.target,
            "target_port": self.target_port,
        }

    @classmethod
    def from_dict(cls, data: dict[str, str]) -> DashboardEdge:
        return cls(
            source=data["source"],
            source_port=data["source_port"],
            target=data["target"],
            target_port=data["target_port"],
        )

source instance-attribute

source_port instance-attribute

target instance-attribute

target_port instance-attribute

from_dict(data) classmethod

Source code in src/panel_flowdash/dashboard_store.py
@classmethod
def from_dict(cls, data: dict[str, str]) -> DashboardEdge:
    return cls(
        source=data["source"],
        source_port=data["source_port"],
        target=data["target"],
        target_port=data["target_port"],
    )

to_dict()

Source code in src/panel_flowdash/dashboard_store.py
def to_dict(self) -> dict[str, str]:
    return {
        "source": self.source,
        "source_port": self.source_port,
        "target": self.target,
        "target_port": self.target_port,
    }

DashboardItem dataclass

A component instance on the dashboard.

x, y store the ReactFlow node canvas position. Grid layout (widths, heights, visibility) lives in DashboardModel.tile_layout.

Source code in src/panel_flowdash/dashboard_store.py
@dataclass
class DashboardItem:
    """A component instance on the dashboard.

    x, y store the ReactFlow node canvas position.
    Grid layout (widths, heights, visibility) lives in DashboardModel.tile_layout.
    """

    instance_id: str
    component_id: str
    x: float = 0
    y: float = 0
    config: dict[str, Any] = field(default_factory=dict)

    def to_dict(self) -> dict[str, Any]:
        return {
            "instance_id": self.instance_id,
            "component_id": self.component_id,
            "x": self.x,
            "y": self.y,
            "config": self.config,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> DashboardItem:
        return cls(
            instance_id=data["instance_id"],
            component_id=data["component_id"],
            x=data.get("x", 0),
            y=data.get("y", 0),
            config=data.get("config", {}),
        )

component_id instance-attribute

config = field(default_factory=dict) class-attribute instance-attribute

instance_id instance-attribute

x = 0 class-attribute instance-attribute

y = 0 class-attribute instance-attribute

from_dict(data) classmethod

Source code in src/panel_flowdash/dashboard_store.py
@classmethod
def from_dict(cls, data: dict[str, Any]) -> DashboardItem:
    return cls(
        instance_id=data["instance_id"],
        component_id=data["component_id"],
        x=data.get("x", 0),
        y=data.get("y", 0),
        config=data.get("config", {}),
    )

to_dict()

Source code in src/panel_flowdash/dashboard_store.py
def to_dict(self) -> dict[str, Any]:
    return {
        "instance_id": self.instance_id,
        "component_id": self.component_id,
        "x": self.x,
        "y": self.y,
        "config": self.config,
    }

DashboardModel dataclass

A persisted dashboard: nodes + edges + tile layout.

Source code in src/panel_flowdash/dashboard_store.py
@dataclass
class DashboardModel:
    """A persisted dashboard: nodes + edges + tile layout."""

    dashboard_id: str
    user_id: str
    title: str
    version: int = 3
    items: list[DashboardItem] = field(default_factory=list)
    edges: list[DashboardEdge] = field(default_factory=list)
    tile_layout: list[dict[str, Any]] = field(default_factory=list)
    breakpoints: list[int] = field(default_factory=list)
    responsive_layouts: dict[str, list[dict[str, Any]]] = field(default_factory=dict)
    permission: Permission = field(default_factory=Permission)

    @property
    def owner(self) -> str:
        """The immutable owner principal (the creating user)."""
        return self.user_id

    def to_dict(self) -> dict[str, Any]:
        return {
            "version": self.version,
            "dashboard_id": self.dashboard_id,
            "user_id": self.user_id,
            "title": self.title,
            "items": [item.to_dict() for item in self.items],
            "edges": [edge.to_dict() for edge in self.edges],
            "tile_layout": self.tile_layout,
            "breakpoints": self.breakpoints,
            "responsive_layouts": self.responsive_layouts,
            "permission": self.permission.to_dict(),
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> DashboardModel:
        return cls(
            dashboard_id=data["dashboard_id"],
            user_id=data["user_id"],
            title=data["title"],
            version=data.get("version", 1),
            items=[DashboardItem.from_dict(i) for i in data.get("items", [])],
            edges=[DashboardEdge.from_dict(e) for e in data.get("edges", [])],
            tile_layout=data.get("tile_layout", []),
            breakpoints=data.get("breakpoints", []),
            responsive_layouts=data.get("responsive_layouts", {}),
            permission=Permission.from_dict(data.get("permission")),
        )

breakpoints = field(default_factory=list) class-attribute instance-attribute

dashboard_id instance-attribute

edges = field(default_factory=list) class-attribute instance-attribute

items = field(default_factory=list) class-attribute instance-attribute

owner property

The immutable owner principal (the creating user).

permission = field(default_factory=Permission) class-attribute instance-attribute

responsive_layouts = field(default_factory=dict) class-attribute instance-attribute

tile_layout = field(default_factory=list) class-attribute instance-attribute

title instance-attribute

user_id instance-attribute

version = 3 class-attribute instance-attribute

from_dict(data) classmethod

Source code in src/panel_flowdash/dashboard_store.py
@classmethod
def from_dict(cls, data: dict[str, Any]) -> DashboardModel:
    return cls(
        dashboard_id=data["dashboard_id"],
        user_id=data["user_id"],
        title=data["title"],
        version=data.get("version", 1),
        items=[DashboardItem.from_dict(i) for i in data.get("items", [])],
        edges=[DashboardEdge.from_dict(e) for e in data.get("edges", [])],
        tile_layout=data.get("tile_layout", []),
        breakpoints=data.get("breakpoints", []),
        responsive_layouts=data.get("responsive_layouts", {}),
        permission=Permission.from_dict(data.get("permission")),
    )

to_dict()

Source code in src/panel_flowdash/dashboard_store.py
def to_dict(self) -> dict[str, Any]:
    return {
        "version": self.version,
        "dashboard_id": self.dashboard_id,
        "user_id": self.user_id,
        "title": self.title,
        "items": [item.to_dict() for item in self.items],
        "edges": [edge.to_dict() for edge in self.edges],
        "tile_layout": self.tile_layout,
        "breakpoints": self.breakpoints,
        "responsive_layouts": self.responsive_layouts,
        "permission": self.permission.to_dict(),
    }

DashboardStore

SQLite-backed store for dashboard models.

Source code in src/panel_flowdash/dashboard_store.py
class DashboardStore:
    """SQLite-backed store for dashboard models."""

    def __init__(self, db_path: str | Path):
        self._db_path = str(db_path)
        self._init_db()

    @contextmanager
    def _get_conn(self):
        conn = sqlite3.connect(self._db_path)
        conn.execute("PRAGMA journal_mode=WAL")
        conn.row_factory = sqlite3.Row
        try:
            yield conn
            conn.commit()
        except Exception:
            conn.rollback()
            raise
        finally:
            conn.close()

    def _init_db(self):
        with self._get_conn() as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS dashboards (
                    dashboard_id TEXT PRIMARY KEY,
                    user_id TEXT NOT NULL,
                    title TEXT NOT NULL,
                    version INTEGER NOT NULL DEFAULT 1,
                    items_json TEXT NOT NULL DEFAULT '[]',
                    edges_json TEXT NOT NULL DEFAULT '[]',
                    tile_layout_json TEXT NOT NULL DEFAULT '[]',
                    created_at TEXT NOT NULL DEFAULT (datetime('now')),
                    updated_at TEXT NOT NULL DEFAULT (datetime('now'))
                )
            """)
            conn.execute("""
                CREATE INDEX IF NOT EXISTS idx_dashboards_user
                ON dashboards (user_id)
            """)
            migrations = [
                ("edges_json", "'[]'"),
                ("tile_layout_json", "'[]'"),
                ("breakpoints_json", "'[]'"),
                ("responsive_layouts_json", "'{}'"),
                ("permission_json", "'{}'"),
            ]
            for col, default in migrations:
                try:
                    conn.execute(
                        f"ALTER TABLE dashboards ADD COLUMN {col} TEXT NOT NULL DEFAULT {default}"
                    )
                except sqlite3.OperationalError:
                    pass

    def list_dashboards(self, user_id: str) -> list[DashboardModel]:
        with self._get_conn() as conn:
            rows = conn.execute(
                "SELECT * FROM dashboards WHERE user_id = ? ORDER BY updated_at DESC",
                (user_id,),
            ).fetchall()
        return [self._row_to_model(row) for row in rows]

    def title_exists(self, user_id: str, title: str, exclude_id: str | None = None) -> bool:
        """Check if a dashboard with the given title already exists for this user."""
        with self._get_conn() as conn:
            if exclude_id:
                row = conn.execute(
                    "SELECT 1 FROM dashboards WHERE user_id = ? AND title = ? AND dashboard_id != ? LIMIT 1",
                    (user_id, title, exclude_id),
                ).fetchone()
            else:
                row = conn.execute(
                    "SELECT 1 FROM dashboards WHERE user_id = ? AND title = ? LIMIT 1",
                    (user_id, title),
                ).fetchone()
        return row is not None

    def load_dashboard(self, user_id: str, dashboard_id: str) -> DashboardModel | None:
        with self._get_conn() as conn:
            row = conn.execute(
                "SELECT * FROM dashboards WHERE dashboard_id = ? AND user_id = ?",
                (dashboard_id, user_id),
            ).fetchone()
        if row is None:
            return None
        return self._row_to_model(row)

    def _load_any(self, dashboard_id: str) -> DashboardModel | None:
        """Load a dashboard by id regardless of owner (for access checks)."""
        with self._get_conn() as conn:
            row = conn.execute(
                "SELECT * FROM dashboards WHERE dashboard_id = ?",
                (dashboard_id,),
            ).fetchone()
        if row is None:
            return None
        return self._row_to_model(row)

    def list_accessible(
        self, identity: Identity, *, default_allow: bool = True
    ) -> list[DashboardModel]:
        """List dashboards the *identity* owns or has been granted access to.

        Owned dashboards sort first (both groups by recency), so a user's own
        dashboards stay visually grouped ahead of ones shared with them.
        """
        with self._get_conn() as conn:
            rows = conn.execute(
                "SELECT * FROM dashboards ORDER BY updated_at DESC",
            ).fetchall()
        owned: list[DashboardModel] = []
        shared: list[DashboardModel] = []
        for row in rows:
            model = self._row_to_model(row)
            if model.owner in identity.user_names:
                owned.append(model)
            elif is_authorized(
                model.permission,
                identity,
                default_allow=default_allow,
                owner=model.owner,
            ):
                shared.append(model)
        return owned + shared

    def load_for_access(
        self, identity: Identity, dashboard_id: str, *, default_allow: bool = True
    ) -> DashboardModel | None:
        """Load a dashboard if *identity* is authorized, else ``None``.

        Returns ``None`` both when the dashboard does not exist and when access
        is denied, so callers render a single "not found / denied" view.
        """
        model = self._load_any(dashboard_id)
        if model is None:
            return None
        if is_authorized(
            model.permission,
            identity,
            default_allow=default_allow,
            owner=model.owner,
        ):
            return model
        return None

    def find_by_id_or_title(self, ref: str) -> DashboardModel | None:
        """Resolve a dashboard by its id first, then by title.

        Titles are only unique per user, so a title match returns the most
        recently updated dashboard. Used to resolve the operator-configured
        home dashboard, which may be given as either an id or a title.
        """
        model = self._load_any(ref)
        if model is not None:
            return model
        with self._get_conn() as conn:
            row = conn.execute(
                "SELECT * FROM dashboards WHERE title = ? ORDER BY updated_at DESC LIMIT 1",
                (ref,),
            ).fetchone()
        if row is None:
            return None
        return self._row_to_model(row)

    def get_owner(self, dashboard_id: str) -> str | None:
        """Return the owner (user_id) of a dashboard, or ``None`` if missing."""
        model = self._load_any(dashboard_id)
        return model.owner if model else None

    def can_administer(
        self, identity: Identity, dashboard_id: str, admin_groups: frozenset[str] = frozenset()
    ) -> bool:
        """Whether *identity* may administer (edit/delete/share) the dashboard."""
        model = self._load_any(dashboard_id)
        if model is None:
            return False
        return can_administer(identity, model.owner, admin_groups)

    def set_permission(self, dashboard_id: str, permission: Permission) -> bool:
        """Persist a new permission set on a dashboard. Returns success."""
        with self._get_conn() as conn:
            cursor = conn.execute(
                "UPDATE dashboards SET permission_json = ?, updated_at = datetime('now') WHERE dashboard_id = ?",
                (json.dumps(permission.to_dict()), dashboard_id),
            )
        return cursor.rowcount > 0

    def save_dashboard(self, dashboard: DashboardModel) -> None:
        items_json = json.dumps([item.to_dict() for item in dashboard.items])
        edges_json = json.dumps([edge.to_dict() for edge in dashboard.edges])
        tile_layout_json = json.dumps(dashboard.tile_layout)
        breakpoints_json = json.dumps(dashboard.breakpoints)
        responsive_layouts_json = json.dumps(dashboard.responsive_layouts)
        permission_json = json.dumps(dashboard.permission.to_dict())
        with self._get_conn() as conn:
            conn.execute(
                """
                INSERT INTO dashboards (dashboard_id, user_id, title, version, items_json, edges_json, tile_layout_json, breakpoints_json, responsive_layouts_json, permission_json, updated_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now'))
                ON CONFLICT(dashboard_id) DO UPDATE SET
                    title = excluded.title,
                    version = excluded.version,
                    items_json = excluded.items_json,
                    edges_json = excluded.edges_json,
                    tile_layout_json = excluded.tile_layout_json,
                    breakpoints_json = excluded.breakpoints_json,
                    responsive_layouts_json = excluded.responsive_layouts_json,
                    permission_json = excluded.permission_json,
                    updated_at = datetime('now')
                """,
                (
                    dashboard.dashboard_id,
                    dashboard.user_id,
                    dashboard.title,
                    dashboard.version,
                    items_json,
                    edges_json,
                    tile_layout_json,
                    breakpoints_json,
                    responsive_layouts_json,
                    permission_json,
                ),
            )

    def delete_dashboard(self, user_id: str, dashboard_id: str) -> bool:
        with self._get_conn() as conn:
            cursor = conn.execute(
                "DELETE FROM dashboards WHERE dashboard_id = ? AND user_id = ?",
                (dashboard_id, user_id),
            )
        return cursor.rowcount > 0

    def create_dashboard(self, user_id: str, title: str) -> DashboardModel:
        dashboard = DashboardModel(
            dashboard_id=uuid.uuid4().hex[:12],
            user_id=user_id,
            title=title,
        )
        self.save_dashboard(dashboard)
        return dashboard

    def rename_dashboard(self, user_id: str, dashboard_id: str, new_title: str) -> bool:
        with self._get_conn() as conn:
            cursor = conn.execute(
                "UPDATE dashboards SET title = ?, updated_at = datetime('now') WHERE dashboard_id = ? AND user_id = ?",
                (new_title, dashboard_id, user_id),
            )
        return cursor.rowcount > 0

    def _row_to_model(self, row: sqlite3.Row) -> DashboardModel:
        items = json.loads(row["items_json"])
        keys = row.keys()
        edges_raw = row["edges_json"] if "edges_json" in keys else "[]"
        tile_layout_raw = row["tile_layout_json"] if "tile_layout_json" in keys else "[]"
        breakpoints_raw = row["breakpoints_json"] if "breakpoints_json" in keys else "[]"
        responsive_raw = (
            row["responsive_layouts_json"] if "responsive_layouts_json" in keys else "{}"
        )
        permission_raw = row["permission_json"] if "permission_json" in keys else "{}"
        edges = json.loads(edges_raw)
        tile_layout = json.loads(tile_layout_raw)
        breakpoints = json.loads(breakpoints_raw)
        responsive_layouts = json.loads(responsive_raw)
        permission = Permission.from_dict(json.loads(permission_raw))
        return DashboardModel(
            dashboard_id=row["dashboard_id"],
            user_id=row["user_id"],
            title=row["title"],
            version=row["version"],
            items=[DashboardItem.from_dict(i) for i in items],
            edges=[DashboardEdge.from_dict(e) for e in edges],
            tile_layout=tile_layout,
            breakpoints=breakpoints,
            responsive_layouts=responsive_layouts,
            permission=permission,
        )

can_administer(identity, dashboard_id, admin_groups=frozenset())

Whether identity may administer (edit/delete/share) the dashboard.

Source code in src/panel_flowdash/dashboard_store.py
def can_administer(
    self, identity: Identity, dashboard_id: str, admin_groups: frozenset[str] = frozenset()
) -> bool:
    """Whether *identity* may administer (edit/delete/share) the dashboard."""
    model = self._load_any(dashboard_id)
    if model is None:
        return False
    return can_administer(identity, model.owner, admin_groups)

create_dashboard(user_id, title)

Source code in src/panel_flowdash/dashboard_store.py
def create_dashboard(self, user_id: str, title: str) -> DashboardModel:
    dashboard = DashboardModel(
        dashboard_id=uuid.uuid4().hex[:12],
        user_id=user_id,
        title=title,
    )
    self.save_dashboard(dashboard)
    return dashboard

delete_dashboard(user_id, dashboard_id)

Source code in src/panel_flowdash/dashboard_store.py
def delete_dashboard(self, user_id: str, dashboard_id: str) -> bool:
    with self._get_conn() as conn:
        cursor = conn.execute(
            "DELETE FROM dashboards WHERE dashboard_id = ? AND user_id = ?",
            (dashboard_id, user_id),
        )
    return cursor.rowcount > 0

find_by_id_or_title(ref)

Resolve a dashboard by its id first, then by title.

Titles are only unique per user, so a title match returns the most recently updated dashboard. Used to resolve the operator-configured home dashboard, which may be given as either an id or a title.

Source code in src/panel_flowdash/dashboard_store.py
def find_by_id_or_title(self, ref: str) -> DashboardModel | None:
    """Resolve a dashboard by its id first, then by title.

    Titles are only unique per user, so a title match returns the most
    recently updated dashboard. Used to resolve the operator-configured
    home dashboard, which may be given as either an id or a title.
    """
    model = self._load_any(ref)
    if model is not None:
        return model
    with self._get_conn() as conn:
        row = conn.execute(
            "SELECT * FROM dashboards WHERE title = ? ORDER BY updated_at DESC LIMIT 1",
            (ref,),
        ).fetchone()
    if row is None:
        return None
    return self._row_to_model(row)

get_owner(dashboard_id)

Return the owner (user_id) of a dashboard, or None if missing.

Source code in src/panel_flowdash/dashboard_store.py
def get_owner(self, dashboard_id: str) -> str | None:
    """Return the owner (user_id) of a dashboard, or ``None`` if missing."""
    model = self._load_any(dashboard_id)
    return model.owner if model else None

list_accessible(identity, *, default_allow=True)

List dashboards the identity owns or has been granted access to.

Owned dashboards sort first (both groups by recency), so a user's own dashboards stay visually grouped ahead of ones shared with them.

Source code in src/panel_flowdash/dashboard_store.py
def list_accessible(
    self, identity: Identity, *, default_allow: bool = True
) -> list[DashboardModel]:
    """List dashboards the *identity* owns or has been granted access to.

    Owned dashboards sort first (both groups by recency), so a user's own
    dashboards stay visually grouped ahead of ones shared with them.
    """
    with self._get_conn() as conn:
        rows = conn.execute(
            "SELECT * FROM dashboards ORDER BY updated_at DESC",
        ).fetchall()
    owned: list[DashboardModel] = []
    shared: list[DashboardModel] = []
    for row in rows:
        model = self._row_to_model(row)
        if model.owner in identity.user_names:
            owned.append(model)
        elif is_authorized(
            model.permission,
            identity,
            default_allow=default_allow,
            owner=model.owner,
        ):
            shared.append(model)
    return owned + shared

list_dashboards(user_id)

Source code in src/panel_flowdash/dashboard_store.py
def list_dashboards(self, user_id: str) -> list[DashboardModel]:
    with self._get_conn() as conn:
        rows = conn.execute(
            "SELECT * FROM dashboards WHERE user_id = ? ORDER BY updated_at DESC",
            (user_id,),
        ).fetchall()
    return [self._row_to_model(row) for row in rows]

load_dashboard(user_id, dashboard_id)

Source code in src/panel_flowdash/dashboard_store.py
def load_dashboard(self, user_id: str, dashboard_id: str) -> DashboardModel | None:
    with self._get_conn() as conn:
        row = conn.execute(
            "SELECT * FROM dashboards WHERE dashboard_id = ? AND user_id = ?",
            (dashboard_id, user_id),
        ).fetchone()
    if row is None:
        return None
    return self._row_to_model(row)

load_for_access(identity, dashboard_id, *, default_allow=True)

Load a dashboard if identity is authorized, else None.

Returns None both when the dashboard does not exist and when access is denied, so callers render a single "not found / denied" view.

Source code in src/panel_flowdash/dashboard_store.py
def load_for_access(
    self, identity: Identity, dashboard_id: str, *, default_allow: bool = True
) -> DashboardModel | None:
    """Load a dashboard if *identity* is authorized, else ``None``.

    Returns ``None`` both when the dashboard does not exist and when access
    is denied, so callers render a single "not found / denied" view.
    """
    model = self._load_any(dashboard_id)
    if model is None:
        return None
    if is_authorized(
        model.permission,
        identity,
        default_allow=default_allow,
        owner=model.owner,
    ):
        return model
    return None

rename_dashboard(user_id, dashboard_id, new_title)

Source code in src/panel_flowdash/dashboard_store.py
def rename_dashboard(self, user_id: str, dashboard_id: str, new_title: str) -> bool:
    with self._get_conn() as conn:
        cursor = conn.execute(
            "UPDATE dashboards SET title = ?, updated_at = datetime('now') WHERE dashboard_id = ? AND user_id = ?",
            (new_title, dashboard_id, user_id),
        )
    return cursor.rowcount > 0

save_dashboard(dashboard)

Source code in src/panel_flowdash/dashboard_store.py
def save_dashboard(self, dashboard: DashboardModel) -> None:
    items_json = json.dumps([item.to_dict() for item in dashboard.items])
    edges_json = json.dumps([edge.to_dict() for edge in dashboard.edges])
    tile_layout_json = json.dumps(dashboard.tile_layout)
    breakpoints_json = json.dumps(dashboard.breakpoints)
    responsive_layouts_json = json.dumps(dashboard.responsive_layouts)
    permission_json = json.dumps(dashboard.permission.to_dict())
    with self._get_conn() as conn:
        conn.execute(
            """
            INSERT INTO dashboards (dashboard_id, user_id, title, version, items_json, edges_json, tile_layout_json, breakpoints_json, responsive_layouts_json, permission_json, updated_at)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now'))
            ON CONFLICT(dashboard_id) DO UPDATE SET
                title = excluded.title,
                version = excluded.version,
                items_json = excluded.items_json,
                edges_json = excluded.edges_json,
                tile_layout_json = excluded.tile_layout_json,
                breakpoints_json = excluded.breakpoints_json,
                responsive_layouts_json = excluded.responsive_layouts_json,
                permission_json = excluded.permission_json,
                updated_at = datetime('now')
            """,
            (
                dashboard.dashboard_id,
                dashboard.user_id,
                dashboard.title,
                dashboard.version,
                items_json,
                edges_json,
                tile_layout_json,
                breakpoints_json,
                responsive_layouts_json,
                permission_json,
            ),
        )

set_permission(dashboard_id, permission)

Persist a new permission set on a dashboard. Returns success.

Source code in src/panel_flowdash/dashboard_store.py
def set_permission(self, dashboard_id: str, permission: Permission) -> bool:
    """Persist a new permission set on a dashboard. Returns success."""
    with self._get_conn() as conn:
        cursor = conn.execute(
            "UPDATE dashboards SET permission_json = ?, updated_at = datetime('now') WHERE dashboard_id = ?",
            (json.dumps(permission.to_dict()), dashboard_id),
        )
    return cursor.rowcount > 0

title_exists(user_id, title, exclude_id=None)

Check if a dashboard with the given title already exists for this user.

Source code in src/panel_flowdash/dashboard_store.py
def title_exists(self, user_id: str, title: str, exclude_id: str | None = None) -> bool:
    """Check if a dashboard with the given title already exists for this user."""
    with self._get_conn() as conn:
        if exclude_id:
            row = conn.execute(
                "SELECT 1 FROM dashboards WHERE user_id = ? AND title = ? AND dashboard_id != ? LIMIT 1",
                (user_id, title, exclude_id),
            ).fetchone()
        else:
            row = conn.execute(
                "SELECT 1 FROM dashboards WHERE user_id = ? AND title = ? LIMIT 1",
                (user_id, title),
            ).fetchone()
    return row is not None

Identity dataclass

The resolved principal for a session.

Source code in src/panel_flowdash/auth.py
@dataclass(frozen=True)
class Identity:
    """The resolved principal for a session."""

    user: str
    oauth_user: str | None = None
    system_user: str | None = None
    groups: frozenset[str] = field(default_factory=frozenset)
    user_info: dict[str, Any] = field(default_factory=dict)

    @property
    def user_names(self) -> frozenset[str]:
        """All names this identity may be referenced by in a rule."""
        names = {self.user}
        if self.oauth_user:
            names.add(self.oauth_user)
        if self.system_user:
            names.add(self.system_user)
        return frozenset(names)

    def in_groups(self, groups: Iterable[str]) -> bool:
        """Whether the identity belongs to any of *groups*."""
        return bool(self.groups & frozenset(groups))

    def is_user(self, users: Iterable[str]) -> bool:
        """Whether the identity matches any of *users* (OAuth or system name)."""
        return bool(self.user_names & frozenset(users))

groups = field(default_factory=frozenset) class-attribute instance-attribute

oauth_user = None class-attribute instance-attribute

system_user = None class-attribute instance-attribute

user instance-attribute

user_info = field(default_factory=dict) class-attribute instance-attribute

user_names property

All names this identity may be referenced by in a rule.

in_groups(groups)

Whether the identity belongs to any of groups.

Source code in src/panel_flowdash/auth.py
def in_groups(self, groups: Iterable[str]) -> bool:
    """Whether the identity belongs to any of *groups*."""
    return bool(self.groups & frozenset(groups))

is_user(users)

Whether the identity matches any of users (OAuth or system name).

Source code in src/panel_flowdash/auth.py
def is_user(self, users: Iterable[str]) -> bool:
    """Whether the identity matches any of *users* (OAuth or system name)."""
    return bool(self.user_names & frozenset(users))

Permission dataclass

An allow/deny rule set evaluated against an :class:Identity.

All four fields match either the resolved user (OAuth login or system user) or one of the identity's groups. An empty Permission declares no constraints and defers entirely to the caller's default policy.

Source code in src/panel_flowdash/auth.py
@dataclass(frozen=True)
class Permission:
    """An allow/deny rule set evaluated against an :class:`Identity`.

    All four fields match either the resolved ``user`` (OAuth login *or* system
    user) or one of the identity's ``groups``. An empty ``Permission`` declares
    no constraints and defers entirely to the caller's default policy.
    """

    allow_users: frozenset[str] = field(default_factory=frozenset)
    allow_groups: frozenset[str] = field(default_factory=frozenset)
    deny_users: frozenset[str] = field(default_factory=frozenset)
    deny_groups: frozenset[str] = field(default_factory=frozenset)

    @property
    def is_empty(self) -> bool:
        """Whether the permission declares no allow or deny rules."""
        return not (self.allow_users or self.allow_groups or self.deny_users or self.deny_groups)

    @classmethod
    def from_spec(
        cls,
        *,
        allow_users: Iterable[str] | None = None,
        allow_groups: Iterable[str] | None = None,
        deny_users: Iterable[str] | None = None,
        deny_groups: Iterable[str] | None = None,
    ) -> Permission:
        """Build a :class:`Permission` from loosely-typed iterables."""
        return cls(
            allow_users=frozenset(allow_users or ()),
            allow_groups=frozenset(allow_groups or ()),
            deny_users=frozenset(deny_users or ()),
            deny_groups=frozenset(deny_groups or ()),
        )

    def to_dict(self) -> dict[str, list[str]]:
        """Serialize to sorted lists for JSON persistence."""
        return {
            "allow_users": sorted(self.allow_users),
            "allow_groups": sorted(self.allow_groups),
            "deny_users": sorted(self.deny_users),
            "deny_groups": sorted(self.deny_groups),
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any] | None) -> Permission:
        """Deserialize from a (possibly ``None`` or partial) mapping."""
        data = data or {}
        return cls.from_spec(
            allow_users=data.get("allow_users"),
            allow_groups=data.get("allow_groups"),
            deny_users=data.get("deny_users"),
            deny_groups=data.get("deny_groups"),
        )

allow_groups = field(default_factory=frozenset) class-attribute instance-attribute

allow_users = field(default_factory=frozenset) class-attribute instance-attribute

deny_groups = field(default_factory=frozenset) class-attribute instance-attribute

deny_users = field(default_factory=frozenset) class-attribute instance-attribute

is_empty property

Whether the permission declares no allow or deny rules.

from_dict(data) classmethod

Deserialize from a (possibly None or partial) mapping.

Source code in src/panel_flowdash/auth.py
@classmethod
def from_dict(cls, data: dict[str, Any] | None) -> Permission:
    """Deserialize from a (possibly ``None`` or partial) mapping."""
    data = data or {}
    return cls.from_spec(
        allow_users=data.get("allow_users"),
        allow_groups=data.get("allow_groups"),
        deny_users=data.get("deny_users"),
        deny_groups=data.get("deny_groups"),
    )

from_spec(*, allow_users=None, allow_groups=None, deny_users=None, deny_groups=None) classmethod

Build a :class:Permission from loosely-typed iterables.

Source code in src/panel_flowdash/auth.py
@classmethod
def from_spec(
    cls,
    *,
    allow_users: Iterable[str] | None = None,
    allow_groups: Iterable[str] | None = None,
    deny_users: Iterable[str] | None = None,
    deny_groups: Iterable[str] | None = None,
) -> Permission:
    """Build a :class:`Permission` from loosely-typed iterables."""
    return cls(
        allow_users=frozenset(allow_users or ()),
        allow_groups=frozenset(allow_groups or ()),
        deny_users=frozenset(deny_users or ()),
        deny_groups=frozenset(deny_groups or ()),
    )

to_dict()

Serialize to sorted lists for JSON persistence.

Source code in src/panel_flowdash/auth.py
def to_dict(self) -> dict[str, list[str]]:
    """Serialize to sorted lists for JSON persistence."""
    return {
        "allow_users": sorted(self.allow_users),
        "allow_groups": sorted(self.allow_groups),
        "deny_users": sorted(self.deny_users),
        "deny_groups": sorted(self.deny_groups),
    }

can_administer(identity, owner, admin_groups)

Whether identity may administer a resource owned by owner.

When no admin_groups are configured administration is unrestricted: the running user (however resolved) may administer any resource. This keeps the default, auth-less deployment fully editable. Once admin_groups are set, only the owner and members of those groups may administer a resource.

Source code in src/panel_flowdash/auth.py
def can_administer(identity: Identity, owner: str, admin_groups: Iterable[str]) -> bool:
    """Whether *identity* may administer a resource owned by *owner*.

    When no *admin_groups* are configured administration is unrestricted: the
    running user (however resolved) may administer any resource. This keeps the
    default, auth-less deployment fully editable. Once *admin_groups* are set,
    only the owner and members of those groups may administer a resource.
    """
    admin_groups = frozenset(admin_groups)
    if not admin_groups:
        return True
    if owner in identity.user_names:
        return True
    return identity.in_groups(admin_groups)

is_authorized(permission, identity, *, default_allow=True, owner=None)

Evaluate permission against identity.

Order of precedence:

  1. A matching deny_users/deny_groups rule denies access (deny always wins, even for the owner).
  2. The owner, if given and matching, is allowed.
  3. Any allow_* rule present: allowed iff the identity matches at least one of them.
  4. No allow/deny rules at all: fall back to default_allow.
Source code in src/panel_flowdash/auth.py
def is_authorized(
    permission: Permission | None,
    identity: Identity,
    *,
    default_allow: bool = True,
    owner: str | None = None,
) -> bool:
    """Evaluate *permission* against *identity*.

    Order of precedence:

    1. A matching ``deny_users``/``deny_groups`` rule denies access (deny always
       wins, even for the owner).
    2. The *owner*, if given and matching, is allowed.
    3. Any ``allow_*`` rule present: allowed iff the identity matches at least
       one of them.
    4. No allow/deny rules at all: fall back to *default_allow*.
    """
    if permission is None:
        permission = Permission()

    if permission.deny_users and identity.is_user(permission.deny_users):
        return False
    if permission.deny_groups and identity.in_groups(permission.deny_groups):
        return False

    if owner is not None and owner in identity.user_names:
        return True

    if permission.allow_users or permission.allow_groups:
        if permission.allow_users and identity.is_user(permission.allow_users):
            return True
        if permission.allow_groups and identity.in_groups(permission.allow_groups):
            return True
        return False

    return default_allow

App Builder

panel_flowdash.app

Application builder: scans a project directory and constructs the Panel app.

COMPONENTS_ROUTE = '/components' module-attribute

DASH_ROUTE_PREFIX = '/dash/' module-attribute

_COMPONENT_PALETTE_CARD_CSS = '\n:host {\n cursor: pointer;\n border-radius: 6px;\n transition: background-color 0.15s;\n}\n:host(:hover) {\n background-color: rgba(0, 114, 181, 0.08);\n}\n' module-attribute

_LAUNCHER_CARD_CSS = '\n:host {\n cursor: pointer;\n transition: box-shadow 0.2s;\n}\n:host(:hover) {\n box-shadow: 0 4px 12px rgba(0,0,0,0.15);\n}\n:host .MuiCardContent-root {\n display: flex;\n align-items: center;\n justify-content: center;\n flex: 1;\n}\n' module-attribute

_LAUNCHER_DASH_CARD_CSS = '\n:host {\n cursor: pointer;\n transition: box-shadow 0.2s;\n overflow: visible;\n}\n:host(:hover) {\n box-shadow: 0 4px 12px rgba(0,0,0,0.15);\n}\n:host .MuiCardContent-root {\n display: flex;\n align-items: center;\n justify-content: center;\n flex: 1;\n}\n' module-attribute

_LAUNCHER_NEW_CARD_CSS = '\n:host {\n cursor: pointer;\n transition: box-shadow 0.2s, border-color 0.2s;\n}\n:host .MuiPaper-root {\n border: 1px dashed var(--mui-palette-divider, rgba(0,0,0,0.23));\n box-shadow: none;\n background: transparent;\n}\n:host(:hover) .MuiPaper-root {\n border-color: var(--mui-palette-primary-main, #0072b5);\n}\n:host .MuiCardContent-root {\n display: flex;\n align-items: center;\n justify-content: center;\n flex: 1;\n}\n' module-attribute

_LAUNCHER_SPEED_DIAL_CSS = '\n:host {\n position: absolute;\n top: 12px;\n right: 0px;\n z-index: 100;\n}\n:host .MuiSpeedDial-fab {\n width: 28px;\n height: 28px;\n min-height: unset;\n box-shadow: none;\n}\n' module-attribute

logger = logging.getLogger('panel_flowdash') module-attribute

AuthConfig dataclass

Project-level authorization configuration.

Loaded from the project's __init__.py by the serve command. All fields are optional; the defaults reproduce the pre-auth behavior (allow by default, groups read from the standard claim keys, no admin groups).

Source code in src/panel_flowdash/auth.py
@dataclass(frozen=True)
class AuthConfig:
    """Project-level authorization configuration.

    Loaded from the project's ``__init__.py`` by the serve command. All fields
    are optional; the defaults reproduce the pre-auth behavior (allow by
    default, groups read from the standard claim keys, no admin groups).
    """

    group_claims: tuple[str, ...] = DEFAULT_GROUP_CLAIMS
    user_groups: dict[str, frozenset[str]] = field(default_factory=dict)
    resolve_groups: Callable[[Identity], Iterable[str]] | None = None
    admin_groups: frozenset[str] = field(default_factory=frozenset)
    default_allow: bool = True

    @classmethod
    def from_module(cls, module: Any) -> AuthConfig:
        """Build an :class:`AuthConfig` from names on a project ``__init__``.

        Reads ``group_claims``, ``user_groups``, ``resolve_groups``,
        ``admin_groups`` and ``default_allow`` if present, falling back to the
        defaults otherwise. Missing module or names yield a default config.
        """
        if module is None:
            return cls()
        group_claims = getattr(module, "group_claims", None)
        raw_user_groups = getattr(module, "user_groups", None) or {}
        user_groups = {user: frozenset(groups) for user, groups in raw_user_groups.items()}
        admin_groups = getattr(module, "admin_groups", None) or ()
        default_allow = getattr(module, "default_allow", True)
        return cls(
            group_claims=tuple(group_claims) if group_claims else DEFAULT_GROUP_CLAIMS,
            user_groups=user_groups,
            resolve_groups=getattr(module, "resolve_groups", None),
            admin_groups=frozenset(admin_groups),
            default_allow=bool(default_allow),
        )

admin_groups = field(default_factory=frozenset) class-attribute instance-attribute

default_allow = True class-attribute instance-attribute

group_claims = DEFAULT_GROUP_CLAIMS class-attribute instance-attribute

resolve_groups = None class-attribute instance-attribute

user_groups = field(default_factory=dict) class-attribute instance-attribute

from_module(module) classmethod

Build an :class:AuthConfig from names on a project __init__.

Reads group_claims, user_groups, resolve_groups, admin_groups and default_allow if present, falling back to the defaults otherwise. Missing module or names yield a default config.

Source code in src/panel_flowdash/auth.py
@classmethod
def from_module(cls, module: Any) -> AuthConfig:
    """Build an :class:`AuthConfig` from names on a project ``__init__``.

    Reads ``group_claims``, ``user_groups``, ``resolve_groups``,
    ``admin_groups`` and ``default_allow`` if present, falling back to the
    defaults otherwise. Missing module or names yield a default config.
    """
    if module is None:
        return cls()
    group_claims = getattr(module, "group_claims", None)
    raw_user_groups = getattr(module, "user_groups", None) or {}
    user_groups = {user: frozenset(groups) for user, groups in raw_user_groups.items()}
    admin_groups = getattr(module, "admin_groups", None) or ()
    default_allow = getattr(module, "default_allow", True)
    return cls(
        group_claims=tuple(group_claims) if group_claims else DEFAULT_GROUP_CLAIMS,
        user_groups=user_groups,
        resolve_groups=getattr(module, "resolve_groups", None),
        admin_groups=frozenset(admin_groups),
        default_allow=bool(default_allow),
    )

DashboardEdge dataclass

A connection between two component ports.

Source code in src/panel_flowdash/dashboard_store.py
@dataclass
class DashboardEdge:
    """A connection between two component ports."""

    source: str
    source_port: str
    target: str
    target_port: str

    def to_dict(self) -> dict[str, str]:
        return {
            "source": self.source,
            "source_port": self.source_port,
            "target": self.target,
            "target_port": self.target_port,
        }

    @classmethod
    def from_dict(cls, data: dict[str, str]) -> DashboardEdge:
        return cls(
            source=data["source"],
            source_port=data["source_port"],
            target=data["target"],
            target_port=data["target_port"],
        )

source instance-attribute

source_port instance-attribute

target instance-attribute

target_port instance-attribute

from_dict(data) classmethod

Source code in src/panel_flowdash/dashboard_store.py
@classmethod
def from_dict(cls, data: dict[str, str]) -> DashboardEdge:
    return cls(
        source=data["source"],
        source_port=data["source_port"],
        target=data["target"],
        target_port=data["target_port"],
    )

to_dict()

Source code in src/panel_flowdash/dashboard_store.py
def to_dict(self) -> dict[str, str]:
    return {
        "source": self.source,
        "source_port": self.source_port,
        "target": self.target,
        "target_port": self.target_port,
    }

DashboardItem dataclass

A component instance on the dashboard.

x, y store the ReactFlow node canvas position. Grid layout (widths, heights, visibility) lives in DashboardModel.tile_layout.

Source code in src/panel_flowdash/dashboard_store.py
@dataclass
class DashboardItem:
    """A component instance on the dashboard.

    x, y store the ReactFlow node canvas position.
    Grid layout (widths, heights, visibility) lives in DashboardModel.tile_layout.
    """

    instance_id: str
    component_id: str
    x: float = 0
    y: float = 0
    config: dict[str, Any] = field(default_factory=dict)

    def to_dict(self) -> dict[str, Any]:
        return {
            "instance_id": self.instance_id,
            "component_id": self.component_id,
            "x": self.x,
            "y": self.y,
            "config": self.config,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> DashboardItem:
        return cls(
            instance_id=data["instance_id"],
            component_id=data["component_id"],
            x=data.get("x", 0),
            y=data.get("y", 0),
            config=data.get("config", {}),
        )

component_id instance-attribute

config = field(default_factory=dict) class-attribute instance-attribute

instance_id instance-attribute

x = 0 class-attribute instance-attribute

y = 0 class-attribute instance-attribute

from_dict(data) classmethod

Source code in src/panel_flowdash/dashboard_store.py
@classmethod
def from_dict(cls, data: dict[str, Any]) -> DashboardItem:
    return cls(
        instance_id=data["instance_id"],
        component_id=data["component_id"],
        x=data.get("x", 0),
        y=data.get("y", 0),
        config=data.get("config", {}),
    )

to_dict()

Source code in src/panel_flowdash/dashboard_store.py
def to_dict(self) -> dict[str, Any]:
    return {
        "instance_id": self.instance_id,
        "component_id": self.component_id,
        "x": self.x,
        "y": self.y,
        "config": self.config,
    }

DashboardModel dataclass

A persisted dashboard: nodes + edges + tile layout.

Source code in src/panel_flowdash/dashboard_store.py
@dataclass
class DashboardModel:
    """A persisted dashboard: nodes + edges + tile layout."""

    dashboard_id: str
    user_id: str
    title: str
    version: int = 3
    items: list[DashboardItem] = field(default_factory=list)
    edges: list[DashboardEdge] = field(default_factory=list)
    tile_layout: list[dict[str, Any]] = field(default_factory=list)
    breakpoints: list[int] = field(default_factory=list)
    responsive_layouts: dict[str, list[dict[str, Any]]] = field(default_factory=dict)
    permission: Permission = field(default_factory=Permission)

    @property
    def owner(self) -> str:
        """The immutable owner principal (the creating user)."""
        return self.user_id

    def to_dict(self) -> dict[str, Any]:
        return {
            "version": self.version,
            "dashboard_id": self.dashboard_id,
            "user_id": self.user_id,
            "title": self.title,
            "items": [item.to_dict() for item in self.items],
            "edges": [edge.to_dict() for edge in self.edges],
            "tile_layout": self.tile_layout,
            "breakpoints": self.breakpoints,
            "responsive_layouts": self.responsive_layouts,
            "permission": self.permission.to_dict(),
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> DashboardModel:
        return cls(
            dashboard_id=data["dashboard_id"],
            user_id=data["user_id"],
            title=data["title"],
            version=data.get("version", 1),
            items=[DashboardItem.from_dict(i) for i in data.get("items", [])],
            edges=[DashboardEdge.from_dict(e) for e in data.get("edges", [])],
            tile_layout=data.get("tile_layout", []),
            breakpoints=data.get("breakpoints", []),
            responsive_layouts=data.get("responsive_layouts", {}),
            permission=Permission.from_dict(data.get("permission")),
        )

breakpoints = field(default_factory=list) class-attribute instance-attribute

dashboard_id instance-attribute

edges = field(default_factory=list) class-attribute instance-attribute

items = field(default_factory=list) class-attribute instance-attribute

owner property

The immutable owner principal (the creating user).

permission = field(default_factory=Permission) class-attribute instance-attribute

responsive_layouts = field(default_factory=dict) class-attribute instance-attribute

tile_layout = field(default_factory=list) class-attribute instance-attribute

title instance-attribute

user_id instance-attribute

version = 3 class-attribute instance-attribute

from_dict(data) classmethod

Source code in src/panel_flowdash/dashboard_store.py
@classmethod
def from_dict(cls, data: dict[str, Any]) -> DashboardModel:
    return cls(
        dashboard_id=data["dashboard_id"],
        user_id=data["user_id"],
        title=data["title"],
        version=data.get("version", 1),
        items=[DashboardItem.from_dict(i) for i in data.get("items", [])],
        edges=[DashboardEdge.from_dict(e) for e in data.get("edges", [])],
        tile_layout=data.get("tile_layout", []),
        breakpoints=data.get("breakpoints", []),
        responsive_layouts=data.get("responsive_layouts", {}),
        permission=Permission.from_dict(data.get("permission")),
    )

to_dict()

Source code in src/panel_flowdash/dashboard_store.py
def to_dict(self) -> dict[str, Any]:
    return {
        "version": self.version,
        "dashboard_id": self.dashboard_id,
        "user_id": self.user_id,
        "title": self.title,
        "items": [item.to_dict() for item in self.items],
        "edges": [edge.to_dict() for edge in self.edges],
        "tile_layout": self.tile_layout,
        "breakpoints": self.breakpoints,
        "responsive_layouts": self.responsive_layouts,
        "permission": self.permission.to_dict(),
    }

DashboardStore

SQLite-backed store for dashboard models.

Source code in src/panel_flowdash/dashboard_store.py
class DashboardStore:
    """SQLite-backed store for dashboard models."""

    def __init__(self, db_path: str | Path):
        self._db_path = str(db_path)
        self._init_db()

    @contextmanager
    def _get_conn(self):
        conn = sqlite3.connect(self._db_path)
        conn.execute("PRAGMA journal_mode=WAL")
        conn.row_factory = sqlite3.Row
        try:
            yield conn
            conn.commit()
        except Exception:
            conn.rollback()
            raise
        finally:
            conn.close()

    def _init_db(self):
        with self._get_conn() as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS dashboards (
                    dashboard_id TEXT PRIMARY KEY,
                    user_id TEXT NOT NULL,
                    title TEXT NOT NULL,
                    version INTEGER NOT NULL DEFAULT 1,
                    items_json TEXT NOT NULL DEFAULT '[]',
                    edges_json TEXT NOT NULL DEFAULT '[]',
                    tile_layout_json TEXT NOT NULL DEFAULT '[]',
                    created_at TEXT NOT NULL DEFAULT (datetime('now')),
                    updated_at TEXT NOT NULL DEFAULT (datetime('now'))
                )
            """)
            conn.execute("""
                CREATE INDEX IF NOT EXISTS idx_dashboards_user
                ON dashboards (user_id)
            """)
            migrations = [
                ("edges_json", "'[]'"),
                ("tile_layout_json", "'[]'"),
                ("breakpoints_json", "'[]'"),
                ("responsive_layouts_json", "'{}'"),
                ("permission_json", "'{}'"),
            ]
            for col, default in migrations:
                try:
                    conn.execute(
                        f"ALTER TABLE dashboards ADD COLUMN {col} TEXT NOT NULL DEFAULT {default}"
                    )
                except sqlite3.OperationalError:
                    pass

    def list_dashboards(self, user_id: str) -> list[DashboardModel]:
        with self._get_conn() as conn:
            rows = conn.execute(
                "SELECT * FROM dashboards WHERE user_id = ? ORDER BY updated_at DESC",
                (user_id,),
            ).fetchall()
        return [self._row_to_model(row) for row in rows]

    def title_exists(self, user_id: str, title: str, exclude_id: str | None = None) -> bool:
        """Check if a dashboard with the given title already exists for this user."""
        with self._get_conn() as conn:
            if exclude_id:
                row = conn.execute(
                    "SELECT 1 FROM dashboards WHERE user_id = ? AND title = ? AND dashboard_id != ? LIMIT 1",
                    (user_id, title, exclude_id),
                ).fetchone()
            else:
                row = conn.execute(
                    "SELECT 1 FROM dashboards WHERE user_id = ? AND title = ? LIMIT 1",
                    (user_id, title),
                ).fetchone()
        return row is not None

    def load_dashboard(self, user_id: str, dashboard_id: str) -> DashboardModel | None:
        with self._get_conn() as conn:
            row = conn.execute(
                "SELECT * FROM dashboards WHERE dashboard_id = ? AND user_id = ?",
                (dashboard_id, user_id),
            ).fetchone()
        if row is None:
            return None
        return self._row_to_model(row)

    def _load_any(self, dashboard_id: str) -> DashboardModel | None:
        """Load a dashboard by id regardless of owner (for access checks)."""
        with self._get_conn() as conn:
            row = conn.execute(
                "SELECT * FROM dashboards WHERE dashboard_id = ?",
                (dashboard_id,),
            ).fetchone()
        if row is None:
            return None
        return self._row_to_model(row)

    def list_accessible(
        self, identity: Identity, *, default_allow: bool = True
    ) -> list[DashboardModel]:
        """List dashboards the *identity* owns or has been granted access to.

        Owned dashboards sort first (both groups by recency), so a user's own
        dashboards stay visually grouped ahead of ones shared with them.
        """
        with self._get_conn() as conn:
            rows = conn.execute(
                "SELECT * FROM dashboards ORDER BY updated_at DESC",
            ).fetchall()
        owned: list[DashboardModel] = []
        shared: list[DashboardModel] = []
        for row in rows:
            model = self._row_to_model(row)
            if model.owner in identity.user_names:
                owned.append(model)
            elif is_authorized(
                model.permission,
                identity,
                default_allow=default_allow,
                owner=model.owner,
            ):
                shared.append(model)
        return owned + shared

    def load_for_access(
        self, identity: Identity, dashboard_id: str, *, default_allow: bool = True
    ) -> DashboardModel | None:
        """Load a dashboard if *identity* is authorized, else ``None``.

        Returns ``None`` both when the dashboard does not exist and when access
        is denied, so callers render a single "not found / denied" view.
        """
        model = self._load_any(dashboard_id)
        if model is None:
            return None
        if is_authorized(
            model.permission,
            identity,
            default_allow=default_allow,
            owner=model.owner,
        ):
            return model
        return None

    def find_by_id_or_title(self, ref: str) -> DashboardModel | None:
        """Resolve a dashboard by its id first, then by title.

        Titles are only unique per user, so a title match returns the most
        recently updated dashboard. Used to resolve the operator-configured
        home dashboard, which may be given as either an id or a title.
        """
        model = self._load_any(ref)
        if model is not None:
            return model
        with self._get_conn() as conn:
            row = conn.execute(
                "SELECT * FROM dashboards WHERE title = ? ORDER BY updated_at DESC LIMIT 1",
                (ref,),
            ).fetchone()
        if row is None:
            return None
        return self._row_to_model(row)

    def get_owner(self, dashboard_id: str) -> str | None:
        """Return the owner (user_id) of a dashboard, or ``None`` if missing."""
        model = self._load_any(dashboard_id)
        return model.owner if model else None

    def can_administer(
        self, identity: Identity, dashboard_id: str, admin_groups: frozenset[str] = frozenset()
    ) -> bool:
        """Whether *identity* may administer (edit/delete/share) the dashboard."""
        model = self._load_any(dashboard_id)
        if model is None:
            return False
        return can_administer(identity, model.owner, admin_groups)

    def set_permission(self, dashboard_id: str, permission: Permission) -> bool:
        """Persist a new permission set on a dashboard. Returns success."""
        with self._get_conn() as conn:
            cursor = conn.execute(
                "UPDATE dashboards SET permission_json = ?, updated_at = datetime('now') WHERE dashboard_id = ?",
                (json.dumps(permission.to_dict()), dashboard_id),
            )
        return cursor.rowcount > 0

    def save_dashboard(self, dashboard: DashboardModel) -> None:
        items_json = json.dumps([item.to_dict() for item in dashboard.items])
        edges_json = json.dumps([edge.to_dict() for edge in dashboard.edges])
        tile_layout_json = json.dumps(dashboard.tile_layout)
        breakpoints_json = json.dumps(dashboard.breakpoints)
        responsive_layouts_json = json.dumps(dashboard.responsive_layouts)
        permission_json = json.dumps(dashboard.permission.to_dict())
        with self._get_conn() as conn:
            conn.execute(
                """
                INSERT INTO dashboards (dashboard_id, user_id, title, version, items_json, edges_json, tile_layout_json, breakpoints_json, responsive_layouts_json, permission_json, updated_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now'))
                ON CONFLICT(dashboard_id) DO UPDATE SET
                    title = excluded.title,
                    version = excluded.version,
                    items_json = excluded.items_json,
                    edges_json = excluded.edges_json,
                    tile_layout_json = excluded.tile_layout_json,
                    breakpoints_json = excluded.breakpoints_json,
                    responsive_layouts_json = excluded.responsive_layouts_json,
                    permission_json = excluded.permission_json,
                    updated_at = datetime('now')
                """,
                (
                    dashboard.dashboard_id,
                    dashboard.user_id,
                    dashboard.title,
                    dashboard.version,
                    items_json,
                    edges_json,
                    tile_layout_json,
                    breakpoints_json,
                    responsive_layouts_json,
                    permission_json,
                ),
            )

    def delete_dashboard(self, user_id: str, dashboard_id: str) -> bool:
        with self._get_conn() as conn:
            cursor = conn.execute(
                "DELETE FROM dashboards WHERE dashboard_id = ? AND user_id = ?",
                (dashboard_id, user_id),
            )
        return cursor.rowcount > 0

    def create_dashboard(self, user_id: str, title: str) -> DashboardModel:
        dashboard = DashboardModel(
            dashboard_id=uuid.uuid4().hex[:12],
            user_id=user_id,
            title=title,
        )
        self.save_dashboard(dashboard)
        return dashboard

    def rename_dashboard(self, user_id: str, dashboard_id: str, new_title: str) -> bool:
        with self._get_conn() as conn:
            cursor = conn.execute(
                "UPDATE dashboards SET title = ?, updated_at = datetime('now') WHERE dashboard_id = ? AND user_id = ?",
                (new_title, dashboard_id, user_id),
            )
        return cursor.rowcount > 0

    def _row_to_model(self, row: sqlite3.Row) -> DashboardModel:
        items = json.loads(row["items_json"])
        keys = row.keys()
        edges_raw = row["edges_json"] if "edges_json" in keys else "[]"
        tile_layout_raw = row["tile_layout_json"] if "tile_layout_json" in keys else "[]"
        breakpoints_raw = row["breakpoints_json"] if "breakpoints_json" in keys else "[]"
        responsive_raw = (
            row["responsive_layouts_json"] if "responsive_layouts_json" in keys else "{}"
        )
        permission_raw = row["permission_json"] if "permission_json" in keys else "{}"
        edges = json.loads(edges_raw)
        tile_layout = json.loads(tile_layout_raw)
        breakpoints = json.loads(breakpoints_raw)
        responsive_layouts = json.loads(responsive_raw)
        permission = Permission.from_dict(json.loads(permission_raw))
        return DashboardModel(
            dashboard_id=row["dashboard_id"],
            user_id=row["user_id"],
            title=row["title"],
            version=row["version"],
            items=[DashboardItem.from_dict(i) for i in items],
            edges=[DashboardEdge.from_dict(e) for e in edges],
            tile_layout=tile_layout,
            breakpoints=breakpoints,
            responsive_layouts=responsive_layouts,
            permission=permission,
        )

can_administer(identity, dashboard_id, admin_groups=frozenset())

Whether identity may administer (edit/delete/share) the dashboard.

Source code in src/panel_flowdash/dashboard_store.py
def can_administer(
    self, identity: Identity, dashboard_id: str, admin_groups: frozenset[str] = frozenset()
) -> bool:
    """Whether *identity* may administer (edit/delete/share) the dashboard."""
    model = self._load_any(dashboard_id)
    if model is None:
        return False
    return can_administer(identity, model.owner, admin_groups)

create_dashboard(user_id, title)

Source code in src/panel_flowdash/dashboard_store.py
def create_dashboard(self, user_id: str, title: str) -> DashboardModel:
    dashboard = DashboardModel(
        dashboard_id=uuid.uuid4().hex[:12],
        user_id=user_id,
        title=title,
    )
    self.save_dashboard(dashboard)
    return dashboard

delete_dashboard(user_id, dashboard_id)

Source code in src/panel_flowdash/dashboard_store.py
def delete_dashboard(self, user_id: str, dashboard_id: str) -> bool:
    with self._get_conn() as conn:
        cursor = conn.execute(
            "DELETE FROM dashboards WHERE dashboard_id = ? AND user_id = ?",
            (dashboard_id, user_id),
        )
    return cursor.rowcount > 0

find_by_id_or_title(ref)

Resolve a dashboard by its id first, then by title.

Titles are only unique per user, so a title match returns the most recently updated dashboard. Used to resolve the operator-configured home dashboard, which may be given as either an id or a title.

Source code in src/panel_flowdash/dashboard_store.py
def find_by_id_or_title(self, ref: str) -> DashboardModel | None:
    """Resolve a dashboard by its id first, then by title.

    Titles are only unique per user, so a title match returns the most
    recently updated dashboard. Used to resolve the operator-configured
    home dashboard, which may be given as either an id or a title.
    """
    model = self._load_any(ref)
    if model is not None:
        return model
    with self._get_conn() as conn:
        row = conn.execute(
            "SELECT * FROM dashboards WHERE title = ? ORDER BY updated_at DESC LIMIT 1",
            (ref,),
        ).fetchone()
    if row is None:
        return None
    return self._row_to_model(row)

get_owner(dashboard_id)

Return the owner (user_id) of a dashboard, or None if missing.

Source code in src/panel_flowdash/dashboard_store.py
def get_owner(self, dashboard_id: str) -> str | None:
    """Return the owner (user_id) of a dashboard, or ``None`` if missing."""
    model = self._load_any(dashboard_id)
    return model.owner if model else None

list_accessible(identity, *, default_allow=True)

List dashboards the identity owns or has been granted access to.

Owned dashboards sort first (both groups by recency), so a user's own dashboards stay visually grouped ahead of ones shared with them.

Source code in src/panel_flowdash/dashboard_store.py
def list_accessible(
    self, identity: Identity, *, default_allow: bool = True
) -> list[DashboardModel]:
    """List dashboards the *identity* owns or has been granted access to.

    Owned dashboards sort first (both groups by recency), so a user's own
    dashboards stay visually grouped ahead of ones shared with them.
    """
    with self._get_conn() as conn:
        rows = conn.execute(
            "SELECT * FROM dashboards ORDER BY updated_at DESC",
        ).fetchall()
    owned: list[DashboardModel] = []
    shared: list[DashboardModel] = []
    for row in rows:
        model = self._row_to_model(row)
        if model.owner in identity.user_names:
            owned.append(model)
        elif is_authorized(
            model.permission,
            identity,
            default_allow=default_allow,
            owner=model.owner,
        ):
            shared.append(model)
    return owned + shared

list_dashboards(user_id)

Source code in src/panel_flowdash/dashboard_store.py
def list_dashboards(self, user_id: str) -> list[DashboardModel]:
    with self._get_conn() as conn:
        rows = conn.execute(
            "SELECT * FROM dashboards WHERE user_id = ? ORDER BY updated_at DESC",
            (user_id,),
        ).fetchall()
    return [self._row_to_model(row) for row in rows]

load_dashboard(user_id, dashboard_id)

Source code in src/panel_flowdash/dashboard_store.py
def load_dashboard(self, user_id: str, dashboard_id: str) -> DashboardModel | None:
    with self._get_conn() as conn:
        row = conn.execute(
            "SELECT * FROM dashboards WHERE dashboard_id = ? AND user_id = ?",
            (dashboard_id, user_id),
        ).fetchone()
    if row is None:
        return None
    return self._row_to_model(row)

load_for_access(identity, dashboard_id, *, default_allow=True)

Load a dashboard if identity is authorized, else None.

Returns None both when the dashboard does not exist and when access is denied, so callers render a single "not found / denied" view.

Source code in src/panel_flowdash/dashboard_store.py
def load_for_access(
    self, identity: Identity, dashboard_id: str, *, default_allow: bool = True
) -> DashboardModel | None:
    """Load a dashboard if *identity* is authorized, else ``None``.

    Returns ``None`` both when the dashboard does not exist and when access
    is denied, so callers render a single "not found / denied" view.
    """
    model = self._load_any(dashboard_id)
    if model is None:
        return None
    if is_authorized(
        model.permission,
        identity,
        default_allow=default_allow,
        owner=model.owner,
    ):
        return model
    return None

rename_dashboard(user_id, dashboard_id, new_title)

Source code in src/panel_flowdash/dashboard_store.py
def rename_dashboard(self, user_id: str, dashboard_id: str, new_title: str) -> bool:
    with self._get_conn() as conn:
        cursor = conn.execute(
            "UPDATE dashboards SET title = ?, updated_at = datetime('now') WHERE dashboard_id = ? AND user_id = ?",
            (new_title, dashboard_id, user_id),
        )
    return cursor.rowcount > 0

save_dashboard(dashboard)

Source code in src/panel_flowdash/dashboard_store.py
def save_dashboard(self, dashboard: DashboardModel) -> None:
    items_json = json.dumps([item.to_dict() for item in dashboard.items])
    edges_json = json.dumps([edge.to_dict() for edge in dashboard.edges])
    tile_layout_json = json.dumps(dashboard.tile_layout)
    breakpoints_json = json.dumps(dashboard.breakpoints)
    responsive_layouts_json = json.dumps(dashboard.responsive_layouts)
    permission_json = json.dumps(dashboard.permission.to_dict())
    with self._get_conn() as conn:
        conn.execute(
            """
            INSERT INTO dashboards (dashboard_id, user_id, title, version, items_json, edges_json, tile_layout_json, breakpoints_json, responsive_layouts_json, permission_json, updated_at)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now'))
            ON CONFLICT(dashboard_id) DO UPDATE SET
                title = excluded.title,
                version = excluded.version,
                items_json = excluded.items_json,
                edges_json = excluded.edges_json,
                tile_layout_json = excluded.tile_layout_json,
                breakpoints_json = excluded.breakpoints_json,
                responsive_layouts_json = excluded.responsive_layouts_json,
                permission_json = excluded.permission_json,
                updated_at = datetime('now')
            """,
            (
                dashboard.dashboard_id,
                dashboard.user_id,
                dashboard.title,
                dashboard.version,
                items_json,
                edges_json,
                tile_layout_json,
                breakpoints_json,
                responsive_layouts_json,
                permission_json,
            ),
        )

set_permission(dashboard_id, permission)

Persist a new permission set on a dashboard. Returns success.

Source code in src/panel_flowdash/dashboard_store.py
def set_permission(self, dashboard_id: str, permission: Permission) -> bool:
    """Persist a new permission set on a dashboard. Returns success."""
    with self._get_conn() as conn:
        cursor = conn.execute(
            "UPDATE dashboards SET permission_json = ?, updated_at = datetime('now') WHERE dashboard_id = ?",
            (json.dumps(permission.to_dict()), dashboard_id),
        )
    return cursor.rowcount > 0

title_exists(user_id, title, exclude_id=None)

Check if a dashboard with the given title already exists for this user.

Source code in src/panel_flowdash/dashboard_store.py
def title_exists(self, user_id: str, title: str, exclude_id: str | None = None) -> bool:
    """Check if a dashboard with the given title already exists for this user."""
    with self._get_conn() as conn:
        if exclude_id:
            row = conn.execute(
                "SELECT 1 FROM dashboards WHERE user_id = ? AND title = ? AND dashboard_id != ? LIMIT 1",
                (user_id, title, exclude_id),
            ).fetchone()
        else:
            row = conn.execute(
                "SELECT 1 FROM dashboards WHERE user_id = ? AND title = ? LIMIT 1",
                (user_id, title),
            ).fetchone()
    return row is not None

DataflowGraph

Manages node state instances and wires edges with runtime validation.

Source code in src/panel_flowdash/dataflow_engine.py
class DataflowGraph:
    """Manages node state instances and wires edges with runtime validation."""

    def __init__(
        self,
        specs: dict[str, ComponentSpec],
        on_error: Callable[[str, str, str, str, Exception], None] | None = None,
    ):
        """Initialize the dataflow graph.

        Parameters
        ----------
        specs
            Component specs keyed by component_id.
        on_error
            Callback invoked when a runtime value assignment fails.
            Signature: (source_id, source_port, target_id, target_port, exception)
        """
        self._specs = specs
        self._state_classes: dict[str, type[param.Parameterized]] = {}
        self._nodes: dict[str, param.Parameterized] = {}
        self._config_states: dict[str, param.Parameterized] = {}
        self._node_specs: dict[str, ComponentSpec] = {}
        self._edges: list[dict[str, str]] = []
        self._watchers: dict[tuple, param.parameterized.Watcher] = {}
        self._on_error = on_error

        for comp_id, spec in specs.items():
            self._state_classes[comp_id] = build_node_state_class(spec)

    def add_node(self, instance_id: str, component_id: str) -> param.Parameterized:
        """Create a new node state instance."""
        spec = self._specs[component_id]
        cls = self._state_classes[component_id]
        state = cls(name=instance_id)
        self._nodes[instance_id] = state
        self._node_specs[instance_id] = spec
        if spec.config_state_class is not None:
            self._config_states[instance_id] = spec.config_state_class(
                name=f"{instance_id}_config"
            )
        return state

    def remove_node(self, instance_id: str):
        """Remove a node and any edges connected to it."""
        keys_to_remove = [k for k in self._watchers if k[0] == instance_id or k[2] == instance_id]
        for key in keys_to_remove:
            watcher = self._watchers.pop(key)
            src = self._nodes.get(key[0])
            if src is not None:
                src.param.unwatch(watcher)

        self._edges = [
            e for e in self._edges if e["source"] != instance_id and e["target"] != instance_id
        ]
        self._nodes.pop(instance_id, None)
        self._config_states.pop(instance_id, None)
        self._node_specs.pop(instance_id, None)

    def _is_list_input(self, target_id: str, target_port: str) -> bool:
        """Return True if the target port is a List-typed multi-connection input."""
        spec = self._node_specs.get(target_id)
        if not spec:
            return False
        for port in spec.inputs:
            if port.name == target_port:
                return _is_list_port(port)
        return False

    def _rebuild_list_port(self, target_id: str, target_port: str):
        """Recompute the list value for a multi-connection port from all sources."""
        target_state = self._nodes.get(target_id)
        if target_state is None:
            return
        values = []
        for e in self._edges:
            if e["target"] == target_id and e["target_port"] == target_port:
                src_state = self._nodes.get(e["source"])
                if src_state is not None:
                    val = getattr(src_state, e["source_port"], None)
                    if val is not None:
                        values.append(val)
        try:
            setattr(target_state, target_port, values)
        except Exception as exc:
            if self._on_error:
                self._on_error("", "", target_id, target_port, exc)

    def add_edge(
        self,
        source_id: str,
        source_port: str,
        target_id: str,
        target_port: str,
    ) -> bool | str:
        """Wire an edge between two ports.

        Returns True on success, or an error message string on failure.
        """
        source_state = self._nodes.get(source_id)
        target_state = self._nodes.get(target_id)
        if source_state is None or target_state is None:
            return "Source or target node not found."
        if not hasattr(source_state.param, source_port):
            return f"Output port '{source_port}' does not exist on source node."
        if not hasattr(target_state.param, target_port):
            return f"Input port '{target_port}' does not exist on target node."

        is_list = self._is_list_input(target_id, target_port)

        if not is_list:
            for e in self._edges:
                if e["target"] == target_id and e["target_port"] == target_port:
                    return f"Input '{target_port}' already has a connection. Disconnect it first."

        if self._would_create_cycle(source_id, target_id):
            return "Connection rejected: would create a cycle."

        source_spec = self._node_specs.get(source_id)
        target_spec = self._node_specs.get(target_id)
        if source_spec and target_spec and not is_list:
            error = self._check_type_compatibility(
                source_spec, source_port, target_spec, target_port
            )
            if error:
                return error

        if is_list:

            def _propagate(
                event,
                _src_id=source_id,
                _src_port=source_port,
                _tgt_id=target_id,
                _tgt_port=target_port,
            ):
                try:
                    self._rebuild_list_port(_tgt_id, _tgt_port)
                except Exception as exc:
                    if self._on_error:
                        self._on_error(_src_id, _src_port, _tgt_id, _tgt_port, exc)
        else:

            def _propagate(
                event,
                _src_id=source_id,
                _src_port=source_port,
                _tgt_id=target_id,
                _tgt_port=target_port,
                _target=target_state,
            ):
                try:
                    setattr(_target, _tgt_port, event.new)
                except Exception as exc:
                    if self._on_error:
                        self._on_error(_src_id, _src_port, _tgt_id, _tgt_port, exc)

        watcher = source_state.param.watch(_propagate, source_port)
        edge_key = (source_id, source_port, target_id, target_port)
        self._watchers[edge_key] = watcher

        self._edges.append(
            {
                "source": source_id,
                "source_port": source_port,
                "target": target_id,
                "target_port": target_port,
            }
        )

        if is_list:
            self._rebuild_list_port(target_id, target_port)
        else:
            current = getattr(source_state, source_port)
            if current is not None:
                try:
                    setattr(target_state, target_port, current)
                except Exception as exc:
                    if self._on_error:
                        self._on_error(source_id, source_port, target_id, target_port, exc)

        return True

    def _would_create_cycle(self, source_id: str, target_id: str) -> bool:
        """Return True if adding an edge from source to target would create a cycle."""
        if source_id == target_id:
            return True
        visited = set()
        queue = [target_id]
        while queue:
            node = queue.pop(0)
            if node == source_id:
                return True
            if node in visited:
                continue
            visited.add(node)
            for e in self._edges:
                if e["source"] == node:
                    queue.append(e["target"])
        return False

    def _check_type_compatibility(
        self,
        source_spec: ComponentSpec,
        source_port: str,
        target_spec: ComponentSpec,
        target_port: str,
    ) -> str | None:
        """Return an error message if types are incompatible, None if OK."""
        source_type = None
        for port in source_spec.outputs:
            if port.name == source_port:
                source_type = port.type
                break

        target_type = None
        for port in target_spec.inputs:
            if port.name == target_port:
                target_type = port.type
                break

        if source_type is None or target_type is None:
            return None

        if source_type.lower() == target_type.lower():
            return None

        return (
            f"Type mismatch: output '{source_port}' produces '{source_type}' "
            f"but input '{target_port}' expects '{target_type}'."
        )

    def remove_edge(self, source_id: str, source_port: str, target_id: str, target_port: str):
        """Remove an edge, unsubscribe the watcher, and reset target to default."""
        is_list = self._is_list_input(target_id, target_port)

        self._edges = [
            e
            for e in self._edges
            if not (
                e["source"] == source_id
                and e["source_port"] == source_port
                and e["target"] == target_id
                and e["target_port"] == target_port
            )
        ]

        edge_key = (source_id, source_port, target_id, target_port)
        watcher = self._watchers.pop(edge_key, None)
        if watcher is not None:
            source_state = self._nodes.get(source_id)
            if source_state is not None:
                source_state.param.unwatch(watcher)

        if is_list:
            self._rebuild_list_port(target_id, target_port)
        else:
            target_state = self._nodes.get(target_id)
            if target_state is not None and hasattr(target_state, target_port):
                spec = self._node_specs.get(target_id)
                default = None
                if spec:
                    for port in spec.inputs:
                        if port.name == target_port:
                            default = port.default
                            break
                setattr(target_state, target_port, default)

    def get_state(self, instance_id: str) -> param.Parameterized | None:
        """Get the state instance for a node."""
        return self._nodes.get(instance_id)

    def get_config_state(self, instance_id: str) -> param.Parameterized | None:
        """Get the config state instance for a node, if it has config."""
        return self._config_states.get(instance_id)

    @property
    def edges(self) -> list[dict[str, str]]:
        """All current edges."""
        return list(self._edges)

    @property
    def node_ids(self) -> list[str]:
        """All current node instance IDs."""
        return list(self._nodes.keys())

edges property

All current edges.

node_ids property

All current node instance IDs.

add_edge(source_id, source_port, target_id, target_port)

Wire an edge between two ports.

Returns True on success, or an error message string on failure.

Source code in src/panel_flowdash/dataflow_engine.py
def add_edge(
    self,
    source_id: str,
    source_port: str,
    target_id: str,
    target_port: str,
) -> bool | str:
    """Wire an edge between two ports.

    Returns True on success, or an error message string on failure.
    """
    source_state = self._nodes.get(source_id)
    target_state = self._nodes.get(target_id)
    if source_state is None or target_state is None:
        return "Source or target node not found."
    if not hasattr(source_state.param, source_port):
        return f"Output port '{source_port}' does not exist on source node."
    if not hasattr(target_state.param, target_port):
        return f"Input port '{target_port}' does not exist on target node."

    is_list = self._is_list_input(target_id, target_port)

    if not is_list:
        for e in self._edges:
            if e["target"] == target_id and e["target_port"] == target_port:
                return f"Input '{target_port}' already has a connection. Disconnect it first."

    if self._would_create_cycle(source_id, target_id):
        return "Connection rejected: would create a cycle."

    source_spec = self._node_specs.get(source_id)
    target_spec = self._node_specs.get(target_id)
    if source_spec and target_spec and not is_list:
        error = self._check_type_compatibility(
            source_spec, source_port, target_spec, target_port
        )
        if error:
            return error

    if is_list:

        def _propagate(
            event,
            _src_id=source_id,
            _src_port=source_port,
            _tgt_id=target_id,
            _tgt_port=target_port,
        ):
            try:
                self._rebuild_list_port(_tgt_id, _tgt_port)
            except Exception as exc:
                if self._on_error:
                    self._on_error(_src_id, _src_port, _tgt_id, _tgt_port, exc)
    else:

        def _propagate(
            event,
            _src_id=source_id,
            _src_port=source_port,
            _tgt_id=target_id,
            _tgt_port=target_port,
            _target=target_state,
        ):
            try:
                setattr(_target, _tgt_port, event.new)
            except Exception as exc:
                if self._on_error:
                    self._on_error(_src_id, _src_port, _tgt_id, _tgt_port, exc)

    watcher = source_state.param.watch(_propagate, source_port)
    edge_key = (source_id, source_port, target_id, target_port)
    self._watchers[edge_key] = watcher

    self._edges.append(
        {
            "source": source_id,
            "source_port": source_port,
            "target": target_id,
            "target_port": target_port,
        }
    )

    if is_list:
        self._rebuild_list_port(target_id, target_port)
    else:
        current = getattr(source_state, source_port)
        if current is not None:
            try:
                setattr(target_state, target_port, current)
            except Exception as exc:
                if self._on_error:
                    self._on_error(source_id, source_port, target_id, target_port, exc)

    return True

add_node(instance_id, component_id)

Create a new node state instance.

Source code in src/panel_flowdash/dataflow_engine.py
def add_node(self, instance_id: str, component_id: str) -> param.Parameterized:
    """Create a new node state instance."""
    spec = self._specs[component_id]
    cls = self._state_classes[component_id]
    state = cls(name=instance_id)
    self._nodes[instance_id] = state
    self._node_specs[instance_id] = spec
    if spec.config_state_class is not None:
        self._config_states[instance_id] = spec.config_state_class(
            name=f"{instance_id}_config"
        )
    return state

get_config_state(instance_id)

Get the config state instance for a node, if it has config.

Source code in src/panel_flowdash/dataflow_engine.py
def get_config_state(self, instance_id: str) -> param.Parameterized | None:
    """Get the config state instance for a node, if it has config."""
    return self._config_states.get(instance_id)

get_state(instance_id)

Get the state instance for a node.

Source code in src/panel_flowdash/dataflow_engine.py
def get_state(self, instance_id: str) -> param.Parameterized | None:
    """Get the state instance for a node."""
    return self._nodes.get(instance_id)

remove_edge(source_id, source_port, target_id, target_port)

Remove an edge, unsubscribe the watcher, and reset target to default.

Source code in src/panel_flowdash/dataflow_engine.py
def remove_edge(self, source_id: str, source_port: str, target_id: str, target_port: str):
    """Remove an edge, unsubscribe the watcher, and reset target to default."""
    is_list = self._is_list_input(target_id, target_port)

    self._edges = [
        e
        for e in self._edges
        if not (
            e["source"] == source_id
            and e["source_port"] == source_port
            and e["target"] == target_id
            and e["target_port"] == target_port
        )
    ]

    edge_key = (source_id, source_port, target_id, target_port)
    watcher = self._watchers.pop(edge_key, None)
    if watcher is not None:
        source_state = self._nodes.get(source_id)
        if source_state is not None:
            source_state.param.unwatch(watcher)

    if is_list:
        self._rebuild_list_port(target_id, target_port)
    else:
        target_state = self._nodes.get(target_id)
        if target_state is not None and hasattr(target_state, target_port):
            spec = self._node_specs.get(target_id)
            default = None
            if spec:
                for port in spec.inputs:
                    if port.name == target_port:
                        default = port.default
                        break
            setattr(target_state, target_port, default)

remove_node(instance_id)

Remove a node and any edges connected to it.

Source code in src/panel_flowdash/dataflow_engine.py
def remove_node(self, instance_id: str):
    """Remove a node and any edges connected to it."""
    keys_to_remove = [k for k in self._watchers if k[0] == instance_id or k[2] == instance_id]
    for key in keys_to_remove:
        watcher = self._watchers.pop(key)
        src = self._nodes.get(key[0])
        if src is not None:
            src.param.unwatch(watcher)

    self._edges = [
        e for e in self._edges if e["source"] != instance_id and e["target"] != instance_id
    ]
    self._nodes.pop(instance_id, None)
    self._config_states.pop(instance_id, None)
    self._node_specs.pop(instance_id, None)

FlowDashApp

Bases: Viewer

FlowDash application: scans a project directory and serves its pages and components.

Source code in src/panel_flowdash/app.py
 136
 137
 138
 139
 140
 141
 142
 143
 144
 145
 146
 147
 148
 149
 150
 151
 152
 153
 154
 155
 156
 157
 158
 159
 160
 161
 162
 163
 164
 165
 166
 167
 168
 169
 170
 171
 172
 173
 174
 175
 176
 177
 178
 179
 180
 181
 182
 183
 184
 185
 186
 187
 188
 189
 190
 191
 192
 193
 194
 195
 196
 197
 198
 199
 200
 201
 202
 203
 204
 205
 206
 207
 208
 209
 210
 211
 212
 213
 214
 215
 216
 217
 218
 219
 220
 221
 222
 223
 224
 225
 226
 227
 228
 229
 230
 231
 232
 233
 234
 235
 236
 237
 238
 239
 240
 241
 242
 243
 244
 245
 246
 247
 248
 249
 250
 251
 252
 253
 254
 255
 256
 257
 258
 259
 260
 261
 262
 263
 264
 265
 266
 267
 268
 269
 270
 271
 272
 273
 274
 275
 276
 277
 278
 279
 280
 281
 282
 283
 284
 285
 286
 287
 288
 289
 290
 291
 292
 293
 294
 295
 296
 297
 298
 299
 300
 301
 302
 303
 304
 305
 306
 307
 308
 309
 310
 311
 312
 313
 314
 315
 316
 317
 318
 319
 320
 321
 322
 323
 324
 325
 326
 327
 328
 329
 330
 331
 332
 333
 334
 335
 336
 337
 338
 339
 340
 341
 342
 343
 344
 345
 346
 347
 348
 349
 350
 351
 352
 353
 354
 355
 356
 357
 358
 359
 360
 361
 362
 363
 364
 365
 366
 367
 368
 369
 370
 371
 372
 373
 374
 375
 376
 377
 378
 379
 380
 381
 382
 383
 384
 385
 386
 387
 388
 389
 390
 391
 392
 393
 394
 395
 396
 397
 398
 399
 400
 401
 402
 403
 404
 405
 406
 407
 408
 409
 410
 411
 412
 413
 414
 415
 416
 417
 418
 419
 420
 421
 422
 423
 424
 425
 426
 427
 428
 429
 430
 431
 432
 433
 434
 435
 436
 437
 438
 439
 440
 441
 442
 443
 444
 445
 446
 447
 448
 449
 450
 451
 452
 453
 454
 455
 456
 457
 458
 459
 460
 461
 462
 463
 464
 465
 466
 467
 468
 469
 470
 471
 472
 473
 474
 475
 476
 477
 478
 479
 480
 481
 482
 483
 484
 485
 486
 487
 488
 489
 490
 491
 492
 493
 494
 495
 496
 497
 498
 499
 500
 501
 502
 503
 504
 505
 506
 507
 508
 509
 510
 511
 512
 513
 514
 515
 516
 517
 518
 519
 520
 521
 522
 523
 524
 525
 526
 527
 528
 529
 530
 531
 532
 533
 534
 535
 536
 537
 538
 539
 540
 541
 542
 543
 544
 545
 546
 547
 548
 549
 550
 551
 552
 553
 554
 555
 556
 557
 558
 559
 560
 561
 562
 563
 564
 565
 566
 567
 568
 569
 570
 571
 572
 573
 574
 575
 576
 577
 578
 579
 580
 581
 582
 583
 584
 585
 586
 587
 588
 589
 590
 591
 592
 593
 594
 595
 596
 597
 598
 599
 600
 601
 602
 603
 604
 605
 606
 607
 608
 609
 610
 611
 612
 613
 614
 615
 616
 617
 618
 619
 620
 621
 622
 623
 624
 625
 626
 627
 628
 629
 630
 631
 632
 633
 634
 635
 636
 637
 638
 639
 640
 641
 642
 643
 644
 645
 646
 647
 648
 649
 650
 651
 652
 653
 654
 655
 656
 657
 658
 659
 660
 661
 662
 663
 664
 665
 666
 667
 668
 669
 670
 671
 672
 673
 674
 675
 676
 677
 678
 679
 680
 681
 682
 683
 684
 685
 686
 687
 688
 689
 690
 691
 692
 693
 694
 695
 696
 697
 698
 699
 700
 701
 702
 703
 704
 705
 706
 707
 708
 709
 710
 711
 712
 713
 714
 715
 716
 717
 718
 719
 720
 721
 722
 723
 724
 725
 726
 727
 728
 729
 730
 731
 732
 733
 734
 735
 736
 737
 738
 739
 740
 741
 742
 743
 744
 745
 746
 747
 748
 749
 750
 751
 752
 753
 754
 755
 756
 757
 758
 759
 760
 761
 762
 763
 764
 765
 766
 767
 768
 769
 770
 771
 772
 773
 774
 775
 776
 777
 778
 779
 780
 781
 782
 783
 784
 785
 786
 787
 788
 789
 790
 791
 792
 793
 794
 795
 796
 797
 798
 799
 800
 801
 802
 803
 804
 805
 806
 807
 808
 809
 810
 811
 812
 813
 814
 815
 816
 817
 818
 819
 820
 821
 822
 823
 824
 825
 826
 827
 828
 829
 830
 831
 832
 833
 834
 835
 836
 837
 838
 839
 840
 841
 842
 843
 844
 845
 846
 847
 848
 849
 850
 851
 852
 853
 854
 855
 856
 857
 858
 859
 860
 861
 862
 863
 864
 865
 866
 867
 868
 869
 870
 871
 872
 873
 874
 875
 876
 877
 878
 879
 880
 881
 882
 883
 884
 885
 886
 887
 888
 889
 890
 891
 892
 893
 894
 895
 896
 897
 898
 899
 900
 901
 902
 903
 904
 905
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
1627
1628
1629
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
1657
1658
1659
1660
1661
1662
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
1680
1681
1682
1683
1684
1685
1686
1687
1688
1689
1690
1691
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
1702
1703
1704
1705
1706
1707
1708
1709
1710
1711
1712
1713
1714
1715
1716
1717
1718
1719
1720
1721
1722
1723
1724
1725
1726
1727
1728
1729
1730
1731
1732
1733
1734
1735
1736
1737
1738
1739
1740
1741
1742
1743
1744
1745
1746
1747
1748
1749
1750
1751
1752
1753
1754
1755
1756
1757
1758
1759
1760
1761
1762
1763
1764
1765
1766
1767
1768
1769
1770
1771
1772
1773
1774
1775
1776
1777
1778
1779
1780
1781
1782
1783
1784
1785
1786
1787
1788
1789
1790
1791
1792
1793
1794
1795
1796
1797
1798
1799
1800
1801
1802
1803
1804
1805
1806
1807
1808
1809
1810
1811
1812
1813
1814
1815
1816
1817
1818
1819
1820
1821
1822
1823
1824
1825
1826
1827
1828
1829
1830
1831
1832
1833
1834
1835
1836
1837
1838
1839
1840
1841
1842
1843
1844
1845
1846
1847
1848
1849
1850
1851
1852
1853
1854
1855
1856
1857
1858
1859
1860
1861
1862
1863
1864
1865
1866
1867
1868
1869
1870
1871
1872
1873
1874
1875
1876
1877
1878
1879
1880
1881
1882
1883
1884
1885
1886
1887
1888
1889
1890
1891
1892
1893
1894
1895
1896
1897
1898
1899
1900
1901
1902
1903
1904
1905
1906
1907
1908
1909
1910
1911
1912
1913
1914
1915
1916
1917
1918
1919
1920
1921
1922
1923
1924
1925
1926
1927
1928
1929
1930
1931
1932
1933
1934
1935
1936
1937
1938
1939
1940
1941
1942
1943
1944
1945
1946
1947
1948
1949
1950
1951
1952
1953
1954
1955
1956
1957
1958
1959
1960
1961
1962
1963
1964
1965
1966
1967
1968
1969
1970
1971
1972
1973
1974
1975
1976
1977
1978
1979
1980
1981
1982
1983
1984
1985
1986
1987
1988
1989
1990
1991
1992
1993
1994
1995
1996
1997
1998
1999
2000
2001
2002
2003
2004
2005
2006
2007
2008
2009
2010
2011
2012
2013
2014
2015
2016
2017
2018
2019
2020
2021
2022
2023
2024
2025
2026
2027
2028
2029
2030
2031
2032
2033
2034
2035
2036
2037
2038
2039
2040
2041
2042
2043
2044
2045
2046
2047
2048
2049
2050
2051
2052
2053
class FlowDashApp(Viewer):
    """FlowDash application: scans a project directory and serves its pages and components."""

    auth_config = param.ClassSelector(
        class_=AuthConfig,
        doc="""
        Project-level authorization configuration controlling group discovery,
        the admin groups and the default access policy.""",
    )

    breakpoints = param.List(default=[768, 1200], doc="Responsive breakpoints for the tile grid.")

    configure_layout = param.Callable(
        default=None,
        doc="""
        Optional callback invoked on every navigation with
        (app, content, route). Use it to set `app.sidebar` and
        `app.contextbar` for the page currently being served.""",
    )

    contextbar = Children(default=[], doc="Items prepended to the contextbar.")

    contextbar_open = param.Boolean(default=False, doc="Whether the contextbar is open.")

    home_dashboard = param.String(
        default=None,
        doc="""
        Dashboard shown on the homepage ('/'). Accepts a dashboard id or
        title. When unset, the homepage shows the dashboard grid launcher.""",
    )

    nav_variant = param.Selector(
        default="drawer",
        objects=["drawer", "menubar"],
        doc="""
        Where the navigation menu is rendered. 'drawer' docks a MenuList in a
        right-hand drawer; 'menubar' places a MenuBar in the page header with
        quick-action icons alongside it.""",
    )

    page_options = param.Dict(
        default={},
        doc="""
        Extra keyword arguments passed through to the underlying
        `panel_material_ui.Page`, overriding the app's own defaults.""",
    )

    project_dir = param.Path(doc="Path to the project directory.")

    sidebar = Children(default=[], doc="Items prepended to the sidebar.")

    store = param.ClassSelector(
        class_=DashboardStore, doc="DashboardStore instance for persistence."
    )

    title = param.String(default="FlowDash", doc="Application title shown in the browser tab.")

    _main_task: asyncio.Task | None = None

    def __init__(self, registry: dict[str, RegistryEntry] | None = None, **params):
        super().__init__(**params)

        if self.auth_config is None:
            self.auth_config = AuthConfig()

        if registry is None:
            registry = build_registry(pathlib.Path(self.project_dir))
        page_entries = {k: v for k, v in registry.items() if v.metadata.page}
        component_entries = {k: v for k, v in registry.items() if v.metadata.component}
        # Session state is built from AST metadata — no imports needed.
        session_state_class = build_session_state_class(registry)
        self._registry = registry
        self._page_entries = page_entries
        self._component_entries = component_entries
        # Component specs and dataflow graph are built lazily on first editor visit.
        self._component_specs: dict = {}
        self._session_state_class = session_state_class

        self._session_state = self._session_state_class()
        self._identity = resolve_identity(self.auth_config)
        self._user_id = self._resolve_user_id()
        self._loading = False
        self._dirty = False
        self._components_loaded = False
        self._edge_id_map: dict[str, tuple[str, str, str, str]] = {}
        self._current_dashboard: DashboardModel | None = None
        self._tile_items: list[dict] = []
        self._tile_objects: list[Viewable] = []
        self._sidebar_views: list[Viewable] = []
        self._sidebar_container = pn.Column(sizing_mode="stretch_width")
        self._component_picker = self._make_component_picker()
        self._dataflow_graph = DataflowGraph({}, on_error=self._on_wiring_error)
        self._flow_canvas = self._build_flow_canvas()
        self._component_view = self._build_component_view()
        self._menu_bar = None
        self._nav_menu = self._build_nav_menu()
        self._nav_drawer = pmui.Drawer(
            pmui.Typography(
                "Navigation",
                variant="overline",
                margin=(8, 16, 0, 16),
                styles={"opacity": "0.6", "letter-spacing": "0.08em"},
            ),
            pmui.Divider(margin=(4, 0, 4, 0)),
            self._nav_menu,
            anchor="right",
            inline=True,
            variant="docked",
            width_policy="fixed",
        )
        if self.nav_variant == "menubar":
            self._build_nav_bar()
        self._menubar_mode = self.nav_variant == "menubar"
        self._build_dialog()
        self._build_unsaved_dialog()
        self._build_share_dialog()
        # The share dialog is a portaled overlay; mount it once inside the
        # always-present nav drawer rather than in each _page.main layout.
        page_kwargs = {
            "title": self.title,
            "theme_config": {"palette": {"primary": {"main": "#0072B5"}}},
            "sidebar_open": False,
            "sidebar": self.param.sidebar.rx() + [self._sidebar_container],  # noqa: RUF005
            "contextbar": self.param.contextbar.rx(),
            "contextbar_variant": "persistent",
            "contextbar_open": self.param.contextbar_open,
            **self.page_options,
        }
        overflow_patch = {".main-content": {"overflow-x": "hidden"}}
        if "sx" in page_kwargs:
            page_kwargs["sx"].update(**overflow_patch)
        else:
            page_kwargs["sx"] = {".main-content": {"overflow-x": "hidden"}}
        if self._menu_bar is not None:
            page_kwargs.setdefault("header", [self._menu_bar])
        self._page = pmui.Page(**page_kwargs)
        pn.state.onload(self._load_page_layout)

    @asynccontextmanager
    async def _loading_screen(self, delay: float = 0.5):
        """Show a loading placeholder if the block takes longer than *delay* seconds."""

        async def _show_after_delay():
            await asyncio.sleep(delay)
            self._page.main = [
                pmui.LinearProgress(sizing_mode="stretch_width"),
                self._dialog,
                self._unsaved_dialog,
                self._share_dialog,
            ]

        task = asyncio.create_task(_show_after_delay())
        try:
            yield
        finally:
            task.cancel()

    def _nav_content(self, content):
        """Wrap page content with the docked nav drawer, unless in menubar mode.

        In menubar mode navigation lives in the page header, so the content is
        returned as-is; in drawer mode it is paired with the docked drawer.
        """
        if self._menubar_mode:
            return content
        return pn.Row(content, self._nav_drawer, sizing_mode="stretch_both")

    def _resolve_user_id(self) -> str:
        return self._identity.user

    @staticmethod
    @cache
    def _accepted_injected_params(app):
        if inspect.isclass(app) and issubclass(app, pn.viewable.Viewer):
            return {
                p for p in ("config", "executor", "instance_config", "context") if hasattr(app, p)
            }
        return inspect.signature(app).parameters.keys() & {
            "config",
            "executor",
            "instance_config",
            "context",
        }

    def _add_kwargs_dict(self, app, *, context: str, instance_config: dict | None = None):
        params = self._accepted_injected_params(app)
        kwargs = {}
        if "context" in params:
            kwargs["context"] = context
        if "instance_config" in params and instance_config is not None:
            kwargs["instance_config"] = instance_config
        if "config" in params:
            kwargs["config"] = self._session_state
        return kwargs

    def _entry_from_key(self, key):
        app_id = "/".join(key)
        return self._page_entries.get(app_id)

    def _default_allow(self) -> bool:
        return self.auth_config.default_allow if self.auth_config else True

    def _can_access_entry(self, entry: RegistryEntry) -> bool:
        """Whether the current identity may access a page/component entry."""
        authorize = entry.metadata.authorize
        if authorize is not None:
            try:
                return bool(authorize(self._identity))
            except Exception:
                logger.exception("authorize callback failed for '%s'", entry.app_id)
                return False
        return is_authorized(
            entry.metadata.permission,
            self._identity,
            default_allow=self._default_allow(),
        )

    def _admin_groups(self) -> frozenset[str]:
        return self.auth_config.admin_groups if self.auth_config else frozenset()

    def _accessible_page_entries(self) -> dict[str, RegistryEntry]:
        """Page entries the current identity is authorized to see."""
        return {
            app_id: entry
            for app_id, entry in self._page_entries.items()
            if self._can_access_entry(entry)
        }

    def _can_administer_dashboard(self, dashboard_id: str) -> bool:
        """Whether the current identity may edit/delete/share a dashboard."""
        return self.store.can_administer(self._identity, dashboard_id, self._admin_groups())

    def _access_denied_view(self, title: str | None = None):
        """Generate the view shown when the user is not authorized for the target."""
        label = f" **{title}**" if title else ""
        return pmui.Alert(
            object=(
                f"Access denied.{label} You are not authorized to view this "
                f"(signed in as `{self._identity.user}`)."
            ),
            severity="error",
            title="Access denied",
            sizing_mode="stretch_width",
        )

    async def _instantiate_entry(
        self,
        entry: RegistryEntry,
        *,
        context: str,
        instance_config: dict | None = None,
    ):
        unsatisfied = check_requirements(self._session_state, entry.metadata.requires)
        blocking = [u for u in unsatisfied if u["blocking"]]
        if blocking:
            keys = ", ".join(u["key"] for u in blocking)
            return pn.pane.Alert(
                f"**{entry.title}** is waiting for: `{keys}`",
                alert_type="warning",
            )

        app = await asyncio.to_thread(entry.load)
        if not callable(app):
            return pn.panel(app)
        kwargs = self._add_kwargs_dict(app, context=context, instance_config=instance_config)
        if inspect.iscoroutinefunction(app):
            return await app(**kwargs)
        return await asyncio.to_thread(lambda: pn.panel(app(**kwargs)))

    async def _render_page(self, key):
        entry = self._entry_from_key(key)
        if entry is None:
            return f"Unknown page: {'/'.join(key)}"

        if not self._can_access_entry(entry):
            return self._access_denied_view(entry.title)

        if self._main_task is not None and not self._main_task.done():
            self._main_task.cancel()

        try:
            async with self._loading_screen():
                coroutine = self._instantiate_entry(entry, context="page")
                self._main_task = asyncio.create_task(coroutine)
                return await self._main_task
        except asyncio.CancelledError:
            return None
        except Exception as e:
            logger.exception("Error rendering page '%s'", "/".join(key))
            err_name = type(e).__name__
            return pn.pane.Alert(
                f"**{err_name}**: {e}\n<hr>\n<pre> {escape(traceback.format_exc())}</pre>\n",
                alert_type="danger",
                styles={"color": "black"},
            )

    def _make_component_picker(self):
        groups: dict[str, dict[str, str]] = {}
        for app_id, entry in self._component_entries.items():
            section = entry.section.replace("_", " ")
            groups.setdefault(section, {})[entry.title] = app_id
        value = next(iter(self._component_entries), None)
        return pmui.Select(
            label="Component",
            groups=groups,
            value=value,
            searchable=True,
            filter_on_search=True,
            size="small",
        )

    def _node_types_from_specs(self):
        node_types = {}
        node_editors = {}
        for comp_id, spec in self._component_specs.items():
            type_key = comp_id.replace("/", "__")
            node_types[type_key] = pr.NodeType(
                type=type_key,
                label=spec.title,
                schema=spec.config_state_class,
                inputs=[
                    {"id": port.name, "label": port.label or port.name} for port in spec.inputs
                ],
                outputs=[
                    {"id": port.name, "label": port.label or port.name} for port in spec.outputs
                ],
            )
            if spec.config_editor is not None:
                node_editors[type_key] = spec.config_editor
        return node_types, node_editors

    def _rebuild_flow_canvas(self):
        """Update node_types on the live ReactFlow canvas after component load."""
        node_types, node_editors = self._node_types_from_specs()
        self._flow.param.update(node_types=node_types, node_editors=node_editors)

    def _build_flow_canvas(self):
        node_types, node_editors = self._node_types_from_specs()

        flow = pr.ReactFlow(
            nodes=[],
            edges=[],
            node_types=node_types,
            node_editors=node_editors,
            editable=True,
            enable_connect=True,
            show_minimap=True,
            sizing_mode="stretch_both",
            min_height=600,
            stylesheets=[
                """\
            .react-flow__node {
              padding: 0;
              border-radius: 6px;
              border: 1px solid var(--xy-node-border, var(--panel-border-color));
              background-color: var(--xy-node-background-color, var(--panel-background-color));
              box-shadow: 0 1px 2px var(--panel-shadow-color);
              color: var(--xy-node-color, var(--panel-on-background-color));
              font-size: 13px;
              min-width: 140px;
            }
            .react-flow__handle {
              width: 14px;
              height: 14px;
              border: 1px solid black;
              background: transparent;
            }"""
            ],
        )

        def _on_edge_added(event):
            if self._loading:
                return
            edge = event.get("edge", event) if isinstance(event, dict) else {}
            src_id = edge.get("source", "")
            tgt_id = edge.get("target", "")
            src_handle = edge.get("sourceHandle", "")
            tgt_handle = edge.get("targetHandle", "")
            if src_id and tgt_id and src_handle and tgt_handle:
                result = self._dataflow_graph.add_edge(src_id, src_handle, tgt_id, tgt_handle)
                if result is True:
                    edge_id = edge.get("id", "")
                    if edge_id:
                        self._edge_id_map[edge_id] = (src_id, src_handle, tgt_id, tgt_handle)
                    self._dirty = True
                    pn.state.notifications.success(
                        f"Wired: {src_handle}{tgt_handle}", duration=3000
                    )
                else:
                    logger.warning("Edge rejected: %s", result)
                    pn.state.notifications.error(result, duration=5000)
                    flow.remove_edge(edge.get("id", ""))

        def _on_edge_deleted(event):
            if self._loading:
                return
            edge_id = event.get("edge_id", "") if isinstance(event, dict) else ""
            if not edge_id:
                return
            mapping = self._edge_id_map.pop(edge_id, None)
            if mapping:
                self._dataflow_graph.remove_edge(*mapping)
                self._dirty = True

        def _on_node_data_changed(event):
            if self._loading:
                return
            node_id = event.get("node_id", "") if isinstance(event, dict) else ""
            patch = event.get("patch", {}) if isinstance(event, dict) else {}
            if not node_id or not patch:
                return
            self._apply_config_patch(node_id, patch)

        def _on_node_deleted(event):
            node_id = event.get("node_id", "") if isinstance(event, dict) else ""
            if node_id:
                self._dataflow_graph.remove_node(node_id)
                idx = next(
                    (
                        i
                        for i, item in enumerate(self._tile_items)
                        if item["instance_id"] == node_id
                    ),
                    None,
                )
                if idx is not None:
                    self._tile_items.pop(idx)
                    self._tile_objects.pop(idx)
                self._dirty = True

        flow.on("edge_added", _on_edge_added)
        flow.on("edge_deleted", _on_edge_deleted)
        flow.on("node_data_changed", _on_node_data_changed)
        flow.on("node_deleted", _on_node_deleted)

        self._flow = flow
        return flow

    def _apply_config_patch(self, node_id, patch):
        """Apply an editor patch to a node's config state and persist it."""
        config_state = self._dataflow_graph.get_config_state(node_id)
        applied = {}
        for key, value in patch.items():
            if config_state is not None and hasattr(config_state.param, key):
                try:
                    setattr(config_state, key, value)
                except Exception as exc:
                    logger.warning("Config '%s' rejected on %s: %s", key, node_id, exc)
                    continue
            applied[key] = value
        if not applied:
            return
        for item in self._tile_items:
            if item["instance_id"] == node_id:
                item.setdefault("config", {}).update(applied)
                break
        self._dirty = True

    def _seed_config_state(self, instance_id, config):
        """Overlay saved config onto a node's config state and return node data seed."""
        config_state = self._dataflow_graph.get_config_state(instance_id)
        if config_state is None:
            return {}
        for key, value in (config or {}).items():
            if hasattr(config_state.param, key):
                try:
                    setattr(config_state, key, value)
                except Exception as exc:
                    logger.warning("Saved config '%s' rejected on %s: %s", key, instance_id, exc)
        return {name: getattr(config_state, name) for name in config_state.param if name != "name"}

    def _bind_config_to_viewer(self, instance, config_state):
        """Sync config-state params onto a Viewer instance, live."""
        for name in config_state.param:
            if name == "name" or not hasattr(instance.param, name):
                continue
            try:
                setattr(instance, name, getattr(config_state, name))
            except Exception as exc:
                logger.warning("Config '%s' could not be set: %s", name, exc)
                continue

            def _propagate(event, _name=name):
                try:
                    setattr(instance, _name, event.new)
                except Exception as exc:
                    logger.warning("Config '%s' update failed: %s", _name, exc)

            config_state.param.watch(_propagate, name)

    def _instantiate_for_node(self, entry, node_state, config_state=None):
        """Create a live component view wired to the node_state."""
        app_fn = entry.load()

        if not callable(app_fn):
            return pn.panel(app_fn)

        if inspect.isclass(app_fn) and issubclass(app_fn, pn.viewable.Viewer):
            return self._instantiate_viewer_for_node(app_fn, entry, node_state, config_state)

        sig = inspect.signature(app_fn)
        kwargs = {}
        if "config" in sig.parameters:
            kwargs["config"] = node_state
        if "instance_config" in sig.parameters and config_state is not None:
            kwargs["instance_config"] = config_state
        if "context" in sig.parameters:
            kwargs["context"] = "component"

        result = app_fn(**kwargs)
        return pn.panel(result)

    def _instantiate_viewer_for_node(self, viewer_cls, entry, node_state, config_state=None):
        """Instantiate a Viewer and wire its params to the node_state."""
        spec = self._component_specs.get(entry.app_id)
        instance = viewer_cls()

        if config_state is not None:
            self._bind_config_to_viewer(instance, config_state)

        input_names = [p.name for p in spec.inputs] if spec else []
        for name in input_names:
            if not hasattr(instance.param, name):
                continue

            def _propagate_input(event, _name=name):
                setattr(instance, _name, event.new)

            node_state.param.watch(_propagate_input, name)

        output_info = instance.param.outputs()
        for name, (_, method, _) in output_info.items():
            if not hasattr(node_state.param, name):
                continue
            method_name = method.__name__ if callable(method) else method
            deps = instance.param.method_dependencies(method_name)
            dep_names = [d.name for d in deps if d.name != "name"]

            def _propagate_output(event, _method=method, _name=name):
                try:
                    val = _method() if callable(_method) else getattr(instance, _method)()
                    setattr(node_state, _name, val)
                except Exception as exc:
                    logger.error("Output '%s' failed: %s", _name, exc, exc_info=exc)

            if dep_names:
                instance.param.watch(_propagate_output, dep_names)
            try:
                val = method() if callable(method) else getattr(instance, method)()
                setattr(node_state, name, val)
            except Exception:
                pass

        return pn.panel(instance)

    def _build_component_view(self):
        self._add_button = pmui.Button(icon="add", color="primary", variant="outlined")
        self._clear_button = pmui.Button(icon="delete_sweep", color="danger", variant="outlined")
        self._save_button = pmui.Button(icon="save", color="primary", variant="outlined")
        self._share_button = pmui.Button(
            icon="share", color="primary", variant="outlined", visible=False
        )
        self._add_button.on_click(self._add_component_to_graph)
        self._clear_button.on_click(lambda _event: self._clear_components())
        self._save_button.on_click(lambda _event: self._save_current_dashboard())
        self._share_button.on_click(lambda _event: self._share_current_dashboard())

        no_components = len(self._component_entries) == 0
        self._component_picker.disabled = no_components
        self._add_button.disabled = no_components

        self._preview_switch = pmui.Switch(
            label="Preview",
            value=False,
            align="center",
            margin=(0, 10),
        )
        self._preview_switch.param.watch(
            lambda e: self._tile_grid.param.update(editable=not e.new, card=not e.new), "value"
        )
        self._mode_toggle = pmui.RadioButtonGroup(
            options={":material/cable:": "wiring", ":material/dashboard:": "dashboard"},
            value="wiring",
        )
        self._workspace_area = pn.Column(
            self._flow_canvas, sizing_mode="stretch_both", scroll="y-auto"
        )

        self._preview_switch.visible = False

        @pn.io.hold()
        def _on_mode_change(event):
            if event.new == "dashboard":
                self._workspace_area[:] = [self._tile_grid]
                self._rebuild_tile_grid()
                self._preview_switch.visible = True
            else:
                self._pending_tile_layout = self._tile_grid.layout
                self._pending_breakpoints = self._tile_grid.breakpoints
                self._pending_responsive_layouts = self._tile_grid.responsive_layouts
                self._workspace_area[:] = [self._flow_canvas]
                self._preview_switch.visible = False
                self._preview_switch.value = False

        self._mode_toggle.param.watch(_on_mode_change, "value")

        self._controls_row = pn.Row(
            self._component_picker,
            self._add_button,
            self._clear_button,
            self._save_button,
            self._share_button,
            pn.layout.HSpacer(),
            self._preview_switch,
            self._mode_toggle,
            sizing_mode="stretch_width",
            align="center",
        )
        return pn.Column(
            self._controls_row,
            self._workspace_area,
            sizing_mode="stretch_both",
        )

    @property
    def _tile_grid(self):
        if not hasattr(self, "_tile__grid"):
            self._tile__grid = TileGrid(
                breakpoints=list(self.breakpoints),
                card=False,
                close_action="hide",
                editable=False,
                local_save=False,
                min_height=320,
                sizing_mode="stretch_both",
            )
        return self._tile__grid

    def _apply_responsive_config(self, breakpoints, responsive_layouts):
        if breakpoints:
            self._tile_grid.breakpoints = breakpoints
        if responsive_layouts:
            self._tile_grid.responsive_layouts = responsive_layouts

    def _on_wiring_error(self, source_id, source_port, target_id, target_port, exc):
        logger.error(
            "Runtime wiring error (%s.%s -> %s.%s): %s",
            source_id,
            source_port,
            target_id,
            target_port,
            exc,
            exc_info=exc,
        )
        pn.state.notifications.error(
            f"Runtime wiring error ({source_port}{target_port}): {exc}",
            duration=5000,
        )

    async def _add_component_to_graph(self, _event=None):
        component_id = self._component_picker.value
        entry = self._component_entries.get(component_id)
        if entry is None:
            pn.state.notifications.warning("Select a valid component first.", duration=3000)
            return

        # The editor can be entered in-session (dashboard create/edit) without a
        # navigation, so the specs may not be built yet.
        if not self._components_loaded:
            async with self._loading_screen():
                await self._ensure_components_loaded()

        spec = self._component_specs.get(component_id)
        if spec is None:
            pn.state.notifications.error(
                f"Component '{component_id}' could not be loaded.", duration=5000
            )
            return

        type_key = component_id.replace("/", "__")
        instance_id = f"{type_key}_{uuid.uuid4().hex[:6]}"

        node_state = self._dataflow_graph.add_node(instance_id, component_id)
        config_state = self._dataflow_graph.get_config_state(instance_id)
        config_data = self._seed_config_state(instance_id, {})

        try:
            view = self._instantiate_for_node(entry, node_state, config_state)
        except Exception as e:
            logger.exception("Failed to add component '%s'", component_id)
            self._dataflow_graph.remove_node(instance_id)
            pn.state.notifications.error(f"Failed to add component: {e}", duration=5000)
            return

        node_count = len(self._tile_items)
        col = node_count % 3
        row = node_count // 3
        position = {"x": col * 350, "y": row * 250}

        node = pr.NodeSpec(
            id=instance_id,
            type=type_key,
            position=position,
            label=spec.title,
            data=config_data,
        )
        node_dict = node.to_dict()
        node_dict["view"] = view
        self._flow.add_node(node_dict)

        self._tile_items.append(
            {"instance_id": instance_id, "component_id": component_id, "config": {}}
        )
        self._tile_objects.append(view)
        self._dirty = True

        pn.state.notifications.success(f"Added component: {entry.title}", duration=3000)

    def _rebuild_sidebar(self):
        """Populate the page sidebar from tiles whose component opts into it.

        The sidebar is independent of the wiring/dashboard toggle, so this runs
        whenever a dashboard is shown, not only when the tile grid is visible.
        """
        sidebar_views = []
        for i, item in enumerate(self._tile_items):
            component_id = item["component_id"]
            entry = self._component_entries.get(component_id)
            if entry is None or not entry.metadata.sidebar:
                continue
            view = self._tile_objects[i] if i < len(self._tile_objects) else None
            if view is None:
                view = pn.pane.Markdown(f"*{entry.title}*")
            sidebar_views.append(view)
        self._sidebar_views = sidebar_views
        self._sidebar_container.objects = sidebar_views
        self._page.sidebar_open = bool(sidebar_views)

    @pn.io.hold()
    def _rebuild_tile_grid(self):
        grid_views = []
        for i, item in enumerate(self._tile_items):
            component_id = item["component_id"]
            entry = self._component_entries.get(component_id)
            if entry is None:
                continue
            if entry.metadata.sidebar:
                continue
            view = self._tile_objects[i] if i < len(self._tile_objects) else None
            if view is None:
                view = pn.pane.Markdown(f"*{entry.title}*")
            grid_views.append(view)
        self._tile_grid[:] = grid_views
        self._rebuild_sidebar()
        pending = getattr(self, "_pending_tile_layout", [])
        if pending:
            self._tile_grid.layout = pending
            self._pending_tile_layout = []
        pending_bp = getattr(self, "_pending_breakpoints", [])
        pending_rl = getattr(self, "_pending_responsive_layouts", {})
        if pending_bp or pending_rl:
            self._apply_responsive_config(pending_bp, pending_rl)
            self._pending_breakpoints = []
            self._pending_responsive_layouts = {}

    @pn.io.hold()
    def _clear_components(self):
        had_items = bool(self._tile_items)
        self._reset_canvas()
        if had_items:
            self._dirty = True
        pn.state.notifications.info("Cleared all component tiles.", duration=3000)

    def _save_current_dashboard(self):
        if self._current_dashboard is None:
            pn.state.notifications.warning(
                "No dashboard loaded. Create one from the sidebar.", duration=4000
            )
            return

        if not self._can_administer_dashboard(self._current_dashboard.dashboard_id):
            pn.state.notifications.error(
                "You have view-only access to this dashboard.", duration=4000
            )
            return

        positions = {}
        for node in self._flow.nodes:
            node_id = node.get("id", "")
            pos = node.get("position", {})
            positions[node_id] = (pos.get("x", 0), pos.get("y", 0))

        self._current_dashboard.items = [
            DashboardItem(
                instance_id=item["instance_id"],
                component_id=item["component_id"],
                x=positions.get(item["instance_id"], (0, 0))[0],
                y=positions.get(item["instance_id"], (0, 0))[1],
                config=item.get("config", {}),
            )
            for item in self._tile_items
        ]
        self._current_dashboard.edges = [
            DashboardEdge(
                source=edge["source"],
                source_port=edge["source_port"],
                target=edge["target"],
                target_port=edge["target_port"],
            )
            for edge in self._dataflow_graph.edges
        ]
        self._current_dashboard.tile_layout = self._tile_grid.layout
        self._current_dashboard.breakpoints = self._tile_grid.breakpoints
        self._current_dashboard.responsive_layouts = self._tile_grid.responsive_layouts

        try:
            self.store.save_dashboard(self._current_dashboard)
        except Exception as exc:
            logger.exception("Failed to save dashboard")
            pn.state.notifications.error(f"Save failed: {exc}", duration=5000)
            return
        self._dirty = False
        pn.state.notifications.success(
            f'Dashboard "{self._current_dashboard.title}" saved.', duration=3000
        )

    def _share_current_dashboard(self):
        if self._current_dashboard is None:
            pn.state.notifications.warning("No dashboard loaded.", duration=3000)
            return
        self._open_share_dialog(
            self._current_dashboard.dashboard_id, self._current_dashboard.title
        )

    def _reset_canvas(self):
        """Tear down all node/edge state and clear the ReactFlow canvas."""
        for node_id in list(self._dataflow_graph.node_ids):
            self._dataflow_graph.remove_node(node_id)
        self._tile_items = []
        self._tile_objects = []
        self._edge_id_map.clear()
        self._sidebar_views = []
        self._sidebar_container.objects = []
        self._flow.nodes = []
        self._flow.edges = []

    async def _load_dashboard(self, dashboard_id: str, edit: bool = False):
        await self._ensure_components_loaded()
        with pn.io.hold():
            self._load_dashboard_sync(dashboard_id, edit=edit)

    def _load_dashboard_sync(self, dashboard_id: str, edit: bool = False):
        dashboard = self.store.load_for_access(
            self._identity, dashboard_id, default_allow=self._default_allow()
        )
        if edit and dashboard is not None and not self._can_administer_dashboard(dashboard_id):
            # A viewer with read access followed an edit link; downgrade to view.
            edit = False
        if dashboard is None:
            self._page.main = [
                self._nav_content(
                    pn.pane.Alert(f"Dashboard not found: {dashboard_id}", alert_type="danger")
                ),
                self._dialog,
                self._unsaved_dialog,
                self._share_dialog,
            ]
            return
        self._current_dashboard = dashboard
        self._loading = True

        self._reset_canvas()

        for item in dashboard.items:
            component_id = item.component_id
            entry = self._component_entries.get(component_id)
            if entry is None:
                continue
            spec = self._component_specs.get(component_id)
            if spec is None:
                continue

            instance_id = item.instance_id
            type_key = component_id.replace("/", "__")
            node_state = self._dataflow_graph.add_node(instance_id, component_id)
            config_state = self._dataflow_graph.get_config_state(instance_id)
            config_data = self._seed_config_state(instance_id, item.config)

            try:
                view = self._instantiate_for_node(entry, node_state, config_state)
            except Exception:
                logger.exception("Error loading component '%s' (%s)", component_id, instance_id)
                self._dataflow_graph.remove_node(instance_id)
                continue

            position = {"x": item.x, "y": item.y}
            node = pr.NodeSpec(
                id=instance_id,
                type=type_key,
                position=position,
                label=spec.title,
                data=config_data,
            )
            node_dict = node.to_dict()
            node_dict["view"] = view
            self._flow.add_node(node_dict)

            self._tile_items.append(item.to_dict())
            self._tile_objects.append(view)

        edge_counter = 0
        for edge in dashboard.edges:
            success = self._dataflow_graph.add_edge(
                edge.source, edge.source_port, edge.target, edge.target_port
            )
            if success is True:
                edge_counter += 1
                edge_id = f"e{edge_counter}"
                self._edge_id_map[edge_id] = (
                    edge.source,
                    edge.source_port,
                    edge.target,
                    edge.target_port,
                )
                self._flow.add_edge(
                    {
                        "id": edge_id,
                        "source": edge.source,
                        "target": edge.target,
                        "sourceHandle": edge.source_port,
                        "targetHandle": edge.target_port,
                        "markerEnd": {"type": "arrowclosed"},
                    }
                )

        self._loading = False
        self._dirty = False
        self._pending_tile_layout = dashboard.tile_layout or []
        self._pending_breakpoints = dashboard.breakpoints or []
        self._pending_responsive_layouts = dashboard.responsive_layouts or {}

        pn.state.notifications.info(
            f'Loaded dashboard "{dashboard.title}" with {len(self._tile_items)} tiles.',
            duration=3000,
        )
        if edit:
            self._show_edit_mode()
        else:
            self._show_view_mode()
        self._sync_menu_active(f"{DASH_ROUTE_PREFIX}{dashboard_id}")
        self._apply_layout_config(self._component_view, f"{DASH_ROUTE_PREFIX}{dashboard_id}")
        self._page.main = [
            self._nav_content(self._component_view),
            self._dialog,
            self._unsaved_dialog,
            self._share_dialog,
        ]

    def _create_new_dashboard(self, title_str: str):
        title_str = title_str.strip()
        if not title_str:
            pn.state.notifications.warning("Dashboard title cannot be empty.", duration=3000)
            return
        dashboard = self.store.create_dashboard(self._user_id, title_str)
        self._current_dashboard = dashboard
        self._reset_canvas()
        self._dirty = False

        pn.state.notifications.success(
            f'Created new dashboard "{dashboard.title}".', duration=3000
        )
        self._refresh_sidebar_dashboards()
        if pn.state.location:
            pn.state.location.param.update(
                pathname=f"{DASH_ROUTE_PREFIX}{dashboard.dashboard_id}",
                search="?edit=true",
            )
        self._show_edit_mode()
        self._sync_menu_active(f"{DASH_ROUTE_PREFIX}{dashboard.dashboard_id}")
        self._apply_layout_config(
            self._component_view, f"{DASH_ROUTE_PREFIX}{dashboard.dashboard_id}"
        )
        self._page.main = [
            self._nav_content(self._component_view),
            self._dialog,
            self._unsaved_dialog,
            self._share_dialog,
        ]

    def _delete_dashboard(self, dashboard_id: str):
        if not self._can_administer_dashboard(dashboard_id):
            pn.state.notifications.error("You are not allowed to delete this dashboard.")
            return
        was_current = bool(
            self._current_dashboard and self._current_dashboard.dashboard_id == dashboard_id
        )
        owner = self.store.get_owner(dashboard_id) or self._user_id
        self.store.delete_dashboard(owner, dashboard_id)
        if was_current:
            self._current_dashboard = None
            self._reset_canvas()
            self._dirty = False

        self._refresh_sidebar_dashboards()
        pn.state.notifications.info("Dashboard deleted.", duration=3000)

        if was_current:
            self._navigate_to("/")
        elif pn.state.location is not None and pn.state.location.pathname == "/":
            self._page.main = [
                self._nav_content(self._build_launcher()),
                self._dialog,
                self._unsaved_dialog,
                self._share_dialog,
            ]

    def _rename_dashboard(self, dashboard_id: str, new_title: str):
        new_title = new_title.strip()
        if not new_title:
            return
        if not self._can_administer_dashboard(dashboard_id):
            pn.state.notifications.error("You are not allowed to rename this dashboard.")
            return
        owner = self.store.get_owner(dashboard_id) or self._user_id
        self.store.rename_dashboard(owner, dashboard_id, new_title)
        if self._current_dashboard and self._current_dashboard.dashboard_id == dashboard_id:
            self._current_dashboard.title = new_title
        self._refresh_sidebar_dashboards()

    def _refresh_sidebar_dashboards(self):
        dash_items = self._get_dashboard_menu_items()
        items = list(self._menu_list.items)
        items[-1] = {**items[-1], "items": dash_items}
        self._menu_list.items = items
        self._refresh_menu_bar()

    def _build_launcher(self):
        sections: dict[str, list[RegistryEntry]] = {}
        for entry in self._accessible_page_entries().values():
            sections.setdefault(entry.section, []).append(entry)

        accordion_items = []
        component_item = None

        for section, entries in sorted(sections.items()):
            cards = []
            for entry in sorted(entries, key=lambda e: e.name):
                icon_name = entry.metadata.icon or "article"
                card = pmui.Card(
                    pmui.ButtonIcon(
                        icon=icon_name,
                        icon_size="3em",
                        disabled=True,
                        stylesheets=[":host { pointer-events: none; opacity: 1; }"],
                    ),
                    title=entry.title,
                    title_variant="h4",
                    collapsible=False,
                    stylesheets=[_LAUNCHER_CARD_CSS],
                    width=200,
                    height=140,
                )
                clickable = pmui.Clickable(object=card)
                clickable.on_click(partial(self._launcher_navigate, entry.page_path))
                cards.append(clickable)

            section_label = section.replace("_", " ")
            content = pn.FlexBox(*cards, gap="12px", margin=(0, 0, 12, 0))
            if section_label.lower() == "components":
                component_item = (section_label, content)
            else:
                accordion_items.append((section_label, content))

        dashboards = self.store.list_accessible(
            self._identity, default_allow=self._default_allow()
        )
        dash_cards = []
        for d in dashboards:
            can_admin = self._can_administer_dashboard(d.dashboard_id)
            speed_dial = None
            if can_admin:
                speed_dial = pmui.SpeedDial(
                    items=self._dashboard_speed_dial_items(can_admin),
                    icon="more_vert",
                    direction="down",
                    color="default",
                    size="small",
                    persistent_tooltips=True,
                    stylesheets=[_LAUNCHER_SPEED_DIAL_CSS],
                )
                speed_dial.param.watch(
                    partial(self._on_launcher_dash_action, d.dashboard_id, d.title), "value"
                )

            card = pmui.Card(
                pmui.ButtonIcon(
                    icon="dashboard",
                    icon_size="3em",
                    disabled=True,
                    stylesheets=[":host { pointer-events: none; opacity: 1; }"],
                ),
                title=d.title,
                collapsible=False,
                stylesheets=[_LAUNCHER_CARD_CSS],
                title_variant="h4",
                width=200,
                height=140,
            )
            path = f"{DASH_ROUTE_PREFIX}{d.dashboard_id}"
            clickable = pmui.Clickable(object=card)
            clickable.on_click(partial(self._launcher_navigate, path))
            wrapper_objects = [clickable]
            if speed_dial is not None:
                wrapper_objects.append(speed_dial)
            wrapper = pn.Column(
                *wrapper_objects,
                styles={"position": "relative", "overflow": "visible"},
                sizing_mode="fixed",
                width=200,
                height=140,
            )
            dash_cards.append(wrapper)

        new_card = pmui.Card(
            pmui.ButtonIcon(
                icon="add",
                icon_size="3em",
                disabled=True,
                stylesheets=[":host { pointer-events: none; opacity: 1; }"],
            ),
            title="New Dashboard",
            collapsible=False,
            stylesheets=[_LAUNCHER_NEW_CARD_CSS],
            title_variant="h4",
            width=200,
            height=140,
        )
        new_clickable = pmui.Clickable(object=new_card)
        new_clickable.on_click(lambda *_args: self._open_create_dialog())
        dash_cards.append(new_clickable)

        accordion_items.append(
            ("Custom Apps", pn.FlexBox(*dash_cards, gap="12px", margin=(0, 0, 12, 0)))
        )

        if component_item:
            accordion_items.append(component_item)

        active = list(range(len(accordion_items)))
        if component_item:
            active.remove(len(accordion_items) - 1)

        return pmui.Accordion(
            *accordion_items,
            active=active,
            toggle=False,
            sizing_mode="stretch_both",
            margin=20,
        )

    def _launcher_navigate(self, path, *_args):
        self._request_navigation(path)

    def _on_launcher_dash_action(self, dashboard_id, title, event):
        value = event.new if hasattr(event, "new") else event
        label = value.get("label") if isinstance(value, dict) else value
        if label == "Edit":
            if pn.state.location:
                pn.state.location.param.update(
                    pathname=f"{DASH_ROUTE_PREFIX}{dashboard_id}",
                    search="?edit=true",
                )
            pn.state.execute(partial(self._load_dashboard_edit, dashboard_id))
        elif label == "Rename":
            self._dialog_name_input.param.update(
                value=title, disabled=False, error_state=False, helper_text=""
            )
            self._dialog_context = {"action": "rename", "dashboard_id": dashboard_id}
            self._dialog.param.update(title="Rename Dashboard", open=True)
        elif label == "Delete":
            self._dialog_name_input.param.update(value=title, disabled=True)
            self._dialog_context = {"action": "delete", "dashboard_id": dashboard_id}
            self._dialog.title = "Delete Dashboard"
            self._dialog.open = True

    async def _ensure_components_loaded(self):
        """Load all component modules and rebuild the dataflow canvas if not done yet."""
        if self._components_loaded:
            return

        already_loaded = all(e.app is not None for e in self._component_entries.values())

        if not already_loaded:
            errors: list[str] = []

            def _load_all():
                for entry in self._component_entries.values():
                    try:
                        entry.load()
                    except Exception as exc:
                        errors.append(f"{entry.app_id}: {exc}")

            await asyncio.to_thread(_load_all)

            for msg in errors:
                logger.warning("Failed to load component: %s", msg)
                pn.state.notifications.warning(f"Component load failed: {msg}", duration=6000)

        self._component_specs = build_component_specs(self._registry)
        self._dataflow_graph = DataflowGraph(self._component_specs, on_error=self._on_wiring_error)
        self._rebuild_flow_canvas()
        self._components_loaded = True

    def _apply_layout_config(self, content, route):
        """Run the ``configure_layout`` hook for the current navigation, if any."""
        if self.configure_layout is None:
            return
        try:
            self.configure_layout(self, content, route)
        except Exception:
            logger.exception("configure_layout hook failed for route '%s'", route)

    async def _show_home_dashboard(self) -> bool:
        """Render the configured home dashboard on '/'. Returns whether it was shown.

        Falls back to the launcher grid (by returning ``False``) if the
        configured dashboard cannot be resolved or the current identity is not
        authorized to view it.
        """
        model = self.store.find_by_id_or_title(self.home_dashboard)
        if model is None:
            logger.warning("Configured home dashboard not found: '%s'", self.home_dashboard)
            return False
        if (
            self.store.load_for_access(
                self._identity, model.dashboard_id, default_allow=self._default_allow()
            )
            is None
        ):
            return False
        async with self._loading_screen():
            await self._ensure_components_loaded()
            await self._load_dashboard(model.dashboard_id, edit=False)
        return True

    async def _load_page_layout(self):
        if pn.state.location is None:
            return
        pathname = pn.state.location.pathname

        if pathname == "/":
            if self.home_dashboard and await self._show_home_dashboard():
                return
            self._current_dashboard = None
            self._sidebar_container.objects = []
            self._page.sidebar_open = False
            launcher = self._build_launcher()
            self._apply_layout_config(launcher, pathname)
            self._page.main = [
                self._nav_content(launcher),
                self._dialog,
                self._unsaved_dialog,
            ]
            return

        if pathname == COMPONENTS_ROUTE:
            self._current_dashboard = None
            self._sidebar_container.objects = []
            async with self._loading_screen():
                await self._ensure_components_loaded()
            self._show_edit_mode()
            self._apply_layout_config(self._component_view, pathname)
            self._page.main = [
                self._nav_content(self._component_view),
                self._dialog,
                self._unsaved_dialog,
            ]
            return

        if pathname.startswith(DASH_ROUTE_PREFIX):
            dashboard_id = pathname[len(DASH_ROUTE_PREFIX) :].strip("/")
            if dashboard_id:
                search = pn.state.location.search or ""
                edit_requested = "edit=true" in search
                async with self._loading_screen():
                    await self._ensure_components_loaded()
                    await self._load_dashboard(dashboard_id, edit=edit_requested)
                return

        self._current_dashboard = None

        self._sidebar_container.objects = []
        key = tuple(pathname.strip("/").split("/"))
        if len(key) == 2 and self._entry_from_key(key):
            content = await self._render_page(key)
            self._apply_layout_config(content, pathname)
            wrapper = pmui.Column(content, sizing_mode="stretch_width")
            self._page.main = [
                self._nav_content(wrapper),
                self._dialog,
                self._unsaved_dialog,
            ]
        else:
            self._apply_layout_config(None, pathname)
            main = [
                f"Invalid URL: {pathname}",
                self._dialog,
                self._unsaved_dialog,
            ]
            if not self._menubar_mode:
                main.append(self._nav_drawer)
            self._page.main = main

    @pn.io.hold()
    def _show_edit_mode(self):
        self._controls_row.visible = True
        self._share_button.visible = (
            self._current_dashboard is not None
            and self._can_administer_dashboard(self._current_dashboard.dashboard_id)
        )
        self._tile_grid.param.update(editable=True, card=True)
        if self._mode_toggle.value == "wiring":
            self._workspace_area[:] = [self._flow_canvas]
            self._rebuild_sidebar()
        else:
            self._workspace_area[:] = [self._tile_grid]
            self._rebuild_tile_grid()
        if pn.state.location is not None:
            pn.state.location.param.update(search="?edit=true")

    @pn.io.hold()
    def _show_view_mode(self):
        self._controls_row.visible = False
        self._tile_grid.param.update(card=False, editable=False)
        self._workspace_area[:] = [self._tile_grid]
        self._rebuild_tile_grid()
        if pn.state.location is not None:
            pn.state.location.param.update(search="")

    _ADMIN_DASHBOARD_ACTIONS = (
        {"label": "Edit", "icon": "edit"},
        {"label": "Rename", "icon": "drive_file_rename_outline"},
        {"label": "Delete", "icon": "delete"},
    )

    _VIEWER_DASHBOARD_ACTIONS = ()

    def _dashboard_actions(self, can_admin: bool) -> tuple:
        """Menu actions for a dashboard, gated by administration rights."""
        return self._ADMIN_DASHBOARD_ACTIONS if can_admin else self._VIEWER_DASHBOARD_ACTIONS

    def _dashboard_speed_dial_items(self, can_admin: bool) -> list[dict]:
        """SpeedDial items for a launcher dashboard card, gated by admin rights."""
        return [dict(action) for action in self._dashboard_actions(can_admin)]

    def _get_dashboard_menu_items(self) -> list[dict]:
        items = []
        dashboards = self.store.list_accessible(
            self._identity, default_allow=self._default_allow()
        )
        for d in dashboards:
            can_admin = self._can_administer_dashboard(d.dashboard_id)
            item = {
                "icon": "dashboard",
                "label": d.title,
                "path": f"{DASH_ROUTE_PREFIX}{d.dashboard_id}",
                "disable_link": True,
            }
            actions = self._dashboard_actions(can_admin)
            if actions:
                item["actions"] = list(actions)
            items.append(item)
        items.append(
            {
                "icon": "add",
                "label": "New Dashboard",
                "path": "__new_dashboard__",
                "disable_link": True,
                "actions": [{"label": "Create", "icon": "add", "inline": True}],
            }
        )
        return items

    def _dashboard_id_from_path(self, path: str) -> str | None:
        if path and path.startswith(DASH_ROUTE_PREFIX):
            return path[len(DASH_ROUTE_PREFIX) :].strip("/")
        return None

    def _on_action_edit(self, item):
        self._nav_drawer.open = False
        path = item.get("path", "")
        dashboard_id = self._dashboard_id_from_path(path)
        if not dashboard_id:
            return
        target_path = f"{DASH_ROUTE_PREFIX}{dashboard_id}"
        if self._dirty and self._current_dashboard is not None:
            self._pending_navigation = target_path
            self._unsaved_dialog.open = True
        else:
            pn.state.execute(partial(self._load_dashboard_edit, dashboard_id))

    async def _load_dashboard_edit(self, dashboard_id: str):
        await self._load_dashboard(dashboard_id, edit=True)

    @pn.io.hold()
    def _on_action_rename(self, item):
        path = item.get("path", "")
        dashboard_id = self._dashboard_id_from_path(path)
        if not dashboard_id:
            return
        self._dialog_name_input.param.update(
            value=item.get("label", ""), disabled=False, error_state=False, helper_text=""
        )
        self._dialog_context = {"action": "rename", "dashboard_id": dashboard_id}
        self._dialog.title = "Rename Dashboard"
        self._nav_drawer.open = False
        self._dialog.open = True

    @pn.io.hold()
    def _on_action_delete(self, item):
        path = item.get("path", "")
        dashboard_id = self._dashboard_id_from_path(path)
        if not dashboard_id:
            return
        self._dialog_name_input.param.update(value=item.get("label", ""), disabled=True)
        self._dialog_context = {"action": "delete", "dashboard_id": dashboard_id}
        self._dialog.title = "Delete Dashboard"
        self._nav_drawer.open = False
        self._dialog.open = True

    @pn.io.hold()
    def _on_action_create(self, item):
        self._open_create_dialog()

    @pn.io.hold()
    def _open_create_dialog(self):
        self._dialog_name_input.param.update(
            value="", disabled=False, error_state=False, helper_text=""
        )
        self._dialog_context = {"action": "create"}
        self._dialog.title = "Create Dashboard"
        self._dialog.open = True
        self._nav_drawer.open = False

    def _validate_dashboard_name(self, title: str) -> str | None:
        """Return an error message if the title is invalid, else None."""
        title = title.strip()
        if not title:
            return "Name cannot be empty."
        exclude_id = self._dialog_context.get("dashboard_id")
        if self.store.title_exists(self._user_id, title, exclude_id=exclude_id):
            return "A dashboard with this name already exists."
        return None

    def _on_dialog_name_changed(self, event):
        error = self._validate_dashboard_name(event.new)
        self._dialog_name_input.error_state = error is not None
        self._dialog_name_input.helper_text = error or ""

    @pn.io.hold()
    def _on_dialog_confirm(self, _event):
        ctx = self._dialog_context
        if not ctx:
            return
        action = ctx.get("action")
        if action in ("create", "rename"):
            error = self._validate_dashboard_name(self._dialog_name_input.value)
            if error:
                self._dialog_name_input.error_state = True
                self._dialog_name_input.helper_text = error
                return
        self._dialog.open = False
        if action == "create":
            t = self._dialog_name_input.value
            if t:
                self._create_new_dashboard(t)
        elif action == "rename":
            new_t = self._dialog_name_input.value
            did = ctx.get("dashboard_id", "")
            if new_t and did:
                self._rename_dashboard(did, new_t)
        elif action == "delete":
            did = ctx.get("dashboard_id", "")
            if did:
                self._delete_dashboard(did)
        self._dialog_name_input.param.update(disabled=False, error_state=False, helper_text="")
        self._dialog_context = {}

    def _build_dialog(self):
        self._dialog_name_input = pmui.TextInput(
            label="Name",
            sizing_mode="stretch_width",
        )
        self._dialog_name_input.param.watch(self._on_dialog_name_changed, "value_input")
        confirm_btn = pmui.Button(label="Confirm", color="primary")
        cancel_btn = pmui.Button(label="Cancel", color="light")
        confirm_btn.on_click(self._on_dialog_confirm)
        cancel_btn.on_click(lambda _: setattr(self._dialog, "open", False))
        self._dialog_context: dict = {}
        self._dialog = pmui.Dialog(
            objects=[
                pn.Column(
                    self._dialog_name_input,
                    pn.Row(confirm_btn, cancel_btn),
                    sizing_mode="stretch_width",
                )
            ],
            title="Dashboard",
            open=False,
            min_width=350,
        )

    def _build_share_dialog(self):
        """Build the (owner/admin-only) dashboard sharing dialog."""
        self._share_context: dict = {}
        common = dict(sizing_mode="stretch_width", solid=True, delete_button=True)
        self._share_allow_groups = pmui.MultiChoice(
            label="Allow groups", helper_text="Members of any listed group.", **common
        )
        self._share_allow_users = pmui.MultiChoice(
            label="Allow users", helper_text="OAuth logins or system users.", **common
        )
        self._share_deny_groups = pmui.MultiChoice(
            label="Deny groups", helper_text="Deny always wins.", **common
        )
        self._share_deny_users = pmui.MultiChoice(
            label="Deny users", helper_text="Deny always wins.", **common
        )
        self._share_widgets = (
            self._share_allow_groups,
            self._share_allow_users,
            self._share_deny_groups,
            self._share_deny_users,
        )
        confirm_btn = pmui.Button(label="Save sharing", color="primary")
        cancel_btn = pmui.Button(label="Cancel", color="light")
        confirm_btn.on_click(self._on_share_confirm)
        cancel_btn.on_click(lambda _: setattr(self._share_dialog, "open", False))
        self._share_dialog = pmui.Dialog(
            objects=[
                pn.Column(
                    pmui.Typography(
                        "Grant access by group or user. With no rules the project "
                        "default applies.",
                        variant="body2",
                        styles={"opacity": "0.7"},
                    ),
                    self._share_allow_groups,
                    self._share_allow_users,
                    self._share_deny_groups,
                    self._share_deny_users,
                    pn.Row(confirm_btn, cancel_btn),
                    sizing_mode="stretch_width",
                )
            ],
            title="Share Dashboard",
            open=False,
            min_width=420,
        )
        return self._share_dialog

    def _known_groups(self) -> list[str]:
        """Discoverable group names to offer in the sharing dialog."""
        groups: set[str] = set(self._identity.groups)
        if self.auth_config is not None:
            groups |= set(self.auth_config.admin_groups)
            for member_groups in self.auth_config.user_groups.values():
                groups |= set(member_groups)
        return sorted(groups)

    def _known_users(self) -> list[str]:
        """Discoverable user names to offer in the sharing dialog."""
        users: set[str] = set(self._identity.user_names)
        if self.auth_config is not None:
            users |= set(self.auth_config.user_groups)
        return sorted(users)

    @pn.io.hold()
    def _open_share_dialog(self, dashboard_id: str, title: str):
        if not self._can_administer_dashboard(dashboard_id):
            pn.state.notifications.error("You are not allowed to share this dashboard.")
            return
        model = self.store.load_for_access(
            self._identity, dashboard_id, default_allow=self._default_allow()
        )
        perm = model.permission if model else Permission()
        self._share_context = {"dashboard_id": dashboard_id}

        # Seed options from discoverable names, extended with any values already
        # stored on the permission so custom entries render (MultiChoice shows
        # out-of-option values as removable chips).
        known_groups = self._known_groups()
        known_users = self._known_users()
        for widget, options, selected in (
            (self._share_allow_groups, known_groups, perm.allow_groups),
            (self._share_allow_users, known_users, perm.allow_users),
            (self._share_deny_groups, known_groups, perm.deny_groups),
            (self._share_deny_users, known_users, perm.deny_users),
        ):
            widget.param.update(
                options=sorted(set(options) | set(selected)),
                value=sorted(selected),
            )

        self._share_dialog.title = f"Share “{title}”" if title else "Share Dashboard"
        self._share_dialog.open = True

    @pn.io.hold()
    def _on_share_confirm(self, _event):
        ctx = self._share_context
        dashboard_id = ctx.get("dashboard_id", "")
        if not dashboard_id:
            return
        if not self._can_administer_dashboard(dashboard_id):
            pn.state.notifications.error("You are not allowed to share this dashboard.")
            self._share_dialog.open = False
            return
        permission = Permission.from_spec(
            allow_groups=self._share_allow_groups.value,
            allow_users=self._share_allow_users.value,
            deny_groups=self._share_deny_groups.value,
            deny_users=self._share_deny_users.value,
        )
        self.store.set_permission(dashboard_id, permission)
        if self._current_dashboard and self._current_dashboard.dashboard_id == dashboard_id:
            self._current_dashboard.permission = permission
        self._share_dialog.open = False
        self._share_context = {}
        self._refresh_sidebar_dashboards()
        pn.state.notifications.success("Sharing updated.", duration=3000)

    def _build_unsaved_dialog(self):
        self._pending_navigation: str | None = None

        discard_btn = pmui.Button(label="Discard", color="danger", variant="outlined")
        save_btn = pmui.Button(label="Save & Continue", color="primary")
        stay_btn = pmui.Button(label="Cancel", color="light")

        def _on_discard(_event):
            self._unsaved_dialog.open = False
            self._dirty = False
            path = self._pending_navigation
            self._pending_navigation = None
            if path:
                self._navigate_to(path)

        def _on_save(_event):
            self._unsaved_dialog.open = False
            self._save_current_dashboard()
            path = self._pending_navigation
            self._pending_navigation = None
            if path:
                self._navigate_to(path)

        def _on_stay(_event):
            self._unsaved_dialog.open = False
            self._pending_navigation = None

        discard_btn.on_click(_on_discard)
        save_btn.on_click(_on_save)
        stay_btn.on_click(_on_stay)

        self._unsaved_dialog = pmui.Dialog(
            objects=[
                pn.Column(
                    pn.pane.Markdown("You have unsaved changes. What would you like to do?"),
                    pn.Row(save_btn, discard_btn, stay_btn),
                    sizing_mode="stretch_width",
                )
            ],
            title="Unsaved Changes",
            open=False,
            min_width=400,
        )
        return self._unsaved_dialog

    def _navigate_to(self, path: str):
        if pn.state.location is None:
            return
        pn.state.location.param.update(pathname=path, search="")
        self._sync_menu_active(path)
        pn.state.execute(self._load_page_layout)

    def _sync_menu_active(self, path: str):
        items = self._menu_list.items
        for si, section in enumerate(items):
            if section.get("path") == path:
                self._menu_list.active = (si,)
                return
            for pi, item in enumerate(section.get("items", [])):
                if item.get("path") == path:
                    self._menu_list.active = (si, pi)
                    return
        self._menu_list.active = None

    def _request_navigation(self, path: str):
        if self._dirty and self._current_dashboard is not None:
            self._pending_navigation = path
            self._unsaved_dialog.open = True
        else:
            self._navigate_to(path)

    _SECTION_ICONS: t.ClassVar[dict[str, str]] = {
        "components": "widgets",
        "pages": "description",
    }

    def _section_icon(self, section: str) -> str:
        """Pick a menu icon for a page section, keyed by its (normalized) name."""
        return self._SECTION_ICONS.get(section.replace("_", " ").lower(), "folder")

    def _build_nav_menu_items(self) -> list[dict]:
        """Assemble the nav tree: Home, page sections and the Custom Apps group."""
        sections: dict[str, list[RegistryEntry]] = {}
        for entry in self._accessible_page_entries().values():
            sections.setdefault(entry.section, []).append(entry)

        menu_items = [
            {
                "label": "Home",
                "icon": "home",
                "path": "/",
                "disable_link": True,
            },
        ]
        for section, section_apps in sorted(sections.items()):
            menu_items.append(
                {
                    "label": section.replace("_", " "),
                    "selectable": False,
                    "icon": self._section_icon(section),
                    "items": [
                        {
                            "icon": None,
                            "label": page_entry.title,
                            "path": page_entry.page_path,
                            "href": page_entry.page_path,
                            "disable_link": True,
                        }
                        for page_entry in sorted(section_apps, key=lambda e: e.name)
                    ],
                }
            )
        menu_items.append(
            {
                "label": "Custom Apps",
                "selectable": False,
                "icon": "dashboard_customize",
                "items": self._get_dashboard_menu_items(),
            }
        )
        return menu_items

    def _initial_menu_active(self, menu_items: list[dict]):
        """Index of the menu item matching the current pathname, if any."""
        current_path = pn.state.location.pathname if pn.state.location is not None else ""
        pathname = "/" + current_path.strip("/")
        for si, s in enumerate(menu_items):
            if s.get("path") == pathname:
                return (si,)
            for pi, p in enumerate(s.get("items", [])):
                if p.get("path") == pathname:
                    return (si, pi)
        return None

    def _on_nav_click(self, event):
        """Shared click handler for both the drawer MenuList and header MenuBar."""
        if "path" not in event or pn.state.location is None:
            return
        path = event["path"]
        if path == "__new_dashboard__":
            self._open_create_dialog()
            return
        if path == pn.state.location.pathname:
            if "edit=true" in (pn.state.location.search or ""):
                pn.state.location.param.update(search="")
                self._show_view_mode()
            return
        self._request_navigation(path)

    def _build_nav_menu(self):
        menu_items = self._build_nav_menu_items()

        self._menu_list = pmui.MenuList(
            items=menu_items,
            on_click=self._on_nav_click,
            dense=True,
            expanded=list(range(len(menu_items))),
            active=self._initial_menu_active(menu_items),
            width_policy="max",
        )

        self._menu_list.on_action("Edit", self._on_action_edit)
        self._menu_list.on_action("Rename", self._on_action_rename)
        self._menu_list.on_action("Delete", self._on_action_delete)
        self._menu_list.on_action("Create", self._on_action_create)

        return self._menu_list

    def _build_menu_bar_items(self) -> list[dict]:
        """MenuBar equivalent of the nav tree, with dashboard management submenus.

        MenuBar has no inline action buttons, so each dashboard is exposed as a
        submenu (Open plus, for administrators, Edit/Rename/Delete) and the
        management verbs are encoded via a ``nav_action`` key routed in
        :meth:`_on_menu_bar_click`.
        """
        sections: dict[str, list[RegistryEntry]] = {}
        for entry in self._accessible_page_entries().values():
            sections.setdefault(entry.section, []).append(entry)

        nav_items: list[dict] = [{"label": "Home", "icon": "home", "path": "/"}]
        for section, section_apps in sorted(sections.items()):
            nav_items.append(
                {
                    "label": section.replace("_", " "),
                    "icon": self._section_icon(section),
                    "items": [
                        {"label": page_entry.title, "path": page_entry.page_path}
                        for page_entry in sorted(section_apps, key=lambda e: e.name)
                    ],
                }
            )

        dashboards = self.store.list_accessible(
            self._identity, default_allow=self._default_allow()
        )
        dash_items: list[dict] = [
            {"label": "New Dashboard", "icon": "add", "path": "__new_dashboard__"},
        ]
        if dashboards:
            dash_items.append(None)
        for d in dashboards:
            path = f"{DASH_ROUTE_PREFIX}{d.dashboard_id}"
            entries = [{"label": "Open", "icon": "open_in_new", "path": path}]
            if self._can_administer_dashboard(d.dashboard_id):
                entries += [
                    {"label": "Edit", "icon": "edit", "path": path, "nav_action": "edit"},
                    {
                        "label": "Rename",
                        "icon": "drive_file_rename_outline",
                        "path": path,
                        "nav_action": "rename",
                        "title": d.title,
                    },
                    {
                        "label": "Delete",
                        "icon": "delete",
                        "path": path,
                        "nav_action": "delete",
                        "title": d.title,
                    },
                ]
            dash_items.append({"label": d.title, "icon": "dashboard", "items": entries})

        return [
            {"label": "Navigate", "icon": "menu", "items": nav_items},
            {"label": "Dashboards", "icon": "dashboard_customize", "items": dash_items},
        ]

    def _on_menu_bar_click(self, item):
        """Route a MenuBar click, dispatching dashboard management verbs."""
        action = item.get("nav_action") if isinstance(item, dict) else None
        if action is None:
            self._on_nav_click(item)
            return
        synthetic = {"path": item.get("path", ""), "label": item.get("title", "")}
        if action == "edit":
            self._on_action_edit(synthetic)
        elif action == "rename":
            self._on_action_rename(synthetic)
        elif action == "delete":
            self._on_action_delete(synthetic)

    def _build_nav_bar(self):
        """Build the header MenuBar and its quick-action icons (menubar variant)."""
        self._menu_bar = pmui.MenuBar(
            items=self._build_menu_bar_items(),
            on_click=self._on_menu_bar_click,
            color="default",
            margin=(0, 0, 0, 30),
            variant="outlined",
            sx={"border": "none", "boxShadow": "none"},
        )
        return self._menu_bar

    def _refresh_menu_bar(self):
        """Rebuild the header MenuBar items after the dashboard list changes."""
        if self._menu_bar is not None:
            self._menu_bar.items = self._build_menu_bar_items()

    def __panel__(self):
        """Render the app."""
        return self._page

    @classmethod
    def build_routes(
        cls,
        project_dir: str | pathlib.Path,
        registry: dict[str, RegistryEntry] | None = None,
        **params,
    ) -> dict[str, t.Any]:
        """Generate route mapping for pn.serve."""
        if registry is None:
            registry = build_registry(pathlib.Path(project_dir))

        def factory():
            return cls(registry=registry, **params)

        routes: dict[str, t.Any] = {
            "/": factory,
            COMPONENTS_ROUTE: factory,
            f"{DASH_ROUTE_PREFIX}[^/]+": factory,
        }
        for app_id, v in registry.items():
            if v.metadata.page:
                routes[f"/{app_id}"] = factory
        return routes

auth_config = param.ClassSelector(class_=AuthConfig, doc='\n Project-level authorization configuration controlling group discovery,\n the admin groups and the default access policy.') class-attribute instance-attribute

breakpoints = param.List(default=[768, 1200], doc='Responsive breakpoints for the tile grid.') class-attribute instance-attribute

configure_layout = param.Callable(default=None, doc='\n Optional callback invoked on every navigation with\n (app, content, route). Use it to set `app.sidebar` and\n `app.contextbar` for the page currently being served.') class-attribute instance-attribute

contextbar = Children(default=[], doc='Items prepended to the contextbar.') class-attribute instance-attribute

contextbar_open = param.Boolean(default=False, doc='Whether the contextbar is open.') class-attribute instance-attribute

home_dashboard = param.String(default=None, doc="\n Dashboard shown on the homepage ('/'). Accepts a dashboard id or\n title. When unset, the homepage shows the dashboard grid launcher.") class-attribute instance-attribute

nav_variant = param.Selector(default='drawer', objects=['drawer', 'menubar'], doc="\n Where the navigation menu is rendered. 'drawer' docks a MenuList in a\n right-hand drawer; 'menubar' places a MenuBar in the page header with\n quick-action icons alongside it.") class-attribute instance-attribute

page_options = param.Dict(default={}, doc="\n Extra keyword arguments passed through to the underlying\n `panel_material_ui.Page`, overriding the app's own defaults.") class-attribute instance-attribute

project_dir = param.Path(doc='Path to the project directory.') class-attribute instance-attribute

sidebar = Children(default=[], doc='Items prepended to the sidebar.') class-attribute instance-attribute

store = param.ClassSelector(class_=DashboardStore, doc='DashboardStore instance for persistence.') class-attribute instance-attribute

title = param.String(default='FlowDash', doc='Application title shown in the browser tab.') class-attribute instance-attribute

build_routes(project_dir, registry=None, **params) classmethod

Generate route mapping for pn.serve.

Source code in src/panel_flowdash/app.py
@classmethod
def build_routes(
    cls,
    project_dir: str | pathlib.Path,
    registry: dict[str, RegistryEntry] | None = None,
    **params,
) -> dict[str, t.Any]:
    """Generate route mapping for pn.serve."""
    if registry is None:
        registry = build_registry(pathlib.Path(project_dir))

    def factory():
        return cls(registry=registry, **params)

    routes: dict[str, t.Any] = {
        "/": factory,
        COMPONENTS_ROUTE: factory,
        f"{DASH_ROUTE_PREFIX}[^/]+": factory,
    }
    for app_id, v in registry.items():
        if v.metadata.page:
            routes[f"/{app_id}"] = factory
    return routes

Permission dataclass

An allow/deny rule set evaluated against an :class:Identity.

All four fields match either the resolved user (OAuth login or system user) or one of the identity's groups. An empty Permission declares no constraints and defers entirely to the caller's default policy.

Source code in src/panel_flowdash/auth.py
@dataclass(frozen=True)
class Permission:
    """An allow/deny rule set evaluated against an :class:`Identity`.

    All four fields match either the resolved ``user`` (OAuth login *or* system
    user) or one of the identity's ``groups``. An empty ``Permission`` declares
    no constraints and defers entirely to the caller's default policy.
    """

    allow_users: frozenset[str] = field(default_factory=frozenset)
    allow_groups: frozenset[str] = field(default_factory=frozenset)
    deny_users: frozenset[str] = field(default_factory=frozenset)
    deny_groups: frozenset[str] = field(default_factory=frozenset)

    @property
    def is_empty(self) -> bool:
        """Whether the permission declares no allow or deny rules."""
        return not (self.allow_users or self.allow_groups or self.deny_users or self.deny_groups)

    @classmethod
    def from_spec(
        cls,
        *,
        allow_users: Iterable[str] | None = None,
        allow_groups: Iterable[str] | None = None,
        deny_users: Iterable[str] | None = None,
        deny_groups: Iterable[str] | None = None,
    ) -> Permission:
        """Build a :class:`Permission` from loosely-typed iterables."""
        return cls(
            allow_users=frozenset(allow_users or ()),
            allow_groups=frozenset(allow_groups or ()),
            deny_users=frozenset(deny_users or ()),
            deny_groups=frozenset(deny_groups or ()),
        )

    def to_dict(self) -> dict[str, list[str]]:
        """Serialize to sorted lists for JSON persistence."""
        return {
            "allow_users": sorted(self.allow_users),
            "allow_groups": sorted(self.allow_groups),
            "deny_users": sorted(self.deny_users),
            "deny_groups": sorted(self.deny_groups),
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any] | None) -> Permission:
        """Deserialize from a (possibly ``None`` or partial) mapping."""
        data = data or {}
        return cls.from_spec(
            allow_users=data.get("allow_users"),
            allow_groups=data.get("allow_groups"),
            deny_users=data.get("deny_users"),
            deny_groups=data.get("deny_groups"),
        )

allow_groups = field(default_factory=frozenset) class-attribute instance-attribute

allow_users = field(default_factory=frozenset) class-attribute instance-attribute

deny_groups = field(default_factory=frozenset) class-attribute instance-attribute

deny_users = field(default_factory=frozenset) class-attribute instance-attribute

is_empty property

Whether the permission declares no allow or deny rules.

from_dict(data) classmethod

Deserialize from a (possibly None or partial) mapping.

Source code in src/panel_flowdash/auth.py
@classmethod
def from_dict(cls, data: dict[str, Any] | None) -> Permission:
    """Deserialize from a (possibly ``None`` or partial) mapping."""
    data = data or {}
    return cls.from_spec(
        allow_users=data.get("allow_users"),
        allow_groups=data.get("allow_groups"),
        deny_users=data.get("deny_users"),
        deny_groups=data.get("deny_groups"),
    )

from_spec(*, allow_users=None, allow_groups=None, deny_users=None, deny_groups=None) classmethod

Build a :class:Permission from loosely-typed iterables.

Source code in src/panel_flowdash/auth.py
@classmethod
def from_spec(
    cls,
    *,
    allow_users: Iterable[str] | None = None,
    allow_groups: Iterable[str] | None = None,
    deny_users: Iterable[str] | None = None,
    deny_groups: Iterable[str] | None = None,
) -> Permission:
    """Build a :class:`Permission` from loosely-typed iterables."""
    return cls(
        allow_users=frozenset(allow_users or ()),
        allow_groups=frozenset(allow_groups or ()),
        deny_users=frozenset(deny_users or ()),
        deny_groups=frozenset(deny_groups or ()),
    )

to_dict()

Serialize to sorted lists for JSON persistence.

Source code in src/panel_flowdash/auth.py
def to_dict(self) -> dict[str, list[str]]:
    """Serialize to sorted lists for JSON persistence."""
    return {
        "allow_users": sorted(self.allow_users),
        "allow_groups": sorted(self.allow_groups),
        "deny_users": sorted(self.deny_users),
        "deny_groups": sorted(self.deny_groups),
    }

RegistryEntry dataclass

A registered component/page with its metadata.

Source code in src/panel_flowdash/registry.py
@dataclass
class RegistryEntry:
    """A registered component/page with its metadata."""

    app_id: str
    section: str
    name: str
    page_path: str
    module_name: str
    metadata: PanelAppMetadata
    module_path: pathlib.Path | None = None
    app: Any = None

    @property
    def title(self) -> str:
        """Human-readable title."""
        return self.metadata.title or self.name.replace("_", " ")

    def load(self) -> Any:
        """Import the module and return the app object.

        Caches the result on ``self.app``.  Raises on import failure.
        """
        if self.app is not None:
            return self.app
        module = importlib.import_module(self.module_name)
        app = getattr(module, "app", None)
        if app is None:
            raise ImportError(f"Module '{self.module_name}' has no 'app' export.")
        # Refresh metadata from the live object (decorators may carry richer info
        # e.g. config_schema / config_editor that AST cannot capture).
        object.__setattr__(self, "app", app)
        live_metadata = PanelAppMetadata.from_app(app)
        # Only replace if the live decorator actually produced a non-default result
        # (guards against bare Viewer classes with no @register decorator).
        if live_metadata != PanelAppMetadata():
            object.__setattr__(self, "metadata", live_metadata)
        return app

app = None class-attribute instance-attribute

app_id instance-attribute

metadata instance-attribute

module_name instance-attribute

module_path = None class-attribute instance-attribute

name instance-attribute

page_path instance-attribute

section instance-attribute

title property

Human-readable title.

load()

Import the module and return the app object.

Caches the result on self.app. Raises on import failure.

Source code in src/panel_flowdash/registry.py
def load(self) -> Any:
    """Import the module and return the app object.

    Caches the result on ``self.app``.  Raises on import failure.
    """
    if self.app is not None:
        return self.app
    module = importlib.import_module(self.module_name)
    app = getattr(module, "app", None)
    if app is None:
        raise ImportError(f"Module '{self.module_name}' has no 'app' export.")
    # Refresh metadata from the live object (decorators may carry richer info
    # e.g. config_schema / config_editor that AST cannot capture).
    object.__setattr__(self, "app", app)
    live_metadata = PanelAppMetadata.from_app(app)
    # Only replace if the live decorator actually produced a non-default result
    # (guards against bare Viewer classes with no @register decorator).
    if live_metadata != PanelAppMetadata():
        object.__setattr__(self, "metadata", live_metadata)
    return app

_DASHBOARD_ACTION_TYPE

Bases: TypedDict

Source code in src/panel_flowdash/app.py
class _DASHBOARD_ACTION_TYPE(t.TypedDict):
    label: str
    icon: str

icon instance-attribute

label instance-attribute

build_component_specs(registry)

Build specs for all component-enabled entries in a registry.

Source code in src/panel_flowdash/component_spec.py
def build_component_specs(
    registry: dict[str, RegistryEntry],
) -> dict[str, ComponentSpec]:
    """Build specs for all component-enabled entries in a registry."""
    specs = {}
    for app_id, entry in registry.items():
        if entry.metadata.component:
            specs[app_id] = build_component_spec(entry)
    return specs

build_registry(project_dir)

Scan project_dir for page/component modules without importing them.

Reads each .py file with the AST to extract @register metadata. Modules are not imported at this stage; each RegistryEntry.app is None until RegistryEntry.load() is called.

Source code in src/panel_flowdash/registry.py
def build_registry(project_dir: Path) -> dict[str, RegistryEntry]:
    """Scan *project_dir* for page/component modules without importing them.

    Reads each ``.py`` file with the AST to extract ``@register`` metadata.
    Modules are **not** imported at this stage; each ``RegistryEntry.app`` is
    ``None`` until ``RegistryEntry.load()`` is called.
    """
    registry: dict[str, RegistryEntry] = {}

    for section_dir in sorted(project_dir.glob("*")):
        if not section_dir.is_dir() or section_dir.name.startswith(("_", ".")):
            continue
        section = section_dir.name
        for module_path in sorted(section_dir.glob("*.py")):
            if module_path.name.startswith("_"):
                continue

            source = module_path.read_text(encoding="utf-8")
            kwargs = _extract_register_kwargs(source)
            if kwargs is None:
                # No @register call found — skip silently (same as before).
                continue

            # Defaults that match PanelAppMetadata
            page = kwargs.get("page", True)
            component = kwargs.get("component", False)
            if not page and not component:
                continue

            metadata = PanelAppMetadata(
                page=bool(page),
                component=bool(component),
                sidebar=bool(kwargs.get("sidebar", False)),
                title=kwargs.get("title"),
                icon=kwargs.get("icon"),
                description=kwargs.get("description"),
                tags=list(kwargs.get("tags") or []),
                default_size=kwargs.get("default_size"),
                min_size=kwargs.get("min_size"),
                max_size=kwargs.get("max_size"),
                singleton=bool(kwargs.get("singleton", False)),
                provides=list(kwargs.get("provides") or []),
                requires=list(kwargs.get("requires") or []),
                config=list(kwargs.get("config") or []),
                allow_users=list(kwargs.get("allow_users") or []),
                allow_groups=list(kwargs.get("allow_groups") or []),
                deny_users=list(kwargs.get("deny_users") or []),
                deny_groups=list(kwargs.get("deny_groups") or []),
            )

            module_name = ".".join(module_path.relative_to(project_dir).with_suffix("").parts)
            app_id = f"{section}/{module_path.stem}"
            registry[app_id] = RegistryEntry(
                app_id=app_id,
                section=section,
                name=module_path.stem,
                page_path=f"/{app_id}",
                module_name=module_name,
                metadata=metadata,
                module_path=module_path,
                app=None,
            )

    return registry

build_session_state_class(registry)

Build a Parameterized subclass with one param per declared state key.

Scans the registry for all provides and requires keys and creates a dynamic class whose parameters represent shared session state.

Source code in src/panel_flowdash/session_state.py
def build_session_state_class(
    registry: dict[str, RegistryEntry],
) -> type[param.Parameterized]:
    """Build a Parameterized subclass with one param per declared state key.

    Scans the registry for all `provides` and `requires` keys and creates
    a dynamic class whose parameters represent shared session state.
    """
    state_keys: dict[str, Any] = {}

    for entry in registry.values():
        for key in entry.metadata.provides:
            if isinstance(key, str) and key not in state_keys:
                state_keys[key] = None
            elif isinstance(key, dict):
                k = key.get("key", "")
                if k and k not in state_keys:
                    state_keys[k] = None
        for req in entry.metadata.requires:
            if isinstance(req, str):
                if req not in state_keys:
                    state_keys[req] = None
            elif isinstance(req, dict):
                k = req.get("key", "")
                if k and k not in state_keys:
                    state_keys[k] = req.get("fallback")

    params = {
        key: param.Parameter(default=default, allow_None=True)
        for key, default in state_keys.items()
    }

    return type("SessionState", (param.Parameterized,), params)

check_requirements(state, requires)

Check which required keys are unsatisfied on the given state instance.

Returns a list of dicts describing each unsatisfied requirement. An empty list means all requirements are met.

Source code in src/panel_flowdash/session_state.py
def check_requirements(state: param.Parameterized, requires: list) -> list[dict]:
    """Check which required keys are unsatisfied on the given state instance.

    Returns a list of dicts describing each unsatisfied requirement.
    An empty list means all requirements are met.
    """
    unsatisfied = []
    for req in requires:
        if isinstance(req, str):
            key, required, blocking, fallback = req, True, True, None
        else:
            key = req.get("key", "")
            required = req.get("required", True)
            blocking = req.get("blocking", True)
            fallback = req.get("fallback")

        if not key or not required:
            continue

        value = getattr(state, key, None)
        if value is None:
            unsatisfied.append({"key": key, "blocking": blocking, "fallback": fallback})

    return unsatisfied

is_authorized(permission, identity, *, default_allow=True, owner=None)

Evaluate permission against identity.

Order of precedence:

  1. A matching deny_users/deny_groups rule denies access (deny always wins, even for the owner).
  2. The owner, if given and matching, is allowed.
  3. Any allow_* rule present: allowed iff the identity matches at least one of them.
  4. No allow/deny rules at all: fall back to default_allow.
Source code in src/panel_flowdash/auth.py
def is_authorized(
    permission: Permission | None,
    identity: Identity,
    *,
    default_allow: bool = True,
    owner: str | None = None,
) -> bool:
    """Evaluate *permission* against *identity*.

    Order of precedence:

    1. A matching ``deny_users``/``deny_groups`` rule denies access (deny always
       wins, even for the owner).
    2. The *owner*, if given and matching, is allowed.
    3. Any ``allow_*`` rule present: allowed iff the identity matches at least
       one of them.
    4. No allow/deny rules at all: fall back to *default_allow*.
    """
    if permission is None:
        permission = Permission()

    if permission.deny_users and identity.is_user(permission.deny_users):
        return False
    if permission.deny_groups and identity.in_groups(permission.deny_groups):
        return False

    if owner is not None and owner in identity.user_names:
        return True

    if permission.allow_users or permission.allow_groups:
        if permission.allow_users and identity.is_user(permission.allow_users):
            return True
        if permission.allow_groups and identity.in_groups(permission.allow_groups):
            return True
        return False

    return default_allow

resolve_identity(auth_config=None)

Resolve the current session's :class:Identity.

Prefers the OAuth login (pn.state.user) when a real provider populated it; otherwise falls back to the system user, then to "anonymous". Groups are the union of claim-derived groups, the static user_groups mapping and any dynamic resolve_groups callback.

Source code in src/panel_flowdash/auth.py
def resolve_identity(auth_config: AuthConfig | None = None) -> Identity:
    """Resolve the current session's :class:`Identity`.

    Prefers the OAuth login (``pn.state.user``) when a real provider populated
    it; otherwise falls back to the system user, then to ``"anonymous"``.
    Groups are the union of claim-derived groups, the static ``user_groups``
    mapping and any dynamic ``resolve_groups`` callback.
    """
    auth_config = auth_config or AuthConfig()

    oauth_user = pn.state.user or None
    # Panel sets pn.state.user to a non-None placeholder when no provider is
    # configured; treat the well-known anonymous sentinel as "no OAuth user".
    if oauth_user in (ANONYMOUS_USER, "anonymous"):
        oauth_user = None

    system_user = _system_user()
    user = oauth_user or system_user or ANONYMOUS_USER

    user_info = dict(pn.state.user_info or {})
    groups = _groups_from_claims(user_info, auth_config.group_claims)

    for name in (user, oauth_user, system_user):
        if name and name in auth_config.user_groups:
            groups |= set(auth_config.user_groups[name])

    identity = Identity(
        user=user,
        oauth_user=oauth_user,
        system_user=system_user,
        groups=frozenset(groups),
        user_info=user_info,
    )

    if auth_config.resolve_groups is not None:
        try:
            extra = auth_config.resolve_groups(identity)
        except Exception:
            logger.exception("resolve_groups callback failed")
            extra = None
        if extra:
            groups |= set(extra)
            identity = Identity(
                user=user,
                oauth_user=oauth_user,
                system_user=system_user,
                groups=frozenset(groups),
                user_info=user_info,
            )

    return identity