Feat: Option to send multiple usernames in get_users, *no changes in …

…API/cli interfaces (#661)

app/db/crud.py

@@ -113,16 +113,19 @@ def get_user_by_id(db: Session, user_id: int):
 def get_users(db: Session,
               offset: Optional[int] = None,
               limit: Optional[int] = None,
-              username: Optional[str] = None,
+              usernames: Optional[List[str]] = None,
               status: Optional[Union[UserStatus, list]] = None,
               sort: Optional[List[UsersSortingOptions]] = None,
               admin: Optional[Admin] = None,
               reset_strategy: Optional[Union[UserDataLimitResetStrategy, list]] = None,
               return_with_count: bool = False) -> Union[List[User], Tuple[List[User], int]]:
     query = db.query(User)
 
-    if username:
-        query = query.filter(User.username.ilike(f'%{username}%'))
+    if usernames:
+        if len(usernames) == 1:
+            query = query.filter(User.username.ilike(f"%{usernames[0]}%"))
+        else:
+            query = query.filter(User.username.in_(usernames))
 
     if status:
         if isinstance(status, list):
@@ -157,7 +160,7 @@ def get_users(db: Session,
     return query.all()
 
 
-def get_user_usages(db: Session, dbuser: User, start: datetime, end: datetime, 
+def get_user_usages(db: Session, dbuser: User, start: datetime, end: datetime,
                     ) -> List[UserUsageResponse]:
     usages = {}
 
@@ -266,7 +269,7 @@ def update_user(db: Session, dbuser: User, modify: UserModify):
             if not dbuser.data_limit or dbuser.used_traffic < dbuser.data_limit:
                 if dbuser.status != UserStatus.on_hold:
                     dbuser.status = UserStatus.active
-                    
+
                 if not dbuser.data_limit or (calculate_usage_percent(
                         dbuser.used_traffic, dbuser.data_limit) < NOTIFY_REACHED_USAGE_PERCENT):
                     delete_notification_reminder_by_type(db, dbuser.id, ReminderType.data_usage)
@@ -286,14 +289,14 @@ def update_user(db: Session, dbuser: User, modify: UserModify):
 
     if modify.note is not None:
         dbuser.note = modify.note or None
-      
+
     if modify.data_limit_reset_strategy is not None:
         dbuser.data_limit_reset_strategy = modify.data_limit_reset_strategy.value
 
-    if modify.on_hold_timeout is not None :
+    if modify.on_hold_timeout is not None:
         dbuser.on_hold_timeout = modify.on_hold_timeout
 
-    if modify.on_hold_expire_duration is not None :
+    if modify.on_hold_expire_duration is not None:
         dbuser.on_hold_expire_duration = modify.on_hold_expire_duration
 
     dbuser.edit_at = datetime.utcnow()
@@ -352,7 +355,7 @@ def reset_all_users_data_usage(db: Session, admin: Optional[Admin] = None):
 
     for dbuser in query.all():
         dbuser.used_traffic = 0
-        if dbuser.status not in [UserStatus.on_hold , UserStatus.expired , UserStatus.disabled]:
+        if dbuser.status not in [UserStatus.on_hold, UserStatus.expired, UserStatus.disabled]:
             dbuser.status = UserStatus.active
         dbuser.usage_logs.clear()
         dbuser.node_usages.clear()
@@ -377,7 +380,7 @@ def set_owner(db: Session, dbuser: User, admin: Admin):
 
 def start_user_expire(db: Session, dbuser: User):
 
-    expire = int(datetime.utcnow().timestamp()) + dbuser.on_hold_expire_duration    
+    expire = int(datetime.utcnow().timestamp()) + dbuser.on_hold_expire_duration
     dbuser.expire = expire
     db.commit()
     db.refresh(dbuser)

app/views/user.py

@@ -1,7 +1,8 @@
 from datetime import datetime, timezone
+from typing import List
 
 import sqlalchemy
-from fastapi import BackgroundTasks, Depends, HTTPException
+from fastapi import BackgroundTasks, Depends, HTTPException, Query
 
 from app import app, logger, xray
 from app.db import Session, crud, get_db
@@ -213,7 +214,7 @@ def revoke_user_subscription(username: str,
 @app.get("/api/users", tags=['User'], response_model=UsersResponse)
 def get_users(offset: int = None,
               limit: int = None,
-              username: str = None,
+              username: List[str] = Query(None),
               status: UserStatus = None,
               sort: str = None,
               db: Session = Depends(get_db),
@@ -236,7 +237,7 @@ def get_users(offset: int = None,
     users, count = crud.get_users(db=db,
                                   offset=offset,
                                   limit=limit,
-                                  username=username,
+                                  usernames=username,
                                   status=status,
                                   sort=sort,
                                   admin=dbadmin if not admin.is_sudo else None,

cli/user.py

@@ -1,12 +1,12 @@
-from typing import Optional
-from rich.table import Table
+from typing import List, Optional
 
 import typer
+from rich.table import Table
 
-from app.db import GetDB
-from app.db import crud
+from app.db import GetDB, crud
 from app.db.models import User
 from app.utils.system import readable_size
+
 from . import utils
 
 app = typer.Typer(no_args_is_help=True)
@@ -16,7 +16,7 @@
 def list_users(
     offset: Optional[int] = typer.Option(None, *utils.FLAGS["offset"]),
     limit: Optional[int] = typer.Option(None, *utils.FLAGS["limit"]),
-    username: Optional[str] = typer.Option(None, *utils.FLAGS["username"], help="Search by username"),
+    username: Optional[List[str]] = typer.Option(None, *utils.FLAGS["username"], help="Search by username(s)"),
     status: Optional[crud.UserStatus] = typer.Option(None, *utils.FLAGS["status"]),
     admin: Optional[str] = typer.Option(None, "--admin", "--owner", help="Search by owner admin's username")
 ):
@@ -28,7 +28,7 @@ def list_users(
     with GetDB() as db:
         users: list[User] = crud.get_users(
             db=db, offset=offset, limit=limit,
-            username=username, status=status,
+            usernames=username, status=status,
             admin=crud.get_admin(db, admin) if admin else None
         )
 
@@ -68,7 +68,7 @@ def set_owner(
         user: User = utils.raise_if_falsy(
             crud.get_user(db, username=username), f'User "{username}" not found.')
 
-        dbadmin: Admin = utils.raise_if_falsy(
+        dbadmin = utils.raise_if_falsy(
             crud.get_admin(db, username=admin), f'Admin "{admin}" not found.')
 
         # Ask for confirmation if user already has an owner