-
Notifications
You must be signed in to change notification settings - Fork 0
Lifecycle Plugin Guide
60plus edited this page Apr 13, 2026
·
1 revision
Lifecycle plugins hook into app events - startup, shutdown, game added, download complete. They can also register custom API endpoints.
| Hook | Args | Purpose |
|---|---|---|
lifecycle_on_startup() |
none | App started, register endpoints, init resources |
lifecycle_on_shutdown() |
none | App stopping, clean up |
lifecycle_on_game_added(game) |
dict | New game added to any library |
lifecycle_on_download_complete(game, path) |
dict, str | Download finished |
The Translator plugin shows key lifecycle patterns:
from plugins.hookspecs import hookimpl
class Plugin:
@hookimpl
def lifecycle_on_startup(self) -> None:
logger.info("Translator plugin loaded")The translator's /api/plugins/translate endpoint is built into GD's plugins_router.py. It calls translate_text() from the plugin module directly.
For your own plugins, you can register FastAPI endpoints in lifecycle_on_startup():
@hookimpl
def lifecycle_on_startup(self) -> None:
from main import app
@app.get("/api/plugins/my-endpoint")
async def my_endpoint():
return {"status": "ok", "data": "..."}If your plugin calls blocking code (external APIs, shell commands), use a thread pool:
import concurrent.futures
def _blocking_work(text):
# This runs in a separate thread
import some_blocking_library
return some_blocking_library.process(text)
async def my_endpoint(text: str):
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
future = pool.submit(_blocking_work, text)
result = future.result(timeout=60)
return {"result": result}The Translator uses this pattern because translate-shell internally calls asyncio.run() which conflicts with FastAPI's event loop.
For processing large texts, split into chunks:
CHUNK_SIZE = 450
def chunk_text(text):
paras = text.split("\n")
chunks, current, current_len = [], [], 0
for p in paras:
if current and current_len + len(p) + 1 > CHUNK_SIZE:
chunks.append("\n".join(current))
current, current_len = [p], len(p)
else:
current.append(p)
current_len += len(p) + 1
if current:
chunks.append("\n".join(current))
return chunksUse config_schema for user settings:
{
"config_schema": {
"from_lang": {
"type": "select",
"options": ["auto", "en", "pl", "de", "fr", "es"],
"default": "en",
"label": "Source language"
},
"to_lang": {
"type": "select",
"options": ["pl", "en", "de", "fr", "es"],
"default": "pl",
"label": "Target language"
}
}
}@hookimpl
def lifecycle_on_game_added(self, game: dict) -> None:
title = game.get("title", "Unknown")
# Send Discord webhook, update spreadsheet, etc.
send_notification(f"New game added: {title}")@hookimpl
def lifecycle_on_download_complete(self, game: dict, path: str) -> None:
# Scan with antivirus, move file, notify user
logger.info(f"Download complete: {game['title']} -> {path}")