Spaces:
Running on Zero
Running on Zero
File size: 1,805 Bytes
b37ebd1 7982d01 b37ebd1 7982d01 b37ebd1 7982d01 d896946 34c3133 7982d01 b37ebd1 c3b9247 b37ebd1 c3b9247 b37ebd1 7982d01 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 | 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
|