Function create_user

apps/api/src/services/users/users.py:51–191

(
    request: Request,
    db_session: AsyncSession,
    current_user: PublicUser | AnonymousUser,
    user_object: UserCreate,
    org_id: int,
    is_oauth: bool = False,
    signup_provider: str = "email",
)

async def create_user(
    request: Request,
    db_session: AsyncSession,
    current_user: PublicUser | AnonymousUser,
    user_object: UserCreate,
    org_id: int,
    is_oauth: bool = False,
    signup_provider: str = "email",
):
    # Validate password complexity (skip for OAuth users who have empty passwords)
    if user_object.password and not is_oauth:
        validation_result = validate_password_complexity(user_object.password)
        if not validation_result.is_valid:
            raise HTTPException(
                status_code=400,
                detail={
                    "code": "WEAK_PASSWORD",
                    "message": "Password does not meet security requirements",
                    "errors": validation_result.errors,
                    "requirements": validation_result.requirements,
                },
            )

    user = User.model_validate(user_object)

    # RBAC check
    await rbac_check(request, current_user, "create", "user_x", db_session)

    # Complete the user object
    user.user_uuid = f"user_{uuid4()}"
    user.password = security_hash_password(user_object.password) if user_object.password else ""

    # OAuth users and OSS mode get auto-verified email
    if is_oauth or get_deployment_mode() != 'saas':
        user.email_verified = True
        user.email_verified_at = datetime.now(timezone.utc).isoformat()
        user.signup_method = signup_provider if is_oauth else "email"
    else:
        user.email_verified = False
        user.email_verified_at = None
        user.signup_method = "email"

    user.creation_date = str(datetime.now())
    user.update_date = str(datetime.now())

    # Verifications

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

    if not result:
        raise HTTPException(
            status_code=400,
            detail="Organization does not exist",
        )

    # Usage check