70 lines
2.1 KiB
Python
70 lines
2.1 KiB
Python
from datetime import datetime
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.db.models import Activity, ActivityStatus, HealthState, SyncUser
|
|
|
|
|
|
class UserRepository:
|
|
def __init__(self, session: Session) -> None:
|
|
self.session = session
|
|
|
|
def create(self, **values) -> SyncUser:
|
|
user = SyncUser(**values)
|
|
self.session.add(user)
|
|
self.session.commit()
|
|
return user
|
|
|
|
def get(self, user_id: int) -> SyncUser | None:
|
|
return self.session.get(SyncUser, user_id)
|
|
|
|
def list_enabled(self) -> list[SyncUser]:
|
|
return list(self.session.scalars(select(SyncUser).where(SyncUser.enabled.is_(True)).order_by(SyncUser.id)))
|
|
|
|
|
|
class ActivityRepository:
|
|
def __init__(self, session: Session) -> None:
|
|
self.session = session
|
|
|
|
def get_or_create_discovered(
|
|
self,
|
|
*,
|
|
user_id: int,
|
|
mywhoosh_activity_id: str,
|
|
activity_name: str,
|
|
activity_timestamp: datetime | None,
|
|
) -> tuple[Activity, bool]:
|
|
existing = self.session.scalar(
|
|
select(Activity).where(
|
|
Activity.user_id == user_id,
|
|
Activity.mywhoosh_activity_id == mywhoosh_activity_id,
|
|
)
|
|
)
|
|
if existing is not None:
|
|
return existing, False
|
|
activity = Activity(
|
|
user_id=user_id,
|
|
mywhoosh_activity_id=mywhoosh_activity_id,
|
|
activity_name=activity_name,
|
|
activity_timestamp=activity_timestamp,
|
|
status=ActivityStatus.DISCOVERED,
|
|
last_completed_stage=ActivityStatus.DISCOVERED,
|
|
)
|
|
self.session.add(activity)
|
|
try:
|
|
self.session.commit()
|
|
except IntegrityError:
|
|
self.session.rollback()
|
|
existing = self.session.scalar(
|
|
select(Activity).where(
|
|
Activity.user_id == user_id,
|
|
Activity.mywhoosh_activity_id == mywhoosh_activity_id,
|
|
)
|
|
)
|
|
if existing is None:
|
|
raise
|
|
return existing, False
|
|
return activity, True
|