Skip to content

Lifecycle Plugin Guide

60plus edited this page Apr 13, 2026 · 1 revision

Lifecycle Plugin Guide

Lifecycle plugins hook into app events - startup, shutdown, game added, download complete. They can also register custom API endpoints.

Hooks

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

Example: Description Translator

The Translator plugin shows key lifecycle patterns:

Plugin Class

from plugins.hookspecs import hookimpl

class Plugin:
    @hookimpl
    def lifecycle_on_startup(self) -> None:
        logger.info("Translator plugin loaded")

Custom Endpoints

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": "..."}

Thread Safety

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.

Text Chunking

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 chunks

Config Schema

Use 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"
    }
  }
}

Event Hook Examples

React to new games

@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}")

Post-download processing

@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}")

Clone this wiki locally