__init__.py 1.5 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344
  1. """
  2. CLI functions for FlexMeasures hosts.
  3. """
  4. import sys
  5. from flask import Flask, current_app
  6. def register_at(app: Flask):
  7. if app.cli:
  8. with app.app_context():
  9. import flexmeasures.cli.jobs
  10. import flexmeasures.cli.monitor
  11. import flexmeasures.cli.data_add
  12. import flexmeasures.cli.data_edit
  13. import flexmeasures.cli.data_show
  14. import flexmeasures.cli.data_delete
  15. import flexmeasures.cli.db_ops
  16. import flexmeasures.cli.testing # noqa: F401
  17. def is_running() -> bool:
  18. """
  19. True if we are running one of the custom FlexMeasures CLI commands.
  20. We use this in combination with authorization logic, e.g. we assume that only sysadmins run commands there,
  21. but also we consider forecasting & scheduling jobs to be in that realm, as well.
  22. This tooling might not live forever, as we could evolve into a more sophisticated auth model for these cases.
  23. For instance, these jobs are queued by the system, but caused by user actions (sending data), and then they are run by the system.
  24. See also: the run_as_cli test fixture, which uses the (non-public) PRETEND_RUNNING_AS_CLI env setting.
  25. """
  26. cli_sets = current_app.cli.list_commands(ctx=None)
  27. command_line = " ".join(sys.argv)
  28. for cli_set in cli_sets:
  29. if f"flexmeasures {cli_set}" in command_line:
  30. return True
  31. if current_app.config.get("PRETEND_RUNNING_AS_CLI", False):
  32. return True
  33. return False