refac
This commit is contained in:
@@ -379,14 +379,69 @@ class UsersTable:
|
|||||||
row = (await session.execute(query)).scalars().first()
|
row = (await session.execute(query)).scalars().first()
|
||||||
return UserModel.model_validate(row) if row else None
|
return UserModel.model_validate(row) if row else None
|
||||||
|
|
||||||
async def get_users(
|
async def get_scim_users(
|
||||||
self,
|
self,
|
||||||
filter: dict | None = None,
|
filter: dict | None = None,
|
||||||
|
sort: dict | None = None,
|
||||||
skip: int | None = None,
|
skip: int | None = None,
|
||||||
limit: int | None = None,
|
limit: int | None = None,
|
||||||
db: AsyncSession | None = None,
|
db: AsyncSession | None = None,
|
||||||
) -> dict:
|
) -> dict:
|
||||||
"""Paginated user listing with optional filters for role, group, and channel."""
|
async with get_async_db_context(db) as session:
|
||||||
|
stmt = select(User).where(or_(User.oauth.cast(String) != 'null', User.scim.cast(String) != 'null'))
|
||||||
|
|
||||||
|
if filter:
|
||||||
|
user_id = filter.get('id')
|
||||||
|
if user_id:
|
||||||
|
stmt = stmt.where(User.id == user_id)
|
||||||
|
|
||||||
|
email = filter.get('email')
|
||||||
|
if email:
|
||||||
|
stmt = stmt.where(func.lower(User.email) == email.lower())
|
||||||
|
|
||||||
|
order_by = sort.get('order_by') if sort else None
|
||||||
|
direction = sort.get('direction') if sort else None
|
||||||
|
|
||||||
|
if order_by == 'created_at':
|
||||||
|
stmt = stmt.order_by(User.created_at.asc() if direction == 'asc' else User.created_at.desc())
|
||||||
|
|
||||||
|
count_result = await session.execute(select(func.count()).select_from(stmt.subquery()))
|
||||||
|
total = count_result.scalar()
|
||||||
|
|
||||||
|
if skip is not None:
|
||||||
|
stmt = stmt.offset(skip)
|
||||||
|
if limit is not None:
|
||||||
|
stmt = stmt.limit(limit)
|
||||||
|
|
||||||
|
result = await session.execute(stmt)
|
||||||
|
users = result.scalars().all()
|
||||||
|
return {
|
||||||
|
'users': [UserModel.model_validate(user) for user in users],
|
||||||
|
'total': total,
|
||||||
|
}
|
||||||
|
|
||||||
|
async def get_scim_user_by_id(
|
||||||
|
self,
|
||||||
|
id: str,
|
||||||
|
db: AsyncSession | None = None,
|
||||||
|
) -> UserModel | None:
|
||||||
|
async with get_async_db_context(db) as session:
|
||||||
|
stmt = select(User).where(
|
||||||
|
User.id == id,
|
||||||
|
or_(User.oauth.cast(String) != 'null', User.scim.cast(String) != 'null'),
|
||||||
|
)
|
||||||
|
user = (await session.execute(stmt)).scalars().first()
|
||||||
|
return UserModel.model_validate(user) if user else None
|
||||||
|
|
||||||
|
async def get_users(
|
||||||
|
self,
|
||||||
|
filter: dict | None = None,
|
||||||
|
sort: dict | None = None,
|
||||||
|
skip: int | None = None,
|
||||||
|
limit: int | None = None,
|
||||||
|
db: AsyncSession | None = None,
|
||||||
|
) -> dict:
|
||||||
|
"""Paginated user listing with optional filters and sort."""
|
||||||
async with get_async_db_context(db) as session:
|
async with get_async_db_context(db) as session:
|
||||||
# Deferred imports to avoid circular dependencies
|
# Deferred imports to avoid circular dependencies
|
||||||
from open_webui.models.channels import ChannelMember
|
from open_webui.models.channels import ChannelMember
|
||||||
@@ -447,64 +502,63 @@ class UsersTable:
|
|||||||
if exclude_roles:
|
if exclude_roles:
|
||||||
stmt = stmt.filter(~User.role.in_(exclude_roles))
|
stmt = stmt.filter(~User.role.in_(exclude_roles))
|
||||||
|
|
||||||
order_by = filter.get('order_by')
|
order_by = sort.get('order_by') if sort else None
|
||||||
direction = filter.get('direction')
|
direction = sort.get('direction') if sort else None
|
||||||
|
|
||||||
if order_by and order_by.startswith('group_id:'):
|
if order_by and order_by.startswith('group_id:'):
|
||||||
group_id = order_by.split(':', 1)[1]
|
group_id = order_by.split(':', 1)[1]
|
||||||
|
|
||||||
# Subquery that checks if the user belongs to the group
|
# Subquery that checks if the user belongs to the group
|
||||||
membership_exists = exists(
|
membership_exists = exists(
|
||||||
select(GroupMember.id).where(
|
select(GroupMember.id).where(
|
||||||
GroupMember.user_id == User.id,
|
GroupMember.user_id == User.id,
|
||||||
GroupMember.group_id == group_id,
|
GroupMember.group_id == group_id,
|
||||||
)
|
|
||||||
)
|
)
|
||||||
|
)
|
||||||
|
|
||||||
# CASE: user in group → 1, user not in group → 0
|
# CASE: user in group → 1, user not in group → 0
|
||||||
group_sort = case((membership_exists, 1), else_=0)
|
group_sort = case((membership_exists, 1), else_=0)
|
||||||
|
|
||||||
if direction == 'asc':
|
if direction == 'asc':
|
||||||
stmt = stmt.order_by(group_sort.asc(), User.name.asc())
|
stmt = stmt.order_by(group_sort.asc(), User.name.asc())
|
||||||
else:
|
else:
|
||||||
stmt = stmt.order_by(group_sort.desc(), User.name.asc())
|
stmt = stmt.order_by(group_sort.desc(), User.name.asc())
|
||||||
|
|
||||||
elif order_by == 'name':
|
elif order_by == 'name':
|
||||||
if direction == 'asc':
|
if direction == 'asc':
|
||||||
stmt = stmt.order_by(User.name.asc())
|
stmt = stmt.order_by(User.name.asc())
|
||||||
else:
|
else:
|
||||||
stmt = stmt.order_by(User.name.desc())
|
stmt = stmt.order_by(User.name.desc())
|
||||||
|
|
||||||
elif order_by == 'email':
|
elif order_by == 'email':
|
||||||
if direction == 'asc':
|
if direction == 'asc':
|
||||||
stmt = stmt.order_by(User.email.asc())
|
stmt = stmt.order_by(User.email.asc())
|
||||||
else:
|
else:
|
||||||
stmt = stmt.order_by(User.email.desc())
|
stmt = stmt.order_by(User.email.desc())
|
||||||
|
|
||||||
elif order_by == 'created_at':
|
elif order_by == 'created_at':
|
||||||
if direction == 'asc':
|
if direction == 'asc':
|
||||||
stmt = stmt.order_by(User.created_at.asc())
|
stmt = stmt.order_by(User.created_at.asc())
|
||||||
else:
|
else:
|
||||||
stmt = stmt.order_by(User.created_at.desc())
|
stmt = stmt.order_by(User.created_at.desc())
|
||||||
|
|
||||||
elif order_by == 'last_active_at':
|
elif order_by == 'last_active_at':
|
||||||
if direction == 'asc':
|
if direction == 'asc':
|
||||||
stmt = stmt.order_by(User.last_active_at.asc())
|
stmt = stmt.order_by(User.last_active_at.asc())
|
||||||
else:
|
else:
|
||||||
stmt = stmt.order_by(User.last_active_at.desc())
|
stmt = stmt.order_by(User.last_active_at.desc())
|
||||||
|
|
||||||
elif order_by == 'updated_at':
|
elif order_by == 'updated_at':
|
||||||
if direction == 'asc':
|
if direction == 'asc':
|
||||||
stmt = stmt.order_by(User.updated_at.asc())
|
stmt = stmt.order_by(User.updated_at.asc())
|
||||||
else:
|
else:
|
||||||
stmt = stmt.order_by(User.updated_at.desc())
|
stmt = stmt.order_by(User.updated_at.desc())
|
||||||
elif order_by == 'role':
|
elif order_by == 'role':
|
||||||
if direction == 'asc':
|
if direction == 'asc':
|
||||||
stmt = stmt.order_by(User.role.asc())
|
stmt = stmt.order_by(User.role.asc())
|
||||||
else:
|
else:
|
||||||
stmt = stmt.order_by(User.role.desc())
|
stmt = stmt.order_by(User.role.desc())
|
||||||
|
elif not filter:
|
||||||
else:
|
|
||||||
stmt = stmt.order_by(User.created_at.desc())
|
stmt = stmt.order_by(User.created_at.desc())
|
||||||
|
|
||||||
# Count BEFORE pagination
|
# Count BEFORE pagination
|
||||||
@@ -632,7 +686,7 @@ class UsersTable:
|
|||||||
self,
|
self,
|
||||||
id: str,
|
id: str,
|
||||||
provider: str,
|
provider: str,
|
||||||
external_id: str,
|
external_id: str | None,
|
||||||
db: AsyncSession | None = None,
|
db: AsyncSession | None = None,
|
||||||
) -> UserModel | None:
|
) -> UserModel | None:
|
||||||
"""Update or insert a SCIM provider/external_id pair into the user's scim JSON field."""
|
"""Update or insert a SCIM provider/external_id pair into the user's scim JSON field."""
|
||||||
|
|||||||
@@ -530,10 +530,6 @@ async def get_channel_members_by_id(
|
|||||||
|
|
||||||
if query:
|
if query:
|
||||||
filter['query'] = query
|
filter['query'] = query
|
||||||
if order_by:
|
|
||||||
filter['order_by'] = order_by
|
|
||||||
if direction:
|
|
||||||
filter['direction'] = direction
|
|
||||||
|
|
||||||
if channel.type == 'group':
|
if channel.type == 'group':
|
||||||
filter['channel_id'] = channel.id
|
filter['channel_id'] = channel.id
|
||||||
@@ -544,7 +540,13 @@ async def get_channel_members_by_id(
|
|||||||
filter['user_ids'] = permitted_ids.get('user_ids')
|
filter['user_ids'] = permitted_ids.get('user_ids')
|
||||||
filter['group_ids'] = permitted_ids.get('group_ids')
|
filter['group_ids'] = permitted_ids.get('group_ids')
|
||||||
|
|
||||||
result = await Users.get_users(filter=filter, skip=skip, limit=limit, db=db)
|
result = await Users.get_users(
|
||||||
|
filter=filter,
|
||||||
|
sort={'order_by': order_by, 'direction': direction},
|
||||||
|
skip=skip,
|
||||||
|
limit=limit,
|
||||||
|
db=db,
|
||||||
|
)
|
||||||
|
|
||||||
fetched_users = result['users']
|
fetched_users = result['users']
|
||||||
total = result['total']
|
total = result['total']
|
||||||
|
|||||||
@@ -517,20 +517,30 @@ async def get_users(
|
|||||||
# Simple filter parsing - supports userName eq, externalId eq
|
# Simple filter parsing - supports userName eq, externalId eq
|
||||||
if 'userName eq' in filter:
|
if 'userName eq' in filter:
|
||||||
email = filter.split('"')[1]
|
email = filter.split('"')[1]
|
||||||
user = await Users.get_user_by_email(email, db=db)
|
response = await Users.get_scim_users(filter={'email': email}, limit=1, db=db)
|
||||||
users_list = [user] if user else []
|
users_list = response['users']
|
||||||
total = 1 if user else 0
|
total = response['total']
|
||||||
elif 'externalId eq' in filter:
|
elif 'externalId eq' in filter:
|
||||||
external_id = filter.split('"')[1]
|
external_id = filter.split('"')[1]
|
||||||
user = await find_user_by_external_id(external_id, db=db)
|
user = await find_user_by_external_id(external_id, db=db)
|
||||||
users_list = [user] if user else []
|
users_list = [user] if user else []
|
||||||
total = 1 if user else 0
|
total = 1 if user else 0
|
||||||
else:
|
else:
|
||||||
response = await Users.get_users(skip=skip, limit=limit, db=db)
|
response = await Users.get_scim_users(
|
||||||
|
sort={'order_by': 'created_at'},
|
||||||
|
skip=skip,
|
||||||
|
limit=limit,
|
||||||
|
db=db,
|
||||||
|
)
|
||||||
users_list = response['users']
|
users_list = response['users']
|
||||||
total = response['total']
|
total = response['total']
|
||||||
else:
|
else:
|
||||||
response = await Users.get_users(skip=skip, limit=limit, db=db)
|
response = await Users.get_scim_users(
|
||||||
|
sort={'order_by': 'created_at'},
|
||||||
|
skip=skip,
|
||||||
|
limit=limit,
|
||||||
|
db=db,
|
||||||
|
)
|
||||||
users_list = response['users']
|
users_list = response['users']
|
||||||
total = response['total']
|
total = response['total']
|
||||||
|
|
||||||
@@ -553,7 +563,7 @@ async def get_user(
|
|||||||
db: AsyncSession = Depends(get_async_session),
|
db: AsyncSession = Depends(get_async_session),
|
||||||
):
|
):
|
||||||
"""Get SCIM User by ID"""
|
"""Get SCIM User by ID"""
|
||||||
user = await Users.get_user_by_id(user_id, db=db)
|
user = await Users.get_scim_user_by_id(user_id, db=db)
|
||||||
if not user:
|
if not user:
|
||||||
return scim_error(status_code=status.HTTP_404_NOT_FOUND, detail=f'User {user_id} not found')
|
return scim_error(status_code=status.HTTP_404_NOT_FOUND, detail=f'User {user_id} not found')
|
||||||
|
|
||||||
@@ -624,11 +634,12 @@ async def create_user(
|
|||||||
detail='Failed to create user',
|
detail='Failed to create user',
|
||||||
)
|
)
|
||||||
|
|
||||||
# Store externalId in the scim field
|
new_user = await Users.update_user_scim_by_id(user_id, get_scim_provider(), user_data.externalId, db=db)
|
||||||
if user_data.externalId:
|
if not new_user:
|
||||||
provider = get_scim_provider()
|
raise HTTPException(
|
||||||
await Users.update_user_scim_by_id(user_id, provider, user_data.externalId, db=db)
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||||
new_user = await Users.get_user_by_id(user_id, db=db)
|
detail='Failed to stamp SCIM user',
|
||||||
|
)
|
||||||
|
|
||||||
await publish_event(
|
await publish_event(
|
||||||
request,
|
request,
|
||||||
@@ -654,7 +665,7 @@ async def update_user(
|
|||||||
db: AsyncSession = Depends(get_async_session),
|
db: AsyncSession = Depends(get_async_session),
|
||||||
):
|
):
|
||||||
"""Update SCIM User (full update)"""
|
"""Update SCIM User (full update)"""
|
||||||
user = await Users.get_user_by_id(user_id, db=db)
|
user = await Users.get_scim_user_by_id(user_id, db=db)
|
||||||
if not user:
|
if not user:
|
||||||
raise HTTPException(
|
raise HTTPException(
|
||||||
status_code=status.HTTP_404_NOT_FOUND,
|
status_code=status.HTTP_404_NOT_FOUND,
|
||||||
@@ -734,7 +745,7 @@ async def patch_user(
|
|||||||
db: AsyncSession = Depends(get_async_session),
|
db: AsyncSession = Depends(get_async_session),
|
||||||
):
|
):
|
||||||
"""Update SCIM User (partial update)"""
|
"""Update SCIM User (partial update)"""
|
||||||
user = await Users.get_user_by_id(user_id, db=db)
|
user = await Users.get_scim_user_by_id(user_id, db=db)
|
||||||
if not user:
|
if not user:
|
||||||
raise HTTPException(
|
raise HTTPException(
|
||||||
status_code=status.HTTP_404_NOT_FOUND,
|
status_code=status.HTTP_404_NOT_FOUND,
|
||||||
@@ -808,7 +819,7 @@ async def delete_user(
|
|||||||
db: AsyncSession = Depends(get_async_session),
|
db: AsyncSession = Depends(get_async_session),
|
||||||
):
|
):
|
||||||
"""Delete SCIM User"""
|
"""Delete SCIM User"""
|
||||||
user = await Users.get_user_by_id(user_id, db=db)
|
user = await Users.get_scim_user_by_id(user_id, db=db)
|
||||||
if not user:
|
if not user:
|
||||||
raise HTTPException(
|
raise HTTPException(
|
||||||
status_code=status.HTTP_404_NOT_FOUND,
|
status_code=status.HTTP_404_NOT_FOUND,
|
||||||
|
|||||||
@@ -112,14 +112,14 @@ async def get_users(
|
|||||||
filter = {}
|
filter = {}
|
||||||
if query:
|
if query:
|
||||||
filter['query'] = query
|
filter['query'] = query
|
||||||
if order_by:
|
|
||||||
filter['order_by'] = order_by
|
|
||||||
if direction:
|
|
||||||
filter['direction'] = direction
|
|
||||||
|
|
||||||
filter['direction'] = direction
|
result = await Users.get_users(
|
||||||
|
filter=filter,
|
||||||
result = await Users.get_users(filter=filter, skip=skip, limit=limit, db=db)
|
sort={'order_by': order_by, 'direction': direction},
|
||||||
|
skip=skip,
|
||||||
|
limit=limit,
|
||||||
|
db=db,
|
||||||
|
)
|
||||||
|
|
||||||
users = result['users']
|
users = result['users']
|
||||||
total = result['total']
|
total = result['total']
|
||||||
@@ -167,12 +167,14 @@ async def search_users(
|
|||||||
filter = {}
|
filter = {}
|
||||||
if query:
|
if query:
|
||||||
filter['query'] = query
|
filter['query'] = query
|
||||||
if order_by:
|
|
||||||
filter['order_by'] = order_by
|
|
||||||
if direction:
|
|
||||||
filter['direction'] = direction
|
|
||||||
|
|
||||||
return await Users.get_users(filter=filter, skip=skip, limit=limit, db=db)
|
return await Users.get_users(
|
||||||
|
filter=filter,
|
||||||
|
sort={'order_by': order_by, 'direction': direction},
|
||||||
|
skip=skip,
|
||||||
|
limit=limit,
|
||||||
|
db=db,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
############################
|
############################
|
||||||
|
|||||||
Reference in New Issue
Block a user