From f36321a918c63703ffd6b658057127a5362260be Mon Sep 17 00:00:00 2001 From: Matthias Mair Date: Sun, 1 Nov 2026 10:43:27 +0100 Subject: [PATCH] add state transtion graph export (#13136) --- .../management/commands/export_fsm.py | 113 ++++++++++++++++++ .../InvenTree/generic/states/api_helpers.py | 14 +++ tasks.py | 4 + 3 files changed, 131 insertions(+) create mode 100644 src/backend/InvenTree/InvenTree/management/commands/export_fsm.py diff --git a/src/backend/InvenTree/InvenTree/management/commands/export_fsm.py b/src/backend/InvenTree/InvenTree/management/commands/export_fsm.py new file mode 100644 index 0000000000..68874fa242 --- /dev/null +++ b/src/backend/InvenTree/InvenTree/management/commands/export_fsm.py @@ -0,0 +1,113 @@ +"""Custom management command to export the fsm states of models.""" + +import json + +from django.apps import apps +from django.core.management.base import BaseCommand +from django.utils.text import capfirst + +from generic.states import DEFERRABLE, RETURN_VALUE +from generic.states.introspection import state_fields, transition_methods + + +class Command(BaseCommand): + """Extract fsm model information, and export to a JSON file.""" + + def add_arguments(self, parser): + """Add custom arguments for this command.""" + parser.add_argument( + 'filename', type=str, help='Output filename for audited model definitions' + ) + + def handle(self, *args, **kwargs): + """Export fsm model information to a JSON file.""" + models = discover_fsm_models() + + filename = kwargs.get('filename', 'inventree_audited_models.json') + + with open(filename, 'w', encoding='utf-8') as f: + json.dump(models, f, indent=4) + + print(f"Exported InvenTree fsm model definitions to '{filename}'") + + +def extract_value(src): + """Return the plain value (or list of values) of a transition source/target.""" + if isinstance(src, DEFERRABLE): + src = src.target + + value = getattr(src, 'value', src) + + if isinstance(value, RETURN_VALUE): + return [state.value for state in value.allowed_states] + return value + + +def describe_operations(methods): + """Describe each transition method with its sources, targets and conditions.""" + return { + name: { + key: { + 'source': extract_value(transition.source), + 'extra_permission': transition.permission, + 'target': extract_value(transition.target), + 'conditions': list(transition.conditions), + } + for key, transition in meta.transitions.items() + } + for name, meta in methods.items() + } + + +def reachable_states(methods): + """Map each source state to the sorted list of states reachable from it.""" + # target -> sources; the last transition of a method defines its targets + sources_by_target = {} + for meta in methods.values(): + sources = [] + targets = [] + for transition in meta.transitions.values(): + targets = extract_value(transition.target) + if not isinstance(targets, list): + targets = [int(targets)] + sources.append(extract_value(transition.source)) + for target in targets: + sources_by_target[target] = sources + + reachable = {} + for target, sources in sources_by_target.items(): + for source in sources: + reachable.setdefault(source, []).append(target) + return {source: sorted(targets) for source, targets in reachable.items()} + + +def discover_fsm_models(): + """Discover all models which have fsm states. + + Returns a list of models, sorted by app and then by model name. + """ + from generic.states.api_helpers import supported_models + + models = [] + + for model in supported_models(): + states = {} + for field in state_fields(model): + methods = transition_methods(model, field) + states[field] = { + 'operations': describe_operations(methods), + 'transitions': reachable_states(methods), + 'states': { + value: str(label) + for value, label in getattr(model, field).field.choices + }, + } + + models.append({ + 'model_type': model._meta.label_lower, + 'app': str(apps.get_app_config(model._meta.app_label).verbose_name), + 'name': capfirst(str(model._meta.verbose_name)), + 'states': states, + }) + + return sorted(models, key=lambda m: (m['app'], m['name'])) diff --git a/src/backend/InvenTree/generic/states/api_helpers.py b/src/backend/InvenTree/generic/states/api_helpers.py index 132caeb15c..60eed6b2c7 100644 --- a/src/backend/InvenTree/generic/states/api_helpers.py +++ b/src/backend/InvenTree/generic/states/api_helpers.py @@ -5,6 +5,7 @@ from __future__ import annotations import inspect from typing import Any +from django.apps import apps from django.core.exceptions import ValidationError as DjangoValidationError from django.utils.translation import gettext_lazy as _ @@ -15,6 +16,7 @@ from rest_framework.exceptions import PermissionDenied from rest_framework.request import Request from rest_framework.response import Response +from generic.states import StateTransitionMixin from InvenTree.serializers import EmptySerializer from users.permissions import check_user_permission @@ -404,6 +406,18 @@ class FSMTransitionMixin: return Response(AvailableTransitionSerializer(data, many=True).data) +def supported_models() -> list: + """Return all installed models which have FSM transitions.""" + return sorted( + ( + model + for model in apps.get_models() + if issubclass(model, StateTransitionMixin) and not model._meta.abstract + ), + key=lambda model: model._meta.label_lower, + ) + + __all__ = [ 'TRANSITION_URL_PREFIX', 'FSMTransitionMixin', diff --git a/tasks.py b/tasks.py index e600877066..134d92d7b6 100644 --- a/tasks.py +++ b/tasks.py @@ -2163,6 +2163,7 @@ def export_definitions(c, basedir: str = ''): base_path.joinpath('inventree_status_codes.json'), base_path.joinpath('inventree_roles.json'), base_path.joinpath('inventree_audited_models.json'), + base_path.joinpath('inventree_fsm.json'), ] info('Exporting definitions...') @@ -2186,6 +2187,9 @@ def export_definitions(c, basedir: str = ''): check_file_existence(filenames[6], overwrite=True) manage(c, f'export_audited_models {filenames[6]}', pty=True) + check_file_existence(filenames[7], overwrite=True) + manage(c, f'export_fsm {filenames[7]}', pty=True) + info('Exporting definitions complete')