Cheng Zhou
69 lines
2.3 KiB
Python
69 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from typing import Sequence
|
|
|
|
from sqlalchemy import delete, select, update
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.study_member import StudyMember
|
|
from app.schemas.member import StudyMemberCreate, StudyMemberUpdate
|
|
|
|
|
|
async def get_member(db: AsyncSession, study_id: uuid.UUID, user_id: uuid.UUID) -> StudyMember | None:
|
|
result = await db.execute(
|
|
select(StudyMember).where(StudyMember.study_id == study_id, StudyMember.user_id == user_id)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def get_member_by_id(db: AsyncSession, member_id: uuid.UUID) -> StudyMember | None:
|
|
result = await db.execute(select(StudyMember).where(StudyMember.id == member_id))
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def add_member(db: AsyncSession, study_id: uuid.UUID, member_in: StudyMemberCreate) -> StudyMember:
|
|
member = StudyMember(
|
|
study_id=study_id,
|
|
user_id=member_in.user_id,
|
|
role_in_study=member_in.role_in_study,
|
|
is_active=member_in.is_active,
|
|
)
|
|
db.add(member)
|
|
await db.commit()
|
|
await db.refresh(member)
|
|
return member
|
|
|
|
|
|
async def list_members(db: AsyncSession, study_id: uuid.UUID, skip: int = 0, limit: int = 100) -> Sequence[StudyMember]:
|
|
result = await db.execute(
|
|
select(StudyMember).where(StudyMember.study_id == study_id).offset(skip).limit(limit)
|
|
)
|
|
return result.scalars().all()
|
|
|
|
|
|
async def update_member(db: AsyncSession, member: StudyMember, member_in: StudyMemberUpdate) -> StudyMember:
|
|
update_data = {k: v for k, v in member_in.model_dump(exclude_unset=True).items() if v is not None}
|
|
if update_data:
|
|
await db.execute(
|
|
update(StudyMember)
|
|
.where(StudyMember.id == member.id)
|
|
.values(**update_data)
|
|
)
|
|
await db.commit()
|
|
await db.refresh(member)
|
|
return member
|
|
|
|
|
|
async def remove_member(db: AsyncSession, member: StudyMember) -> StudyMember:
|
|
await db.execute(delete(StudyMember).where(StudyMember.id == member.id))
|
|
await db.commit()
|
|
return member
|
|
|
|
|
|
async def user_has_memberships(db: AsyncSession, user_id: uuid.UUID) -> bool:
|
|
result = await db.execute(select(StudyMember.id).where(StudyMember.user_id == user_id))
|
|
return result.first() is not None
|