Add boss seed data pipeline for export and import

Add seeder support for boss battles so new database instances come
pre-populated. Adds --export-bosses CLI flag to dump boss data from the
database to JSON seed files, and loads those files during normal seeding.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-02-08 12:36:08 +01:00
parent 76855f4f56
commit 053dece33e
7 changed files with 250 additions and 27 deletions

View File

@@ -1,9 +1,11 @@
"""Database upsert helpers for seed data."""
from sqlalchemy import select
from sqlalchemy import delete, select
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.boss_battle import BossBattle
from app.models.boss_pokemon import BossPokemon
from app.models.evolution import Evolution
from app.models.game import Game
from app.models.pokemon import Pokemon
@@ -204,14 +206,69 @@ async def upsert_route_encounters(
return count
async def upsert_bosses(
session: AsyncSession,
version_group_id: int,
bosses: list[dict],
dex_to_id: dict[int, int],
) -> int:
"""Upsert boss battles for a version group, return count of bosses upserted."""
count = 0
for boss in bosses:
# Upsert the boss battle on (version_group_id, order) conflict
stmt = insert(BossBattle).values(
version_group_id=version_group_id,
name=boss["name"],
boss_type=boss["boss_type"],
badge_name=boss.get("badge_name"),
badge_image_url=boss.get("badge_image_url"),
level_cap=boss["level_cap"],
order=boss["order"],
location=boss["location"],
sprite_url=boss.get("sprite_url"),
).on_conflict_do_update(
constraint="uq_boss_battles_version_group_order",
set_={
"name": boss["name"],
"boss_type": boss["boss_type"],
"badge_name": boss.get("badge_name"),
"badge_image_url": boss.get("badge_image_url"),
"level_cap": boss["level_cap"],
"location": boss["location"],
"sprite_url": boss.get("sprite_url"),
},
).returning(BossBattle.id)
result = await session.execute(stmt)
boss_id = result.scalar_one()
# Delete existing boss_pokemon for this boss, then re-insert
await session.execute(
delete(BossPokemon).where(BossPokemon.boss_battle_id == boss_id)
)
for bp in boss.get("pokemon", []):
pokemon_id = dex_to_id.get(bp["pokeapi_id"])
if pokemon_id is None:
print(f" Warning: no pokemon_id for pokeapi_id {bp['pokeapi_id']}")
continue
session.add(BossPokemon(
boss_battle_id=boss_id,
pokemon_id=pokemon_id,
level=bp["level"],
order=bp["order"],
))
count += 1
await session.flush()
return count
async def upsert_evolutions(
session: AsyncSession,
evolutions: list[dict],
dex_to_id: dict[int, int],
) -> int:
"""Upsert evolution pairs, return count of upserted rows."""
# Clear existing evolutions and re-insert (simpler than complex upsert)
from sqlalchemy import delete
await session.execute(delete(Evolution))
count = 0