097be3e3fe
Instead of storing a cookie for every plugin that requires launch, we create a single cookie with '+' separated value. We make use of the cookieStore native API (available everywhere since June 2025) to access cookie data. The variables are renamed to be more explicit. We now use class-based SCSS for styling, instead of manually setting style.display attribute.
391 lines
11 KiB
Python
391 lines
11 KiB
Python
import asyncio
|
|
import json
|
|
import logging
|
|
import sys
|
|
import typing as t
|
|
|
|
import importlib_metadata
|
|
from markdown import markdown
|
|
from quart import (
|
|
Quart,
|
|
Response,
|
|
g,
|
|
jsonify,
|
|
make_response,
|
|
redirect,
|
|
render_template,
|
|
request,
|
|
url_for,
|
|
)
|
|
from quart.helpers import WerkzeugResponse
|
|
from quart.typing import ResponseTypes
|
|
from tutor.plugins.v1 import discover_package
|
|
|
|
from tutordeck.server.utils import current_page_plugins, pagination_context
|
|
|
|
from . import constants, tutorclient
|
|
|
|
|
|
app = Quart(
|
|
__name__,
|
|
static_url_path="/static",
|
|
static_folder="static",
|
|
)
|
|
|
|
|
|
def run(root: str, **app_kwargs: t.Any) -> None:
|
|
"""
|
|
Bootstrap the Quart app and run it.
|
|
"""
|
|
tutorclient.Project.connect(root)
|
|
|
|
# Configure logging
|
|
formatter = logging.Formatter("%(name)s - %(levelname)s - %(message)s")
|
|
handler = logging.StreamHandler(sys.stdout)
|
|
handler.setFormatter(formatter)
|
|
tutorclient.logger.addHandler(handler)
|
|
tutorclient.logger.setLevel(logging.INFO)
|
|
|
|
# TODO app.run() should be called only in development
|
|
app.run(**app_kwargs)
|
|
|
|
|
|
@app.before_request
|
|
async def before_request() -> None:
|
|
# Shared views and template context
|
|
g.installed_plugins = tutorclient.Client.installed_plugins()
|
|
g.enabled_plugins = tutorclient.Client.enabled_plugins()
|
|
|
|
|
|
@app.get("/")
|
|
async def home() -> str:
|
|
return await render_template("plugin_installed.html")
|
|
|
|
|
|
@app.get("/plugin/store")
|
|
async def plugin_store() -> str:
|
|
return await render_template("plugin_store.html")
|
|
|
|
|
|
@app.get("/plugin/installed")
|
|
async def plugin_installed() -> str:
|
|
return await render_template("plugin_installed.html")
|
|
|
|
|
|
@app.get("/plugin/store/list")
|
|
async def plugin_store_list() -> str:
|
|
search_query = request.args.get("search", "")
|
|
plugins: list[dict[str, t.Any]] = [
|
|
{
|
|
"name": p.name,
|
|
"url": p.url,
|
|
"index": p.index,
|
|
"author": p.author.split("<")[0].strip(),
|
|
"description": p.short_description,
|
|
"is_installed": p.name in g.installed_plugins,
|
|
"is_enabled": p.name in g.enabled_plugins,
|
|
}
|
|
for p in tutorclient.Client.plugins_in_store()
|
|
if p.name in tutorclient.Client.plugins_matching_pattern(search_query)
|
|
]
|
|
|
|
current_page = int(request.args.get("page", "1"))
|
|
plugins = current_page_plugins(plugins, current_page)
|
|
pagination = pagination_context(plugins, current_page)
|
|
|
|
return await render_template(
|
|
"_plugin_store_list.html",
|
|
plugins=plugins,
|
|
pagination=pagination,
|
|
)
|
|
|
|
|
|
@app.get("/plugin/installed/list")
|
|
async def plugin_installed_list() -> str:
|
|
search_query = request.args.get("search", "")
|
|
plugins: list[dict[str, t.Any]] = [
|
|
{
|
|
"name": p.name,
|
|
"url": p.url,
|
|
"index": p.index,
|
|
"author": p.author.split("<")[0].strip(),
|
|
"description": p.short_description,
|
|
"is_enabled": p.name in g.enabled_plugins,
|
|
}
|
|
for p in tutorclient.Client.plugins_in_store()
|
|
if p.name in tutorclient.Client.plugins_matching_pattern(search_query)
|
|
and p.name in g.installed_plugins
|
|
]
|
|
|
|
return await render_template(
|
|
"_plugin_installed_list.html",
|
|
plugins=plugins,
|
|
)
|
|
|
|
|
|
@app.get("/plugin/<name>")
|
|
async def plugin(name: str) -> Response:
|
|
index_entry = tutorclient.Client.plugin_in_store(name)
|
|
if not index_entry:
|
|
return Response("Plugin not found", status=404)
|
|
|
|
seq_command_executed = request.args.get("seq_command_executed")
|
|
author = index_entry.author.split("<")[0].strip()
|
|
description = markdown(index_entry.description)
|
|
rendered_template = await render_template(
|
|
"plugin.html",
|
|
plugin_name=name,
|
|
is_enabled=name in g.enabled_plugins,
|
|
is_installed=name in g.installed_plugins,
|
|
author_name=author,
|
|
plugin_description=description,
|
|
seq_command_executed=seq_command_executed,
|
|
plugin_config_unique=tutorclient.Client.plugin_config_unique(name),
|
|
plugin_config_defaults=tutorclient.Client.plugin_config_defaults(name),
|
|
user_config=tutorclient.Project.get_user_config(),
|
|
)
|
|
|
|
# Redirect to plugin page
|
|
response = Response(rendered_template, status=200, content_type="text/html")
|
|
response.headers["HX-Redirect"] = url_for(
|
|
"plugin", name=name, seq_command_executed=seq_command_executed
|
|
)
|
|
return response
|
|
|
|
|
|
@app.get("/plugin/<name>/is-installed")
|
|
def plugin_installed_status(name: str) -> Response:
|
|
return jsonify({"installed": name in g.installed_plugins})
|
|
|
|
|
|
@app.post("/plugin/<name>/toggle")
|
|
async def plugin_toggle(name: str) -> Response:
|
|
# TODO check plugin exists
|
|
form = await request.form
|
|
enable_plugin = form.get("checked") == "on"
|
|
tutorclient.CliPool.run_sequential(
|
|
["plugins", "enable" if enable_plugin else "disable", name]
|
|
)
|
|
# TODO error management
|
|
|
|
response = t.cast(
|
|
Response,
|
|
await make_response(
|
|
redirect(
|
|
url_for(
|
|
"plugin",
|
|
name=name,
|
|
seq_command_executed=True,
|
|
)
|
|
)
|
|
),
|
|
)
|
|
if enable_plugin:
|
|
update_plugins_requiring_launch(response, add=name)
|
|
else:
|
|
update_plugins_requiring_launch(response, remove=name)
|
|
return response
|
|
|
|
|
|
@app.post("/plugin/<name>/install")
|
|
async def plugin_install(name: str) -> WerkzeugResponse:
|
|
async def bg_install_and_reload() -> None:
|
|
tutorclient.CliPool.run_parallel(app, ["plugins", "install", name])
|
|
while tutorclient.CliPool.THREAD and tutorclient.CliPool.THREAD.is_alive():
|
|
await asyncio.sleep(0.1)
|
|
discover_package(importlib_metadata.entry_points().__getitem__(name))
|
|
|
|
asyncio.create_task(bg_install_and_reload())
|
|
return redirect(
|
|
url_for(
|
|
"plugin",
|
|
name=name,
|
|
)
|
|
)
|
|
|
|
|
|
@app.post("/plugin/<name>/upgrade")
|
|
async def plugin_upgrade(name: str) -> WerkzeugResponse:
|
|
tutorclient.CliPool.run_parallel(app, ["plugins", "upgrade", name])
|
|
return redirect(
|
|
url_for(
|
|
"plugin",
|
|
name=name,
|
|
)
|
|
)
|
|
|
|
|
|
@app.post("/plugins/update")
|
|
async def plugins_update() -> WerkzeugResponse:
|
|
tutorclient.CliPool.run_sequential(["plugins", "update"])
|
|
return redirect(url_for("plugin_store"))
|
|
|
|
|
|
@app.post("/config/<name>/update")
|
|
async def config_update(name: str) -> Response:
|
|
form = await request.form
|
|
|
|
unset = form.get("unset")
|
|
if unset:
|
|
tutorclient.CliPool.run_sequential(["config", "save", f"--unset={unset}"])
|
|
else:
|
|
cmd = ["config", "save"]
|
|
for key, value in form.items():
|
|
if value.startswith("{{"):
|
|
# Templated values that start with {{ should be explicitely converted to string
|
|
# Otherwise there will be a parsing error because it might be considered a dictionary
|
|
value = f"'{value}'"
|
|
cmd.extend(["--set", f"{key}={value}"])
|
|
tutorclient.CliPool.run_sequential(cmd)
|
|
# TODO error management
|
|
response = t.cast(
|
|
Response,
|
|
await make_response(
|
|
redirect(
|
|
url_for(
|
|
"plugin",
|
|
name=name,
|
|
seq_command_executed=True,
|
|
)
|
|
)
|
|
),
|
|
)
|
|
response.set_cookie(
|
|
f"{constants.WARNING_COOKIE_PREFIX}-{name}",
|
|
"requires launch",
|
|
max_age=constants.ONE_MONTH,
|
|
)
|
|
return response
|
|
|
|
|
|
@app.get("/local/launch")
|
|
async def local_launch_view() -> str:
|
|
return await render_template(
|
|
"local_launch.html",
|
|
)
|
|
|
|
|
|
@app.post("/cli/local/launch")
|
|
async def cli_local_launch() -> str:
|
|
tutorclient.CliPool.run_parallel(app, ["local", "launch", "--non-interactive"])
|
|
return await render_template(
|
|
"local_launch.html",
|
|
)
|
|
|
|
|
|
@app.get("/cli/logs/stream")
|
|
async def cli_logs_stream() -> ResponseTypes:
|
|
"""
|
|
We only need single-direction communication, so we use server-sent events, and not
|
|
websockets.
|
|
https://quart.palletsprojects.com/en/latest/how_to_guides/server_sent_events.html
|
|
|
|
Note that server interruption with ctrl+c does not work in Python 3.12 and 3.13
|
|
because of this bug:
|
|
https://github.com/pallets/quart/issues/333
|
|
https://github.com/python/cpython/issues/123720
|
|
|
|
Events are sent with the following format:
|
|
|
|
data: "json-encoded string..."
|
|
event: logs
|
|
|
|
Data is JSON-encoded such that we can sent newline characters, etc.
|
|
"""
|
|
|
|
# TODO check that request accepts event stream (see howto)
|
|
async def send_events() -> t.AsyncIterator[bytes]:
|
|
while True:
|
|
# TODO this is again causing the stream to never stop...
|
|
async for data in tutorclient.CliPool.iter_logs():
|
|
event = f"""data: {
|
|
json.dumps(
|
|
{
|
|
"stdout": data,
|
|
"command": tutorclient.CliPool.current_command(),
|
|
"thread_alive": tutorclient.CliPool.is_thread_alive(),
|
|
}
|
|
)
|
|
}\nevent: logs\n\n"""
|
|
yield event.encode()
|
|
await asyncio.sleep(constants.SHORT_SLEEP_SECONDS)
|
|
|
|
response = await make_response(
|
|
send_events(),
|
|
{
|
|
"Content-Type": "text/event-stream",
|
|
"Cache-Control": "no-cache",
|
|
"Transfer-Encoding": "chunked",
|
|
},
|
|
)
|
|
setattr(response, "timeout", None)
|
|
return response
|
|
|
|
|
|
@app.post("/cli/stop")
|
|
async def cli_stop() -> Response:
|
|
tutorclient.CliPool.stop()
|
|
return Response(status=200)
|
|
|
|
|
|
@app.get("/advanced")
|
|
async def advanced() -> str:
|
|
return await render_template(
|
|
"advanced.html",
|
|
)
|
|
|
|
|
|
@app.post("/suggest")
|
|
async def suggest() -> Response:
|
|
data = await request.get_json()
|
|
partial_command = data.get("command", "")
|
|
suggestions = tutorclient.Client.autocomplete(partial_command)
|
|
return jsonify(suggestions)
|
|
|
|
|
|
@app.post("/command")
|
|
async def command() -> WerkzeugResponse:
|
|
form = await request.form
|
|
command_string = form.get("command", "")
|
|
command_args = command_string.split()
|
|
tutorclient.CliPool.run_parallel(app, command_args)
|
|
return redirect(url_for("advanced"))
|
|
|
|
|
|
def update_plugins_requiring_launch(
|
|
response: Response, add: str | None = None, remove: str | None = None
|
|
) -> None:
|
|
"""
|
|
Store the list of plugins for which a recent set of changes require running "local launch".
|
|
|
|
This list is stored as a "+"-separated string in a cookie. Note that flask will automatically put the content in quotes.
|
|
"""
|
|
# Note that comma, colon and semi-colon are not supported in cookie values
|
|
separator = "+"
|
|
|
|
# Get current plugins
|
|
names = set(
|
|
[
|
|
cookie
|
|
for cookie in request.cookies.get(
|
|
constants.PLUGINS_REQUIRE_LAUNCH_COOKIE_NAME, ""
|
|
).split(separator)
|
|
if cookie
|
|
]
|
|
)
|
|
|
|
# Add new plugins
|
|
if add:
|
|
names.add(add)
|
|
|
|
# Remove plugins
|
|
if remove:
|
|
names.discard(remove)
|
|
|
|
# Update the response
|
|
response.set_cookie(
|
|
constants.PLUGINS_REQUIRE_LAUNCH_COOKIE_NAME,
|
|
separator.join(sorted(names)),
|
|
max_age=60 * 60 * 24 * 30, # 1 month
|
|
)
|
