Skip to content

Instantly share code, notes, and snippets.

@playday3008

playday3008/lutris_api_fetch_all.py

Last active July 24, 2025 23:37
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save playday3008/9e6188d5190f1963bde544972f55b270 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save playday3008/9e6188d5190f1963bde544972f55b270 to your computer and use it in GitHub Desktop.
Download ZIP
Lutris API Stuff
Raw
lutris_api_fetch_all.py
from __future__ import annotations
from datetime import datetime
from typing import Optional, Any, cast
from enum import Enum
import sys
import requests
import math
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from pydantic import BaseModel, Field
LUTRIS_API_BASE = "https://lutris.net/api"
LUTRIS_API_GAMES = LUTRIS_API_BASE + "/games"
LUTRIS_API_INSTALLERS = LUTRIS_API_BASE + "/installers"
class Platform(BaseModel):
name: str
class ProviderGame(BaseModel):
name: str
slug: str
service: str
class GameAlias(BaseModel):
slug: str
name: str
class ShaderCache(BaseModel):
url: str
updated_at: str
class ChangeFor(BaseModel):
id: int
name: str
slug: str
class Game(BaseModel):
id: int
name: str
slug: str
year: Optional[int]
banner_url: str
icon_url: str
coverart: Optional[str]
platforms: list[Platform]
provider_games: list[ProviderGame]
aliases: list[GameAlias]
shaders: list[ShaderCache]
discord_id: Optional[str]
change_for: Optional[ChangeFor]
class Config:
populate_by_name = True
class InstallerScript(BaseModel):
class RunnerEnum(str, Enum):
AGS = "ags"
ATARI800 = "atari800"
BROWSER = "browser"
CITRA = "citra"
DESMUME = "desmume"
DGEN = "dgen"
DOLPHIN = "dolphin"
DOSBOX = "dosbox"
FROTZ = "frotz"
FSUAE = "fsuae"
HATARI = "hatari"
JZINTV = "jzintv"
LIBRETRO = "libretro"
LINUX = "linux"
MAME = "mame"
MEDNAFEN = "mednafen"
MUPEN64PLUS = "mupen64plus"
O2EM = "o2em"
OPENMSX = "openmsx"
OSMOSE = "osmose"
PCSX2 = "pcsx2"
PCSXR = "pcsxr"
PPSSPP = "ppsspp"
REICAST = "reicast"
RPCS3 = "rpcs3"
SCUMMVM = "scummvm"
SNES9X = "snes9x"
STEAM = "steam"
WINESTEAM = "winesteam"
STELLA = "stella"
VICE = "vice"
VIRTUALJAGUAR = "virtualjaguar"
WEB = "web"
WINE = "wine"
YUZU = "yuzu"
ZDOOM = "zdoom"
class ArchiveFormatEnum(str, Enum):
TGZ = "tgz"
TAR = "tar"
ZIP = "zip"
SEVEN_Z = "7z"
RAR = "rar"
TXZ = "txz"
BZ2 = "bz2"
GZIP = "gzip"
DEB = "deb"
EXE = "exe"
GOG = "gog"
class TaskNameEnum(str, Enum):
CREATE_PREFIX = "create_prefix"
WINEEXEC = "wineexec"
WINETRICKS = "winetricks"
EJECT_DISK = "eject_disk"
SET_REGEDIT = "set_regedit"
DELETE_REGISTRY_KEY = "delete_registry_key"
SET_REGEDIT_FILE = "set_regedit_file"
WINEKILL = "winekill"
DOSEXEC = "dosexec"
class LaunchConfig(BaseModel):
name: str = Field(..., description="Identifier to show in the launcher dialog")
command: Optional[str] = Field(
None,
description="Full bash command to execute, overriding default game command entirely",
)
exe: str = Field(
..., description="Main game executable, to combine with the command"
)
args: Optional[str] = Field(None, description="Additional command arguments")
working_dir: Optional[str] = Field(
None, description="Working directory to use when executing the executable"
)
class Game(BaseModel):
exe: Optional[str] = Field(None, description="Main game executable")
main_file: Optional[str] = Field(None, description="Emulator ROM or disk file")
args: Optional[str] = Field(None, description="Additional command arguments")
working_dir: Optional[str] = Field(
None, description="Working directory to use when executing the executable"
)
launch_configs: Optional[list["InstallerScript.LaunchConfig"]] = Field(
None, description="A list of mutiple related executable configurations"
)
arch: Optional[str] = Field(
None, description="The architecture of a Wine prefix"
)
prefix: Optional[str] = Field(None, description="Path to the Wine prefix")
run_without_steam: Optional[bool] = Field(
None,
description="Activate the DRM free mode and no not launch Steam when the game runs",
)
steamless_binary: Optional[str] = Field(
None,
description="Path of the game executable if it's able to run without the Steam client",
)
path: Optional[str] = Field(None, description="Location of the game files")
class FileItem(BaseModel):
url: Optional[str] = Field(None, description="File URL")
filename: Optional[str] = Field(
None, description="The name of the file's saved copy"
)
referer: Optional[str] = Field(None, description="File referer domain")
class InsertDisc(BaseModel):
requires: str = Field(..., description="Required disk")
class Move(BaseModel):
src: str = Field(..., description="Source file ID or path")
dst: str = Field(..., description="Destination path")
class Merge(BaseModel):
src: str = Field(..., description="Source file ID or path")
dst: str = Field(..., description="Destination path")
class Extract(BaseModel):
file: str = Field(..., description="Source file ID or path")
dst: str = Field(..., description="Destination path")
format: Optional["InstallerScript.ArchiveFormatEnum"] = Field(
None, description="Archive's type"
)
class Execute(BaseModel):
command: Optional[str] = Field(None, description="Full bash command to execute")
file: Optional[str] = Field(None, description="File ID or a path to file")
args: Optional[str] = Field(None, description="Executable arguments")
terminal: Optional[bool] = Field(
None, description="Execute in a new terminal window"
)
exclude_processes: Optional[str] = Field(
None,
description="Space-separated list of processes to exclude from being monitored when determining if the execute phase finished",
)
include_processes: Optional[str] = Field(
None,
description="Space-separated list of processes to monitor when determining if the execute phase finished",
)
disable_runtime: Optional[bool] = Field(
None, description="Run a process without Lutris runtime"
)
env: Optional[dict[str, str]] = Field(None, description="Environment variables")
class WriteFile(BaseModel):
content: str = Field(..., description="File content to write")
file: str = Field(..., description="Destination file pathpath")
class WriteConfig(BaseModel):
merge: Optional[bool] = Field(None, description="Truncate the file")
file: str = Field(..., description="Destination file pathpath")
section: Optional[str] = Field(None, description="INI section to write")
key: Optional[str] = Field(None, description="Property key")
value: Optional[str] = Field(None, description="Property value")
data: Optional[dict[str, Any]] = Field(
None, description="Free form data to write"
)
class WriteJson(BaseModel):
merge: Optional[bool] = Field(None, description="Update the file")
file: str = Field(..., description="Destination file pathpath")
data: dict[str, Any] = Field(..., description="Free form data to write")
class Task(BaseModel):
name: Optional["InstallerScript.TaskNameEnum"] = None
prefix: Optional[str] = Field(None, description="The prefix path")
arch: Optional[str] = Field(None, description="Architecture of the prefix")
overrides: Optional[dict[str, str]] = Field(
None, description="WINE DLL overrides"
)
install_gecko: Optional[bool] = Field(
None, description="Enable installing Gecko"
)
install_mono: Optional[bool] = Field(None, description="Enable installing Mono")
executable: Optional[str] = Field(
None, description="Wine executable id or path"
)
args: Optional[str] = Field(None, description="Optional command arguments")
blocking: Optional[bool] = Field(None, description="Suppress threading")
description: Optional[str] = Field(
None,
description="A message be shown to the user during the execution of the task",
)
working_dir: Optional[str] = Field(None, description="Working directory")
exclude_processes: Optional[str] = Field(
None,
description="Space-separated list of processes to exclude from being monitored when determining if the execute phase finished",
)
include_processes: Optional[str] = Field(
None,
description="Space-separated list of processes to monitor when determining if the execute phase finished",
)
env: Optional[dict[str, str]] = Field(None, description="Environment variables")
app: Optional[str] = Field(
None, description="Space-separated list of winetricks apps to run"
)
silent: Optional[bool] = Field(None, description="Run in silent mode")
key: Optional[str] = Field(None, description="Registry key")
value: Optional[str] = Field(None, description="Registry value")
type: Optional[str] = Field(None, description="Registry value type")
filename: Optional[str] = Field(None, description="Registry file")
config_file: Optional[str] = Field(
None, description="File id or path to .conf file"
)
exit: Optional[bool] = Field(
None, description="Exit DOSBox when the executable is terminated"
)
class InputMenu(BaseModel):
description: str = Field(..., description="Dropdown label")
id: Optional[str] = Field(None, description="$INPUT_<id> reference key")
options: list[str | dict[str, Any]] = Field(
..., description="Indented list of `value: label` lines to show"
)
preselect: Optional[str] = Field(None, description="Default selected value")
class InstallerStep(BaseModel):
insert_disc: Optional["InstallerScript.InsertDisc"] = Field(
None, alias="insert-disc"
)
move: Optional["InstallerScript.Move"] = None
merge: Optional["InstallerScript.Merge"] = None
extract: Optional["InstallerScript.Extract"] = None
chmodx: Optional[str] = Field(
None, description="Path to a file to Make executable"
)
execute: Optional["InstallerScript.Execute"] = None
write_file: Optional["InstallerScript.WriteFile"] = None
write_config: Optional["InstallerScript.WriteConfig"] = None
write_json: Optional["InstallerScript.WriteJson"] = None
task: Optional["InstallerScript.Task"] = None
input_menu: Optional["InstallerScript.InputMenu"] = None
class System(BaseModel):
env: Optional[dict[str, str]] = Field(None, description="Environment variables")
terminal: Optional[bool] = Field(
None,
description="Run the game in a terminal if the game is a text based one",
)
single_cpu: Optional[bool] = Field(
None, description="Run the game on a single CPU core"
)
pulse_latency: Optional[bool] = Field(
None, description="Set PulseAudio latency to 60 msecs"
)
use_us_layout: Optional[bool] = Field(
None,
description="Change the keyboard layout to a standard US one while the game is running",
)
class WineOverride(BaseModel):
pass # Additional properties allowed
class WineRunner(BaseModel):
version: Optional[str] = Field(None, description="WINE version override")
Desktop: Optional[bool] = Field(
None, description="Run the game in a Wine virtual desktop"
)
WineDesktop: Optional[str] = Field(
None, description="The resolution of the Wine virtual desktop"
)
dxvk: Optional[bool] = Field(None, description="Enable DXVK")
esync: Optional[bool] = Field(None, description="Enable ESync")
overrides: Optional[list["InstallerScript.WineOverride"]] = Field(
None, description="Overrides for Wine DLLs"
)
class Script(BaseModel):
game: Game = Field(..., description="Game configuration directives")
files: Optional[list["str | InstallerScript.FileItem"]] = Field(
None, description="Fetch required files"
)
installer: list["InstallerScript.InstallerStep"] = Field(
..., description="Installation script"
)
system: Optional["InstallerScript.System"] = Field(
None, description="System configuration directives"
)
wine: Optional["InstallerScript.WineRunner"] = None
name: str = Field(
...,
description="Name of the game, should be surrounded in quotes if containing special characters.",
)
game_slug: str = Field(..., description="Game identifier on the Lutris website")
version: str = Field(..., description="Name of the installer")
slug: str = Field(..., description="Installer identifier")
require_binaries: Optional[str] = Field(
None, alias="require-binaries", description="Additional binaries"
)
requires: Optional[str] = Field(None, description="Mods and add-ons")
extends: Optional[str] = Field(None, description="Extensions / patches")
install_complete_text: Optional[str] = Field(
None, description="Custom end of install text"
)
runner: RunnerEnum = Field(..., description="The runner to be used by this game")
script: Script = Field(..., description="Main script")
class Installer(BaseModel):
id: int
game_id: int
game_slug: str
name: str
year: Optional[int]
user: str
runner: str
slug: str
version: str
description: Optional[str]
notes: str
credits: str
created_at: datetime
updated_at: datetime
draft: bool
published: bool
published_by: Optional[int]
rating: str
is_playable: Optional[bool]
steamid: Optional[int]
gogid: Optional[int]
gogslug: str
humbleid: str
humblestoreid: str
humblestoreid_real: str
script: InstallerScript
content: str
discord_id: Optional[str]
class Config:
populate_by_name = True
class Paginated[T](BaseModel):
count: int
next: Optional[str]
previous: Optional[str]
results: list[T]
class Config:
populate_by_name = True
def update_progress(completed: int, total: int, prefix: str = "Progress"):
"""
Update progress in place using carriage return.
Args:
completed: Number of completed items
total: Total number of items
prefix: Text to show before progress
"""
percentage: float = (completed / total * 100) if total > 0 else 0
bar_length: int = 30
filled_length: int = int(bar_length * completed // total)
bar: str = "█" * filled_length + "░" * (bar_length - filled_length)
progress_text: str = f"\r{prefix}: [{bar}] {completed}/{total} ({percentage:.1f}%)"
sys.stdout.write(progress_text)
sys.stdout.flush()
# Add newline when complete
if completed == total:
print()
def get_page_count(url: str) -> int:
"""
Fetch the first page to calculate total number of pages.
Returns:
Total pages count
"""
print("Fetching first page to determine pagination info...")
try:
response: requests.Response = requests.get(url)
response.raise_for_status()
data = Paginated(**response.json())
total_count: int = data.count
results_per_page: int = len(data.results)
total_pages: int = (
math.ceil(total_count / results_per_page) if results_per_page > 0 else 0
)
print(f"Total count: {total_count}")
print(f"Results per page: {results_per_page}")
print(f"Total pages: {total_pages}")
return total_pages
except requests.exceptions.JSONDecodeError as e:
print(f"Error parsing JSON: {e}")
raise
except requests.exceptions.RequestException as e:
print(f"Error fetching page info: {e}")
raise
def fetch_single_page[T](
_: T, url: str, page_num: int, session: requests.Session
) -> tuple[int, list[T]]:
"""
Fetch a single page of results.
Args:
page_num: Page number to fetch (1-based)
session: Requests session to use
Returns:
Tuple of (page_number, results_list)
"""
url = url + f"?page={page_num}"
try:
response: requests.Response = session.get(url, timeout=30)
response.raise_for_status()
data = Paginated[T].model_validate(response.json())
results: list[T] = data.results
# print(f"Page {page_num}: Fetched {len(results)} results")
return page_num, results
except Exception as e:
print(f"\nError fetching page {page_num}: {e}")
return page_num, []
def fetch_pages_parallel[T](
_: T, url: str, total_pages: int, max_workers: int = 10
) -> list[T]:
"""
Fetch all pages in parallel using ThreadPoolExecutor.
Args:
total_pages: Total number of pages to fetch
max_workers: Maximum number of concurrent workers
Returns:
List of all results from all pages
"""
all_results: list[T] = []
completed_pages: int = 0
print(
f"Starting parallel fetch of {total_pages} pages with {max_workers} workers..."
)
# Create a session for connection pooling
session = requests.Session()
try:
with ThreadPoolExecutor(max_workers=max_workers) as executor:
# Submit all page fetch tasks
future_to_page = {
executor.submit(fetch_single_page, T, url, page_num, session): page_num
for page_num in range(1, total_pages + 1)
}
# Collect results as they complete
page_results: dict[int, list[T]] = {}
for future in as_completed(future_to_page):
page_num: int = future_to_page[future]
try:
page_num_result, results = future.result()
page_results[page_num_result] = cast(list[T], results)
completed_pages += 1
# print(
# f"Progress: {completed_pages}/{total_pages} pages completed "
# f"({completed_pages/total_pages*100:.1f}%)"
# )
update_progress(completed_pages, total_pages, "Fetching pages")
except Exception as e:
print(f"\nPage {page_num} generated an exception: {e}")
page_results[page_num] = []
completed_pages += 1
update_progress(completed_pages, total_pages, "Fetching pages")
# Combine results in correct page order
for page_num in sorted(page_results.keys()):
all_results.extend(page_results[page_num])
finally:
session.close()
print(f"Parallel fetch complete! Total results collected: {len(all_results)}")
return all_results
def fetch_lutris_data_parallel[T](
_: type[T], url: str, max_workers: int = 10
) -> Paginated[T]:
"""
Fetch all data from the Lutris API using parallel requests.
Args:
use_async: Whether to use async/await or ThreadPoolExecutor
max_workers: Maximum number of concurrent workers/requests
Returns:
Dict containing all data
"""
try:
# First, get pagination info
total_pages: int = get_page_count(url)
if total_pages == 0:
print("No pages to fetch!")
return Paginated[T](
count=0,
next=None,
previous=None,
results=[],
)
# Fetch all pages in parallel
start_time: float = time.time()
# Use ThreadPoolExecutor approach
all_results = cast(
list[T], fetch_pages_parallel(T, url, total_pages, max_workers)
)
end_time: float = time.time()
print(f"Total fetch time: {end_time - start_time:.2f} seconds")
print(
f"Average time per page: {(end_time - start_time) / total_pages:.2f} seconds"
)
# Create final data structure
final = Paginated[T](
count=len(all_results),
next=None,
previous=None,
results=all_results,
)
return final
except Exception as e:
print(f"Error in parallel fetch: {e}")
raise
def save_to_json[T](data: Paginated[T], filename: str = "lutris_data.json") -> None:
"""
Save the data to a JSON file.
Args:
data: The data to save
filename: Output filename
"""
try:
with open(filename, "w", encoding="utf-8") as f:
f.write(data.model_dump_json(indent=2))
print(f"Data successfully saved to {filename}")
# Print some statistics
print(f"File contains {data.count} results")
except IOError as e:
print(f"Error saving to file: {e}")
def main():
"""Main function to orchestrate the parallel fetching and saving process."""
try:
# Fetch all games data in parallel
print("Fetching Games...")
games = fetch_lutris_data_parallel(Game, LUTRIS_API_GAMES)
save_to_json(games, "lutris_games.json")
# Fetch all installers data in parallel
print("Fetching Installers...")
installers = fetch_lutris_data_parallel(Installer, LUTRIS_API_INSTALLERS)
save_to_json(installers, "lutris_installers.json")
except KeyboardInterrupt:
print("\nOperation interrupted by user")
except Exception as e:
print(f"An unexpected error occurred: {e}")
if __name__ == "__main__":
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment