File size: 3,988 Bytes
2cd1b38 4b7df59 2cd1b38 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 | """Apply schema migrations and seed to a hosted Supabase project.
Secret keys (`sb_secret_...`) authenticate the Data API (review backend and analytics).
They cannot run DDL. Schema changes use the Management API with a personal
access token (`sbp_...`), which is the current programmatic equivalent of
`supabase db push`.
Docs:
https://supabase.com/docs/guides/deployment/database-migrations
https://supabase.com/docs/reference/api/v1-apply-a-migration
https://supabase.com/docs/guides/getting-started/api-keys
"""
from __future__ import annotations
import sys
from pathlib import Path
import httpx
PROJECT_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
from app.config import (
get_supabase_access_token,
get_supabase_project_ref,
)
# / acts as a shortcut to join file paths together cleanly.
# In this case: PROJECT_ROOT/supabase/migrations
MIGRATIONS_DIR = PROJECT_ROOT / "supabase" / "migrations"
SEED_PATH = PROJECT_ROOT / "supabase" / "seed.sql"
MANAGEMENT_API = "https://api.supabase.com/v1"
def _headers(token: str) -> dict[str, str]:
return {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
}
def _migration_name(path: Path) -> str:
return path.stem
def list_applied(client: httpx.Client, ref: str) -> set[str]:
response = client.get(f"{MANAGEMENT_API}/projects/{ref}/database/migrations")
response.raise_for_status()
payload = response.json()
applied: set[str] = set()
if isinstance(payload, list):
for row in payload:
if isinstance(row, dict):
if row.get("version") is not None:
applied.add(str(row["version"]))
if row.get("name") is not None:
applied.add(str(row["name"]))
return applied
def is_migration_applied(applied: set[str], path: Path) -> bool:
name = _migration_name(path)
if name in applied:
return True
# Management API may record a server-assigned version distinct from the filename prefix.
timestamp_prefix = name.split("_", 1)[0]
return timestamp_prefix in applied
def apply_migration(client: httpx.Client, ref: str, path: Path) -> None:
query = path.read_text(encoding="utf-8")
name = _migration_name(path)
response = client.post(
f"{MANAGEMENT_API}/projects/{ref}/database/migrations",
json={"query": query, "name": name},
)
if not response.is_success:
raise RuntimeError(
f"Migration {name} failed ({response.status_code}): {response.text}"
)
print(f"Applied migration {name}")
def apply_seed(client: httpx.Client, ref: str) -> None:
check = client.post(
f"{MANAGEMENT_API}/projects/{ref}/database/query",
json={"query": "select 1 from invites limit 1"},
)
if check.is_success and check.json():
print("Skipping seed (invites already present)")
else:
query = SEED_PATH.read_text(encoding="utf-8")
response = client.post(
f"{MANAGEMENT_API}/projects/{ref}/database/query",
json={"query": query},
)
if not response.is_success:
raise RuntimeError(f"Seed failed ({response.status_code}): {response.text}")
print("Applied supabase/seed.sql")
def main() -> None:
token = get_supabase_access_token()
ref = get_supabase_project_ref()
files = sorted(MIGRATIONS_DIR.glob("*.sql"))
if not files:
raise RuntimeError(f"No SQL files in {MIGRATIONS_DIR}")
with httpx.Client(headers=_headers(token), timeout=180.0) as client:
applied = list_applied(client, ref)
for path in files:
name = _migration_name(path)
if is_migration_applied(applied, path):
print(f"Skipping already-applied {name}")
continue
apply_migration(client, ref, path)
apply_seed(client, ref)
if __name__ == "__main__":
main()
|