Spaces:
Running on Zero
Running on Zero
| from typing import Optional | |
| from aiogram.types import User as TgUser | |
| from sqlalchemy import inspect | |
| from sqlalchemy.ext.asyncio import AsyncSession | |
| from sqlalchemy.orm import make_transient_to_detached | |
| from app.config import get_settings | |
| from app.db.models import User | |
| from app.db.repos import users as repo | |
| SNAPSHOT_COLS = ( | |
| "id", "username", "full_name", "lang", "stream", "year", "referred_by", | |
| "gate_passed", "points", "streak", "best_streak", "last_active_on", "is_banned", | |
| "active_days", "week_key", "week_xp", "day_key", "day_xp", "trust", | |
| "source", "pending_link", "share_prompt_day", "credit_name", "gate_missing", | |
| ) | |
| async def get_or_create( | |
| session: AsyncSession, tg: TgUser, referrer_hint: Optional[int] = None, source: Optional[str] = None | |
| ) -> User: | |
| user = await repo.get(session, tg.id) | |
| if user: | |
| return user | |
| referred_by = None | |
| if referrer_hint and referrer_hint != tg.id and await repo.get(session, referrer_hint): | |
| referred_by = referrer_hint | |
| return await repo.create( | |
| session, | |
| user_id=tg.id, | |
| username=tg.username, | |
| full_name=tg.full_name or "", | |
| lang=get_settings().default_lang, | |
| referred_by=referred_by, | |
| source=source, | |
| ) | |
| def snapshot(user: User) -> Optional[dict]: | |
| """Plain-dict copy of the user row, or None if some column is not loaded.""" | |
| loaded = inspect(user).dict | |
| if any(col not in loaded for col in SNAPSHOT_COLS): | |
| return None | |
| return {col: loaded[col] for col in SNAPSHOT_COLS} | |
| def from_snapshot(data: dict) -> User: | |
| """Rebuild a clean, detached User (no DB query). Attach with session.add().""" | |
| user = User(**data) | |
| make_transient_to_detached(user) # treat attributes as freshly loaded, no pending changes | |
| return user | |