57 lines
1.9 KiB
Python
57 lines
1.9 KiB
Python
"""
|
|
Check class membership health after migration / repair.
|
|
|
|
Usage:
|
|
python check_membership_health.py
|
|
Or inside Docker:
|
|
docker compose exec backend python check_membership_health.py
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
from sqlalchemy import func, select
|
|
|
|
from app.db.database import async_session
|
|
from app.db.models import ClassMembership, Class_, User
|
|
|
|
|
|
async def main() -> None:
|
|
async with async_session() as db:
|
|
total_users = (await db.execute(select(func.count(User.id)))).scalar() or 0
|
|
total_classes = (await db.execute(select(func.count(Class_.id)))).scalar() or 0
|
|
total_memberships = (await db.execute(select(func.count(ClassMembership.id)))).scalar() or 0
|
|
|
|
users_without_membership = (
|
|
await db.execute(
|
|
select(User.id, User.name, User.email, User.role)
|
|
.outerjoin(ClassMembership, ClassMembership.user_id == User.id)
|
|
.where(User.role != "super_admin")
|
|
.group_by(User.id)
|
|
.having(func.count(ClassMembership.id) == 0)
|
|
)
|
|
).all()
|
|
|
|
classes_without_members = (
|
|
await db.execute(
|
|
select(Class_.id, Class_.name)
|
|
.outerjoin(ClassMembership, ClassMembership.class_id == Class_.id)
|
|
.group_by(Class_.id)
|
|
.having(func.count(ClassMembership.id) == 0)
|
|
)
|
|
).all()
|
|
|
|
print(f"users={total_users} classes={total_classes} memberships={total_memberships}")
|
|
print(f"users_without_membership={len(users_without_membership)}")
|
|
for row in users_without_membership[:20]:
|
|
print(" user_without_membership", row)
|
|
|
|
print(f"classes_without_members={len(classes_without_members)}")
|
|
for row in classes_without_members[:20]:
|
|
print(" class_without_members", row)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|