39 lines
1.3 KiB
Python
39 lines
1.3 KiB
Python
"""User repository"""
|
|
|
|
from typing import Optional
|
|
from sqlalchemy.orm import Session
|
|
from app.db.models import User
|
|
from app.db.repositories.base import BaseRepository
|
|
|
|
|
|
class UserRepository(BaseRepository[User]):
|
|
"""User data access operations"""
|
|
|
|
def __init__(self, session: Session):
|
|
super().__init__(session, User)
|
|
|
|
def get_by_telegram_id(self, telegram_id: int) -> Optional[User]:
|
|
"""Get user by Telegram ID"""
|
|
return self.session.query(User).filter(User.telegram_id == telegram_id).first()
|
|
|
|
def get_by_username(self, username: str) -> Optional[User]:
|
|
"""Get user by username"""
|
|
return self.session.query(User).filter(User.username == username).first()
|
|
|
|
def get_or_create(self, telegram_id: int, **kwargs) -> User:
|
|
"""Get user or create if doesn't exist"""
|
|
user = self.get_by_telegram_id(telegram_id)
|
|
if not user:
|
|
user = self.create(telegram_id=telegram_id, **kwargs)
|
|
return user
|
|
|
|
def update_activity(self, telegram_id: int) -> Optional[User]:
|
|
"""Update user's last activity timestamp"""
|
|
from datetime import datetime
|
|
user = self.get_by_telegram_id(telegram_id)
|
|
if user:
|
|
user.last_activity = datetime.utcnow()
|
|
self.session.commit()
|
|
self.session.refresh(user)
|
|
return user
|