52 lines
1.1 KiB
Python
52 lines
1.1 KiB
Python
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from domains.users.repo import get_user_by_id
|
|
from domains.groups.models import Group, GroupMember
|
|
from domains.groups.repo import (
|
|
create_group,
|
|
get_group_by_id,
|
|
add_group_member,
|
|
get_user_groups
|
|
)
|
|
|
|
|
|
async def create_new_group(
|
|
db: AsyncSession,
|
|
name: str,
|
|
):
|
|
|
|
group = Group(
|
|
name=name,
|
|
)
|
|
|
|
return await create_group(db, group)
|
|
|
|
async def add_member_to_group(
|
|
db: AsyncSession,
|
|
group_id,
|
|
user_id,
|
|
):
|
|
# 1. Check if group exists
|
|
group = await get_group_by_id(db, group_id)
|
|
if not group:
|
|
raise ValueError("Group not found")
|
|
|
|
# 2. Check if user exists
|
|
user = await get_user_by_id(db, user_id)
|
|
if not user:
|
|
raise ValueError("User not found")
|
|
|
|
# TODO: Check if already a member to avoid duplicate constraint if any (optional)
|
|
|
|
membership = GroupMember(
|
|
group_id=group_id,
|
|
user_id=user_id,
|
|
)
|
|
return await add_group_member(db, membership)
|
|
|
|
|
|
async def list_user_groups(
|
|
db: AsyncSession,
|
|
user_id
|
|
):
|
|
return await get_user_groups(db, user_id) |