zzstoatzz.io
search for standard sites
pub-search.waow.tech
search
zig
blog
atproto
1#!/usr/bin/env -S uv run --script --quiet
2# /// script
3# requires-python = ">=3.12"
4# dependencies = ["httpx", "pydantic-settings"]
5# ///
6"""Check what tables exist in the database."""
7
8import os
9import httpx
10from pydantic_settings import BaseSettings, SettingsConfigDict
11
12
13class Settings(BaseSettings):
14 model_config = SettingsConfigDict(
15 env_file=os.environ.get("ENV_FILE", ".env"), extra="ignore"
16 )
17 turso_url: str
18 turso_token: str
19
20 @property
21 def turso_host(self) -> str:
22 url = self.turso_url
23 if url.startswith("libsql://"):
24 url = url[len("libsql://"):]
25 return url
26
27
28settings = Settings() # type: ignore
29
30response = httpx.post(
31 f"https://{settings.turso_host}/v2/pipeline",
32 headers={
33 "Authorization": f"Bearer {settings.turso_token}",
34 "Content-Type": "application/json",
35 },
36 json={
37 "requests": [
38 {"type": "execute", "stmt": {"sql": "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"}},
39 {"type": "close"},
40 ]
41 },
42 timeout=30,
43)
44response.raise_for_status()
45data = response.json()
46
47result = data["results"][0]
48if result["type"] == "error":
49 print(f"Error: {result['error']}")
50else:
51 rows = result["response"]["result"]["rows"]
52 print("Tables in database:")
53 for row in rows:
54 name = row[0]["value"] if isinstance(row[0], dict) else row[0]
55 print(f" - {name}")