feat(admin): afficher l'email dans subscriptions et subscriber_state

user_id seul n'est pas assez parlant pour débugger ; jointure LEFT JOIN
sur users pour afficher l'email en colonne (lecture seule).
This commit is contained in:
Colin Maudry
2026-07-06 10:42:53 +02:00
parent e8f74aa529
commit fb3292d41e
2 changed files with 50 additions and 2 deletions
+22 -2
View File
@@ -15,6 +15,8 @@ class TableConfig:
column_types: dict[str, type]
dropdowns: dict[str, list[str]]
target_user_id: Callable[[dict], int | None]
sort_by: str
join_user_email: bool = False
TABLES: dict[str, TableConfig] = {
@@ -40,11 +42,13 @@ TABLES: dict[str, TableConfig] = {
},
dropdowns={"email_verified": ["0", "1"]},
target_user_id=lambda row: row["id"],
sort_by="updated_at",
),
"subscriptions": TableConfig(
columns=[
"id",
"user_id",
"email",
"frisbii_customer_handle",
"frisbii_subscription_handle",
"plan",
@@ -67,10 +71,13 @@ TABLES: dict[str, TableConfig] = {
"plan": list(PLANS.keys()),
},
target_user_id=lambda row: row["user_id"],
sort_by="updated_at",
join_user_email=True,
),
"subscriber_state": TableConfig(
columns=[
"user_id",
"email",
"trial_used",
"votes_balance",
"votes_last_credited_at",
@@ -87,6 +94,8 @@ TABLES: dict[str, TableConfig] = {
},
dropdowns={"trial_used": ["0", "1"]},
target_user_id=lambda row: row["user_id"],
sort_by="updated_at",
join_user_email=True,
),
"admin_actions": TableConfig(
columns=[
@@ -102,14 +111,25 @@ TABLES: dict[str, TableConfig] = {
column_types={},
dropdowns={},
target_user_id=lambda row: None,
sort_by="created_at",
),
}
def get_rows(table: str) -> list[dict]:
cfg = TABLES[table]
cols_sql = ", ".join(cfg.columns)
rows = get_conn().execute(f"SELECT {cols_sql} FROM {table}").fetchall()
if cfg.join_user_email:
real_cols = [col for col in cfg.columns if col != "email"]
cols_sql = ", ".join(f"{table}.{col}" for col in real_cols)
query = (
f"SELECT {cols_sql}, users.email AS email FROM {table} "
f"LEFT JOIN users ON users.id = {table}.user_id "
f"ORDER BY {table}.{cfg.sort_by} DESC"
)
else:
cols_sql = ", ".join(cfg.columns)
query = f"SELECT {cols_sql} FROM {table} ORDER BY {cfg.sort_by} DESC"
rows = get_conn().execute(query).fetchall()
return [dict(row) for row in rows]
+28
View File
@@ -15,6 +15,34 @@ def test_get_rows_users_excludes_password_hash(users_db_path):
assert "password_hash" not in rows[0]
def test_get_rows_subscriptions_includes_user_email(users_db_path):
from src.auth import db as auth_db
from src.subscriptions import db as sub_db
auth_db.init_schema()
sub_db.init_schema()
uid = auth_db.create_user("a@ex.fr", "hash")
sub_db.create_pending(uid, "cust-1", "simple")
rows = tables.get_rows("subscriptions")
assert rows[0]["email"] == "a@ex.fr"
def test_get_rows_subscriber_state_includes_user_email(users_db_path):
from src.auth import db as auth_db
from src.subscriptions import db as sub_db
auth_db.init_schema()
sub_db.init_schema()
uid = auth_db.create_user("a@ex.fr", "hash")
sub_db.create_pending(uid, "cust-1", "simple")
rows = tables.get_rows("subscriber_state")
assert rows[0]["email"] == "a@ex.fr"
def test_set_cell_rejects_unknown_table(users_db_path):
with pytest.raises(ValueError):
tables.set_cell("not_a_table", 1, "email", "x@ex.fr")