Source code for labfrog.request_state
# SPDX-FileCopyrightText: 2025 Helmholtz-Zentrum Dresden-Rossendorf e.V (HZDR)
# SPDX-License-Identifier: Apache-2.0
"""Compatibility-focused request/session state helpers for dynamic forms.
This version intentionally stays close to the current LabFrog helper contract
and session semantics while still centralizing repeated route logic.
"""
from __future__ import annotations
import logging
import time
from collections.abc import Iterable
from flask import current_app, g, request, session
from labfrog.db import _app_cache_get, ensure_default_form_documents, get_db
from labfrog.form_factory import create_dynamic_form
from labfrog.forms import _get_header_choice, _set_header_choice
from labfrog.helpers.choices import ordered_unique_strings
from labfrog.helpers.demo import apply_demo_defaults
from labfrog.helpers.form_lookups import (
_request_cache_bucket as request_cache_bucket,
)
from labfrog.helpers.form_lookups import (
custom_field_documents,
diagnostic_catalog_docs,
diagnostic_choice_docs,
)
from labfrog.helpers.layout import (
_ensure_shot_identity_fields,
_fallback_required_fields_for_mode,
_known_layout_fields,
apply_campaign_layout,
autofix_header_doc_source,
available_campaign_names,
campaign_is_unset,
get_campaign_header_doc,
get_effective_campaign_layout_doc,
get_effective_layout_doc,
get_source_layout_header_doc,
organize_layout_filter,
paired_mode,
required_identity_fields_for_mode,
resolve_layout_selections,
)
FORM_SESSION_BOOTSTRAP_CONFIG: dict[str, dict[str, bool | str]] = {
"add_entry.add_entry": {
"startup_session_key": "startup_executed_tab1",
"ensure_defaults": True,
"initialize_default_campaign": True,
"apply_demo_defaults_on_startup": True,
},
"diagnostic_options.diagnostic_options": {
"startup_session_key": "startup_executed_tab1",
"ensure_defaults": True,
"initialize_default_campaign": True,
"apply_demo_defaults_on_startup": True,
},
"diagnostic_definitions.diagnostic_definitions": {
"startup_session_key": "startup_executed_tab1",
"ensure_defaults": True,
"initialize_default_campaign": True,
"apply_demo_defaults_on_startup": True,
},
"edit_entry.edit_entry": {
"startup_session_key": "startup_executed_tab2",
"ensure_defaults": False,
"initialize_default_campaign": False,
"apply_demo_defaults_on_startup": False,
},
"field_selection.field_selection": {
"startup_session_key": "init_template_form",
"ensure_defaults": True,
"initialize_default_campaign": True,
"apply_demo_defaults_on_startup": True,
},
"new_parameter.new_parameter": {
"startup_session_key": "init_template_form",
"ensure_defaults": True,
"initialize_default_campaign": True,
"apply_demo_defaults_on_startup": True,
},
"search_records.search_records": {
"startup_session_key": "init_template_form",
"ensure_defaults": True,
"initialize_default_campaign": True,
"apply_demo_defaults_on_startup": True,
},
}
def _diagnostic_cache_ttl_seconds() -> float:
value = current_app.config.get("DIAGNOSTIC_CACHE_SECONDS", 10.0)
try:
return max(float(value), 0.0)
except (TypeError, ValueError):
return 10.0
def _layout_doc_cache_ttl_seconds() -> float:
value = current_app.config.get("LAYOUT_DOC_CACHE_SECONDS", 10.0)
try:
return max(float(value), 0.0)
except (TypeError, ValueError):
return 10.0
def _header_doc_cache_ttl_seconds() -> float:
value = current_app.config.get("HEADER_DOC_CACHE_SECONDS", 10.0)
try:
return max(float(value), 0.0)
except (TypeError, ValueError):
return 10.0
def _campaign_layout_sync_ttl_seconds() -> float:
value = current_app.config.get("CAMPAIGN_LAYOUT_SYNC_SECONDS", 30.0)
try:
return max(float(value), 0.0)
except (TypeError, ValueError):
return 30.0
[docs]
def ensure_campaign_layout_session_state(
mode: str,
campaign: str | None,
*,
parameters_collection=None,
force: bool = False,
hydrate_diagnostics: bool = False,
) -> bool:
"""Synchronize campaign layout/header session state only when needed."""
collection = (
parameters_collection
if parameters_collection is not None
else get_db()[current_app.config["MONGODB_COLLECTION_FOR_SETTINGS"]]
)
normalized_campaign = campaign or ""
state_key = f"_labfrog_campaign_layout_sync_{mode}"
existing_state = session.get(state_key)
now = time.monotonic()
last_synced_at = 0.0
if isinstance(existing_state, dict):
try:
last_synced_at = float(existing_state.get("synced_at", 0.0) or 0.0)
except (TypeError, ValueError):
last_synced_at = 0.0
needs_sync = force or not isinstance(existing_state, dict)
if not needs_sync and existing_state.get("campaign") != normalized_campaign:
needs_sync = True
if not needs_sync and "selected_layout" not in session:
needs_sync = True
if not needs_sync and "SELECTED_FIELDS_LIST" not in session:
needs_sync = True
if not needs_sync:
ttl_seconds = _campaign_layout_sync_ttl_seconds()
if ttl_seconds > 0 and now - last_synced_at >= ttl_seconds:
needs_sync = True
if not needs_sync:
return False
apply_campaign_layout(
mode,
campaign,
collection,
hydrate_diagnostics=hydrate_diagnostics,
)
sync_campaign_header_choice(mode, campaign, parameters_collection=collection)
session[state_key] = {"campaign": normalized_campaign, "synced_at": now}
return True
def _diagnostic_catalog_docs() -> list[dict]:
"""Return the cached diagnostic catalog. See ``form_lookups``."""
return diagnostic_catalog_docs()
def _diagnostic_choice_docs(collection_name: str) -> list[dict]:
"""Return one diagnostic's cached preset choices. See ``form_lookups``."""
return diagnostic_choice_docs(collection_name)
[docs]
def invalidate_diagnostic_choice_caches(*, collection_name: str | None = None) -> None:
"""Drop cached diagnostic preset choices so add/edit forms refresh immediately."""
request_cache = request_cache_bucket("_labfrog_diagnostic_choice_cache")
if collection_name:
request_cache.pop(collection_name, None)
else:
request_cache.clear()
app_ext = getattr(current_app, "extensions", None)
if not isinstance(app_ext, dict):
return
choice_cache = app_ext.get("labfrog_diagnostic_choice_cache")
if isinstance(choice_cache, dict):
if collection_name:
keys_to_remove = [
cache_key
for cache_key in list(choice_cache.keys())
if isinstance(cache_key, tuple)
and len(cache_key) >= 2
and cache_key[1] == collection_name
]
for cache_key in keys_to_remove:
choice_cache.pop(cache_key, None)
else:
choice_cache.clear()
[docs]
def get_custom_options_collection_name() -> str | None:
if current_app.config.get("USE_CUSTOM_OPTIONS"):
return current_app.config["CUSTOM_OPTIONS_COLLECTION"].replace('"', "")
return None
[docs]
def choose_default_campaign_name() -> str:
selected = session.get("selected_campaign")
if not campaign_is_unset(selected):
return selected
configured = str(current_app.config.get("DEMO_DEFAULT_CAMPAIGN", "") or "").strip()
if configured:
return configured
cache = current_app.extensions.get("labfrog_mediawiki_cache", {}) or {}
cached_campaigns = [
name
for name in (cache.get("campaign_dict") or {}).keys()
if name and name != "NONE"
]
if cached_campaigns:
return cached_campaigns[0]
available = [name for name in available_campaign_names() if name and name != "NONE"]
return available[0] if available else "NONE"
def _load_default_layout_doc(parameters_collection, mode: str):
return parameters_collection.find_one({"layout_name": "DEFAULT", "mode": mode})
[docs]
def get_default_layout_doc(parameters_collection, mode: str):
cache = request_cache_bucket("_labfrog_default_layout_doc_cache")
key = (parameters_collection.name, mode)
if key not in cache:
cache[key] = _load_default_layout_doc(parameters_collection, mode)
return cache[key]
[docs]
def get_active_layout_doc(
parameters_collection,
selected_layout: str,
mode: str,
*,
campaign: str | None = None,
) -> dict | None:
"""Return the selected layout doc, falling back to the mode-specific default."""
cache_key = (
current_app.config.get("MONGODB_DATABASE"),
parameters_collection.name,
selected_layout,
mode,
campaign,
)
return _app_cache_get(
"labfrog_active_layout_doc_cache",
cache_key,
lambda: (
get_effective_layout_doc(
parameters_collection,
selected_layout,
mode,
campaign=campaign,
)
or get_default_layout_doc(parameters_collection, mode)
),
ttl_seconds=_layout_doc_cache_ttl_seconds(),
)
def _known_diagnostic_display_names() -> set[str]:
"""Return the current diagnostic display names from the database."""
return {
doc.get("DisplayName")
for doc in _diagnostic_catalog_docs()
if isinstance(doc.get("DisplayName"), str) and doc.get("DisplayName")
}
[docs]
def sync_campaign_header_choice(
mode: str, campaign: str | None, *, parameters_collection=None
) -> None:
collection = (
parameters_collection
if parameters_collection is not None
else get_db()[current_app.config["MONGODB_COLLECTION_FOR_SETTINGS"]]
)
source_layout_key = f"selected_source_layout_{mode}"
sibling_source_layout_key = f"selected_source_layout_{paired_mode(mode)}"
if campaign_is_unset(campaign):
session[source_layout_key] = "DEFAULT"
session[sibling_source_layout_key] = "DEFAULT"
session["selected_layout"] = "DEFAULT"
_set_header_choice(mode, "DEFAULT")
return
field_layout_doc = get_effective_campaign_layout_doc(collection, mode, campaign)
source_layout_name = (field_layout_doc or {}).get("layout_name") or "DEFAULT"
doc = autofix_header_doc_source(
_cached_campaign_header_doc(
mode, campaign, source_layout_name=source_layout_name
),
source_layout_name,
mode=mode,
) or _cached_source_layout_header_doc(mode, source_layout_name)
resolved_source_layout = (
(doc or {}).get("source_layout_name")
if isinstance((doc or {}).get("source_layout_name"), str)
else ""
)
resolved_source_layout = (
resolved_source_layout.strip() if resolved_source_layout else source_layout_name
)
session[source_layout_key] = resolved_source_layout
session[sibling_source_layout_key] = resolved_source_layout
session["selected_layout"] = resolved_source_layout
_set_header_choice(
mode, (doc or {}).get("layout_name") or (doc or {}).get("name") or "DEFAULT"
)
def _find_header_doc_from_settings_or_legacy(mode: str, selected_header: str):
if not selected_header or selected_header == "DEFAULT":
return None
db = get_db()
collection_name = current_app.config["MONGODB_COLLECTION_FOR_SETTINGS"]
for mode_filter in (
{"mode": mode},
{"mode": {"$exists": False}},
{"mode": paired_mode(mode)},
):
doc = db[collection_name].find_one({
"$and": [
{"layout_name": selected_header, **mode_filter},
organize_layout_filter(),
]
})
if doc:
return doc
if "organize_form" in db.list_collection_names():
for mode_filter in (
{"mode": mode},
{"mode": {"$exists": False}},
{"mode": paired_mode(mode)},
):
doc = db["organize_form"].find_one({"name": selected_header, **mode_filter})
if doc:
return doc
return None
def _cached_header_doc_from_settings_or_legacy(mode: str, selected_header: str):
cache_key = (
current_app.config.get("MONGODB_DATABASE"),
current_app.config.get("MONGODB_COLLECTION_FOR_SETTINGS"),
mode,
selected_header,
)
return _app_cache_get(
"labfrog_header_doc_by_name_cache",
cache_key,
lambda: _find_header_doc_from_settings_or_legacy(mode, selected_header),
ttl_seconds=_header_doc_cache_ttl_seconds(),
)
def _cached_campaign_header_doc(
mode: str,
campaign: str | None,
*,
source_layout_name: str | None = None,
):
cache_key = (
current_app.config.get("MONGODB_DATABASE"),
current_app.config.get("MONGODB_COLLECTION_FOR_SETTINGS"),
mode,
campaign or "",
source_layout_name or "",
)
return _app_cache_get(
"labfrog_campaign_header_doc_cache",
cache_key,
lambda: get_campaign_header_doc(
mode, campaign, source_layout_name=source_layout_name
),
ttl_seconds=_header_doc_cache_ttl_seconds(),
)
def _cached_source_layout_header_doc(mode: str, source_layout_name: str | None):
cache_key = (
current_app.config.get("MONGODB_DATABASE"),
current_app.config.get("MONGODB_COLLECTION_FOR_SETTINGS"),
mode,
source_layout_name or "",
)
return _app_cache_get(
"labfrog_source_layout_header_doc_cache",
cache_key,
lambda: get_source_layout_header_doc(mode, source_layout_name),
ttl_seconds=_header_doc_cache_ttl_seconds(),
)
[docs]
def get_active_header_doc(mode: str):
selected_campaign = session.get("selected_campaign")
selected_layout = (
session.get("selected_layout")
or session.get(f"selected_source_layout_{mode}")
or session.get(f"selected_source_layout_{paired_mode(mode)}")
or "DEFAULT"
)
selected_header = _get_header_choice(mode)
doc = _cached_header_doc_from_settings_or_legacy(mode, selected_header)
doc = autofix_header_doc_source(doc, selected_layout, mode=mode)
doc_source_layout = (doc or {}).get("source_layout_name")
if (
doc
and selected_layout not in {"", "DEFAULT"}
and doc_source_layout
and doc_source_layout != selected_layout
):
doc = None
if not doc:
doc = autofix_header_doc_source(
_cached_campaign_header_doc(
mode, selected_campaign, source_layout_name=selected_layout
),
selected_layout,
mode=mode,
) or _cached_source_layout_header_doc(mode, selected_layout)
if doc:
_set_header_choice(mode, doc.get("layout_name") or doc.get("name") or "DEFAULT")
return doc
if selected_header and selected_header != "DEFAULT":
_set_header_choice(mode, "DEFAULT")
return None
[docs]
def flatten_layout_fields(layout_doc: dict | None) -> list[str]:
if not layout_doc:
return []
fields: list[str] = []
sections = layout_doc.get("field_sections_dict") or {}
for value in sections.values():
if isinstance(value, dict):
fields.extend(value.get("fields", []) or [])
else:
fields.extend(value or [])
return fields
[docs]
def refresh_diagnostic_choice_cache(visible_fields: list[str]) -> None:
diagnostics = _diagnostic_catalog_docs()
tooltip_by_name = {
diag.get("DisplayName"): diag.get("Tooltip", "")
for diag in diagnostics
if isinstance(diag.get("DisplayName"), str) and diag.get("DisplayName")
}
diag_names = {}
tooltips = {}
for collection_name in [
field for field in visible_fields if field in tooltip_by_name
]:
docs = _diagnostic_choice_docs(collection_name)
diag_names[collection_name] = [doc["name"] for doc in docs if "name" in doc]
tooltip = tooltip_by_name.get(collection_name, "")
tooltips[collection_name] = [tooltip] if tooltip else []
g.DIAG_NAMES = diag_names
g.TOOLTIPS = tooltips
g.DIAG_NAMES_VISIBLE = [field for field in visible_fields if field in diag_names]
[docs]
def load_custom_field_documents(
fields_collection=None,
) -> tuple[list[dict], dict[str, dict]]:
"""Return the request-cached custom-field docs and their field_name lookup."""
return custom_field_documents(fields_collection)
[docs]
def invalidate_custom_field_doc_cache(*, collection_name: str | None = None) -> None:
"""Invalidate cached custom-field metadata for the current request."""
cache = request_cache_bucket("_labfrog_custom_field_docs_cache")
if collection_name:
cache.pop(collection_name, None)
else:
cache.clear()
metadata_cache = request_cache_bucket("_labfrog_form_runtime_metadata_cache")
if collection_name:
for cache_key in list(metadata_cache.keys()):
if (
isinstance(cache_key, tuple)
and len(cache_key) >= 2
and cache_key[1] == collection_name
):
metadata_cache.pop(cache_key, None)
else:
metadata_cache.clear()
[docs]
def bootstrap_form_session_state(
*,
startup_session_key: str,
ensure_defaults: bool = False,
initialize_default_campaign: bool = False,
apply_demo_defaults_on_startup: bool = False,
) -> bool:
if session.get(startup_session_key, False):
return False
if ensure_defaults:
ensure_default_form_documents(include_organize_form=False)
mode = session.get("mode", "shot")
parameters_collection = get_db()[
current_app.config["MONGODB_COLLECTION_FOR_SETTINGS"]
]
if apply_demo_defaults_on_startup:
apply_demo_defaults(mode, parameters_collection)
base_layout_doc = get_default_layout_doc(parameters_collection, mode)
always_include = list((base_layout_doc or {}).get("always_include", []))
active_layout_doc = (
get_active_layout_doc(
parameters_collection,
session.get("selected_layout", "DEFAULT"),
mode,
campaign=session.get("selected_campaign"),
)
or base_layout_doc
)
_layout_parameters, active_diagnostics, active_fields = resolve_layout_selections(
active_layout_doc,
known_fields=_known_layout_fields(parameters_collection, mode),
)
ordered = ordered_unique_strings([
*active_fields,
*always_include,
*active_diagnostics,
])
ordered = _ensure_shot_identity_fields(ordered)
refresh_diagnostic_choice_cache(ordered)
session["SELECTED_FIELDS_LIST"] = ordered
session["diagnostics_only"] = list(active_diagnostics)
session["always_include"] = always_include
if initialize_default_campaign and campaign_is_unset(
session.get("selected_campaign")
):
session["selected_campaign"] = choose_default_campaign_name()
session[startup_session_key] = True
logging.info("Startup %s complete", startup_session_key)
return True
[docs]
def bootstrap_request_form_session_state(endpoint: str | None = None) -> bool:
active_endpoint = endpoint or request.endpoint
if not active_endpoint:
return False
config = FORM_SESSION_BOOTSTRAP_CONFIG.get(active_endpoint)
if not config:
return False
return bootstrap_form_session_state(**config)
[docs]
def build_route_form_state(
mode: str, parameters_collection, *, apply_demo_defaults_now: bool = False
) -> dict:
if apply_demo_defaults_now:
apply_demo_defaults(mode, parameters_collection)
campaign = session.get("selected_campaign")
source_layout_key = f"selected_source_layout_{mode}"
selected_layout = (
session.get("selected_layout")
or session.get(source_layout_key)
or session.get(f"selected_source_layout_{paired_mode(mode)}")
or "DEFAULT"
)
session[source_layout_key] = selected_layout
session["selected_layout"] = selected_layout
active_header_doc = get_active_header_doc(mode)
header_mode = str((active_header_doc or {}).get("mode") or "").strip()
header_is_mode_compatible = not header_mode or header_mode == mode
header_source_layout = (active_header_doc or {}).get("source_layout_name")
if (
header_is_mode_compatible
and isinstance(header_source_layout, str)
and header_source_layout.strip()
):
selected_layout = header_source_layout.strip()
session[source_layout_key] = selected_layout
session["selected_layout"] = selected_layout
base_layout_doc = get_default_layout_doc(parameters_collection, mode)
active_layout_doc = (
get_active_layout_doc(
parameters_collection,
selected_layout,
mode,
campaign=campaign,
)
or base_layout_doc
)
known_fields = _known_layout_fields(parameters_collection, mode)
_layout_parameters, resolved_diagnostics, resolved_fields = (
resolve_layout_selections(
active_layout_doc,
known_fields=known_fields,
)
)
if header_is_mode_compatible:
_header_parameters, header_diagnostics, header_fields = (
resolve_layout_selections(
active_header_doc,
known_fields=known_fields,
)
)
else:
header_diagnostics = []
header_fields = []
layout_diagnostics = list(
(active_layout_doc or {}).get("diagnostics_list") or resolved_diagnostics or []
)
layout_diagnostics = [
field_name for field_name in layout_diagnostics if field_name in known_fields
]
layout_fields = list(resolved_fields) or [
field_name
for field_name in flatten_layout_fields(active_layout_doc)
if field_name in known_fields
]
if header_is_mode_compatible:
header_fields = list(header_fields) or [
field_name
for field_name in flatten_layout_fields(active_header_doc)
if field_name in known_fields
]
else:
header_fields = []
session_visible_fields = ordered_unique_strings(
field_name
for field_name in session.get("SELECTED_FIELDS_LIST", [])
if field_name in known_fields
)
session_diagnostics = ordered_unique_strings(
field_name
for field_name in session.get("diagnostics_only", [])
if field_name in known_fields
)
if header_fields:
session_extras = [
field_name
for field_name in session_visible_fields
if field_name not in layout_fields and field_name not in header_fields
]
required_layout_fields = [
field_name
for field_name in layout_fields
if field_name not in header_fields
and field_name in _fallback_required_fields_for_mode(mode)
]
visible_fields = ordered_unique_strings([
*session_extras,
*header_fields,
*required_layout_fields,
*header_diagnostics,
])
diagnostics_only = ordered_unique_strings([
*(
field_name
for field_name in session_diagnostics
if field_name not in layout_diagnostics
),
*header_diagnostics,
])
else:
visible_fields = ordered_unique_strings([
*session_visible_fields,
*layout_fields,
])
diagnostics_only = ordered_unique_strings([
*session_diagnostics,
*layout_diagnostics,
])
required_visible_fields = required_identity_fields_for_mode(
mode,
known_fields=known_fields,
include_set_shot_number_list=False,
)
if mode == "set":
# Keep route-state compatibility: do not inject set-mode date_time unless
# the active layout/header explicitly carries it.
required_visible_fields = [
field_name
for field_name in required_visible_fields
if field_name != "date_time"
or field_name in layout_fields
or field_name in header_fields
]
visible_fields = ordered_unique_strings([*visible_fields, *required_visible_fields])
if not visible_fields:
visible_fields = list(layout_fields)
if not diagnostics_only and not session_visible_fields:
diagnostics_only = list(layout_diagnostics)
visible_fields = _ensure_shot_identity_fields(
ordered_unique_strings([*visible_fields, *diagnostics_only])
)
always_include = list((base_layout_doc or {}).get("always_include", []))
logging.log(logging.DEBUG, "Visible fields: %s", visible_fields)
logging.log(logging.DEBUG, "Selected layout: %s", selected_layout)
logging.log(logging.DEBUG, "Always include: %s", always_include)
logging.log(logging.DEBUG, "Active header doc: %s", active_header_doc)
logging.log(logging.DEBUG, "Active layout doc: %s", active_layout_doc)
session["SELECTED_FIELDS_LIST"] = visible_fields
session["diagnostics_only"] = diagnostics_only
session["always_include"] = always_include
refresh_diagnostic_choice_cache(visible_fields)
return {
"visible_fields": visible_fields,
"selected_layout": selected_layout,
"always_include": always_include,
"options_collection": get_custom_options_collection_name(),
"active_header_doc": active_header_doc,
"active_layout_doc": active_layout_doc,
}
[docs]
def merge_layout_session_selection_state(
parameters_collection,
selected_layout: str,
mode: str,
*,
ensure_diagnostics_match_selected: bool = False,
default_selected_to_diagnostics: bool = False,
) -> tuple[list[str], list[str]]:
"""Merge layout-backed and session-backed selection state and persist it."""
layout_doc = get_active_layout_doc(parameters_collection, selected_layout, mode)
_, layout_diagnostics, layout_selected = resolve_layout_selections(layout_doc)
merged_selected = ordered_unique_strings([
*session.get("SELECTED_FIELDS_LIST", []),
*layout_selected,
])
merged_diagnostics = ordered_unique_strings([
*session.get("diagnostics_only", []),
*layout_diagnostics,
])
if ensure_diagnostics_match_selected and merged_selected:
diagnostic_names = _known_diagnostic_display_names()
merged_diagnostics = ordered_unique_strings([
*merged_diagnostics,
*(field for field in merged_selected if field in diagnostic_names),
])
if default_selected_to_diagnostics and not merged_selected and merged_diagnostics:
merged_selected = list(merged_diagnostics)
merged_selected = _ensure_shot_identity_fields(merged_selected)
session["SELECTED_FIELDS_LIST"] = merged_selected
session["diagnostics_only"] = merged_diagnostics
return merged_selected, merged_diagnostics
[docs]
def refresh_dynamic_form_bindings(
*, mode: str | None = None, visible_fields: Iterable[str] | None = None
):
return create_dynamic_form(
collection=current_app.config["MONGODB_COLLECTION_FOR_SETTINGS"],
options_collection=get_custom_options_collection_name(),
mode=mode or session.get("mode", "shot"),
visible_fields=list(visible_fields or session.get("SELECTED_FIELDS_LIST", [])),
)
[docs]
def collect_visible_custom_fields(
form,
visible_fields: Iterable[str],
fields_cursor: Iterable[dict] | None,
) -> tuple[dict[str, str], dict[str, list[str]]]:
layout_fields = {
field_name
for fields in getattr(form, "FIELD_SECTIONS", {}).values()
for field_name in (fields or [])
}
not_used_fields = set(getattr(form, "NOT_USED_FIELDS", []) or [])
visible = set(visible_fields)
custom_field_types: dict[str, str] = {}
custom_fields_by_section: dict[str, list[str]] = {}
for field in fields_cursor or []:
field_name = field.get("field_name")
if (
field_name in not_used_fields
or field_name not in visible
or field_name in layout_fields
):
continue
custom_field_type = field.get("custom_field_type")
section = field.get("section")
if not (field_name and custom_field_type):
continue
custom_field_types[field_name] = custom_field_type
if section:
custom_fields_by_section.setdefault(section, []).append(field_name)
return custom_field_types, custom_fields_by_section