music on atproto
plyr.fm
zzstoatzz.io
1#!/usr/bin/env -S uv run --script
2# /// script
3# requires-python = ">=3.11"
4# dependencies = ["redis"]
5# ///
6"""check recent docket task runs.
7
8usage:
9 ./scripts/docket_runs.py # uses DOCKET_URL from env
10 ./scripts/docket_runs.py --env staging # uses staging redis
11 ./scripts/docket_runs.py --env production # uses production redis
12 ./scripts/docket_runs.py --limit 20 # show more runs
13"""
14
15import argparse
16import os
17
18import redis
19
20
21def main():
22 parser = argparse.ArgumentParser(description="check docket task runs")
23 parser.add_argument(
24 "--env",
25 choices=["local", "staging", "production"],
26 default="local",
27 help="environment to check (default: local, uses DOCKET_URL)",
28 )
29 parser.add_argument(
30 "--limit", type=int, default=10, help="number of runs to show (default: 10)"
31 )
32 args = parser.parse_args()
33
34 # get redis url
35 if args.env == "local":
36 url = os.environ.get("DOCKET_URL", "redis://localhost:6379")
37 elif args.env == "staging":
38 url = os.environ.get("DOCKET_URL_STAGING")
39 if not url:
40 print("error: DOCKET_URL_STAGING not set")
41 print(
42 "hint: export DOCKET_URL_STAGING=rediss://default:xxx@xxx.upstash.io:6379"
43 )
44 return 1
45 elif args.env == "production":
46 url = os.environ.get("DOCKET_URL_PRODUCTION")
47 if not url:
48 print("error: DOCKET_URL_PRODUCTION not set")
49 print(
50 "hint: export DOCKET_URL_PRODUCTION=rediss://default:xxx@xxx.upstash.io:6379"
51 )
52 return 1
53
54 print(f"connecting to {args.env}...")
55 r = redis.from_url(url)
56
57 # get all run keys
58 keys = r.keys("plyr:runs:*")
59 if not keys:
60 print("no runs found")
61 return 0
62
63 print(f"found {len(keys)} total runs, showing last {args.limit}:\n")
64
65 for key in sorted(keys, reverse=True)[: args.limit]:
66 data = r.hgetall(key)
67 run_id = key.decode().split(":")[-1]
68
69 # extract fields safely
70 function = data.get(b"function", b"?").decode()
71 state = data.get(b"state", b"?").decode()
72 started = (
73 data.get(b"started_at", b"").decode()[:19]
74 if data.get(b"started_at")
75 else "?"
76 )
77 completed = (
78 data.get(b"completed_at", b"").decode()[:19]
79 if data.get(b"completed_at")
80 else "-"
81 )
82 # state emoji
83 emoji = {"completed": "✓", "failed": "✗", "running": "⋯"}.get(state, "?")
84
85 print(
86 f"{emoji} {run_id[:8]} {function:<20} {state:<10} {started} → {completed}"
87 )
88
89 return 0
90
91
92if __name__ == "__main__":
93 raise SystemExit(main())