Function create_usergroup

apps/api/src/services/users/usergroups.py:106–167

(
    request: Request,
    db_session: AsyncSession,
    current_user: PublicUser | AnonymousUser,
    usergroup_create: UserGroupCreate,
)

async def create_usergroup(
    request: Request,
    db_session: AsyncSession,
    current_user: PublicUser | AnonymousUser,
    usergroup_create: UserGroupCreate,
) -> UserGroupRead:

    from src.security.auth import resolve_acting_user_id

    usergroup = UserGroup.model_validate(usergroup_create)

    await require_org_membership(
        resolve_acting_user_id(current_user), usergroup_create.org_id, db_session
    )

    # RBAC check
    await rbac_check(
        request,
        usergroup_uuid="usergroup_X",
        current_user=current_user,
        action="create",
        db_session=db_session,
    )

    # Check if Organization exists
    statement = select(Organization).where(Organization.id == usergroup_create.org_id)
    org = (await db_session.execute(statement)).scalars().first()

    if not org or org.id is None:
        raise HTTPException(
            status_code=400,
            detail="Organization does not exist",
        )

    # Usage check
    await check_limits_with_usage("courses", org.id, db_session)

    # Complete the object
    usergroup.usergroup_uuid = f"usergroup_{uuid4()}"
    usergroup.creation_date = str(datetime.now())
    usergroup.update_date = str(datetime.now())

    # Save the object
    db_session.add(usergroup)
    await db_session.commit()
    await db_session.refresh(usergroup)

    # Feature usage
    await increase_feature_usage("usergroups", org.id, db_session)

    await dispatch_webhooks(
        event_name="usergroup_created",
        org_id=org.id,
        data={
            "usergroup_uuid": usergroup.usergroup_uuid,
            "name": usergroup.name,
        },
    )