ilia/punimtag
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Add is_active and role fields to AuthUser schema and update user management logic

Browse Source 
This commit introduces new fields `is_active` and `role` to the `AuthUserResponse` and `AuthUserUpdateRequest` schemas, enhancing user management capabilities. The `deleteUser` and `updateUser` functions are updated to handle user deactivation instead of deletion when linked data exists. Additionally, the ManageUsers component is enhanced with filtering options for active status and roles, improving user experience. Documentation has been updated to reflect these changes.
This commit is contained in:
tanyar09
2025-12-05 14:20:45 -05:00
parent 0e65eac206
commit e9e8fbf3f5
8 changed files with 737 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/web/api/auth_users.py
+413 -88
View File
@@ -2,9 +2,11 @@
from __future__ import annotations
import logging
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
from fastapi.responses import JSONResponse
from sqlalchemy import text
from sqlalchemy.orm import Session
@@ -20,6 +22,28 @@ from src.web.schemas.auth_users import (
from src.web.utils.password import hash_password
router = APIRouter(prefix="/auth-users", tags=["auth-users"])
logger = logging.getLogger(__name__)
def _check_column_exists(auth_db: Session, table_name: str, column_name: str) -> bool:
"""Check if a column exists in a table."""
try:
from sqlalchemy import inspect as sqlalchemy_inspect
inspector = sqlalchemy_inspect(auth_db.bind)
columns = {col["name"] for col in inspector.get_columns(table_name)}
return column_name in columns
except Exception:
return False
def _get_role_from_is_admin(auth_db: Session, is_admin: bool) -> str:
"""Get role value from is_admin boolean. Returns 'Admin' if is_admin is True, 'User' otherwise."""
return "Admin" if is_admin else "User"
def _get_is_admin_from_role(role: str | None) -> bool:
"""Get is_admin boolean from role string. Returns True if role is 'Admin', False otherwise."""
return role == "Admin" if role else False
@router.get("", response_model=AuthUsersListResponse)
@@ -29,53 +53,127 @@ def list_auth_users(
) -> AuthUsersListResponse:
"""List all users from auth database - admin only."""
try:
# Check if optional columns exist
has_role_column = _check_column_exists(auth_db, "users", "role")
has_is_active_column = _check_column_exists(auth_db, "users", "is_active")
# Query users from auth database with all columns from schema
result = auth_db.execute(text("""
SELECT
id,
email,
name,
is_admin,
has_write_access,
created_at,
updated_at
FROM users
ORDER BY COALESCE(name, email) ASC
"""))
# Try to include is_active and role if columns exist
result = None
try:
# Build SELECT query based on which columns exist
select_fields = "id, email, name, is_admin, has_write_access"
if has_is_active_column:
select_fields += ", is_active"
if has_role_column:
select_fields += ", role"
select_fields += ", created_at, updated_at"
result = auth_db.execute(text(f"""
SELECT {select_fields}
FROM users
ORDER BY COALESCE(name, email) ASC
"""))
except Exception:
# Rollback the failed transaction before trying again
auth_db.rollback()
try:
# Try with is_active only (no role)
select_fields = "id, email, name, is_admin, has_write_access"
if has_is_active_column:
select_fields += ", is_active"
select_fields += ", created_at, updated_at"
result = auth_db.execute(text(f"""
SELECT {select_fields}
FROM users
ORDER BY COALESCE(name, email) ASC
"""))
except Exception:
# Rollback again before final attempt
auth_db.rollback()
# Base columns only
result = auth_db.execute(text("""
SELECT
id,
email,
name,
is_admin,
has_write_access,
created_at,
updated_at
FROM users
ORDER BY COALESCE(name, email) ASC
"""))
if result is None:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to query auth users",
)
rows = result.fetchall()
users = []
for row in rows:
# Access row attributes directly - SQLAlchemy Row objects support attribute access
user_id = int(row.id)
email = str(row.email)
name = row.name if row.name is not None else None
# Get boolean fields - convert to proper boolean
# These columns have defaults so they should always have values
is_admin = bool(row.is_admin)
has_write_access = bool(row.has_write_access)
created_at = row.created_at
updated_at = row.updated_at
users.append(AuthUserResponse(
id=user_id,
name=name,
email=email,
is_admin=is_admin,
has_write_access=has_write_access,
created_at=created_at,
updated_at=updated_at,
))
try:
# Access row attributes directly - SQLAlchemy Row objects support attribute access
user_id = int(row.id)
email = str(row.email)
name = row.name if row.name is not None else None
# Get boolean fields - convert to proper boolean
# These columns have defaults so they should always have values
is_admin = bool(row.is_admin)
has_write_access = bool(row.has_write_access)
# Optional columns - try to get from row, default to None if not selected
is_active = None
try:
is_active = row.is_active
if is_active is not None:
is_active = bool(is_active)
else:
# NULL values should be treated as True (active)
is_active = True
except (AttributeError, KeyError):
# Column not selected or doesn't exist - default to True (active)
is_active = True
# Get role - if column doesn't exist, derive from is_admin
if has_role_column:
role = getattr(row, 'role', None)
else:
role = _get_role_from_is_admin(auth_db, is_admin)
created_at = getattr(row, 'created_at', None)
updated_at = getattr(row, 'updated_at', None)
users.append(AuthUserResponse(
id=user_id,
name=name,
email=email,
is_admin=is_admin,
has_write_access=has_write_access,
is_active=is_active,
role=role,
created_at=created_at,
updated_at=updated_at,
))
except Exception as row_error:
logger.warning(f"Error processing auth user row: {row_error}")
# Skip this row and continue
continue
return AuthUsersListResponse(items=users, total=len(users))
except HTTPException:
raise
except Exception as e:
auth_db.rollback()
import traceback
error_detail = f"Failed to list auth users: {str(e)}\n{traceback.format_exc()}"
logger.error(f"Error listing auth users: {error_detail}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=error_detail,
detail=f"Failed to list auth users: {str(e)}",
)
@@ -179,11 +277,48 @@ def get_auth_user(
) -> AuthUserResponse:
"""Get a specific auth user by ID - admin only."""
try:
result = auth_db.execute(text("""
SELECT id, email, name, is_admin, has_write_access, created_at, updated_at
FROM users
WHERE id = :user_id
"""), {"user_id": user_id})
# Check if optional columns exist
has_role_column = _check_column_exists(auth_db, "users", "role")
has_is_active_column = _check_column_exists(auth_db, "users", "is_active")
# Try to include is_active and role if columns exist
try:
# Build SELECT query based on which columns exist
select_fields = "id, email, name, is_admin, has_write_access"
if has_is_active_column:
select_fields += ", is_active"
if has_role_column:
select_fields += ", role"
select_fields += ", created_at, updated_at"
result = auth_db.execute(text(f"""
SELECT {select_fields}
FROM users
WHERE id = :user_id
"""), {"user_id": user_id})
except Exception:
# Rollback the failed transaction before trying again
auth_db.rollback()
try:
# Try with is_active only (no role)
select_fields = "id, email, name, is_admin, has_write_access"
if has_is_active_column:
select_fields += ", is_active"
select_fields += ", created_at, updated_at"
result = auth_db.execute(text(f"""
SELECT {select_fields}
FROM users
WHERE id = :user_id
"""), {"user_id": user_id})
except Exception:
# Rollback again before final attempt
auth_db.rollback()
# Columns don't exist, select without them
result = auth_db.execute(text("""
SELECT id, email, name, is_admin, has_write_access, created_at, updated_at
FROM users
WHERE id = :user_id
"""), {"user_id": user_id})
row = result.first()
if not row:
@@ -194,6 +329,18 @@ def get_auth_user(
is_admin = bool(row.is_admin)
has_write_access = bool(row.has_write_access)
is_active = getattr(row, 'is_active', None)
if is_active is not None:
is_active = bool(is_active)
else:
# NULL values should be treated as True (active)
is_active = True
# Get role - if column doesn't exist, derive from is_admin
if has_role_column:
role = getattr(row, 'role', None)
else:
role = _get_role_from_is_admin(auth_db, is_admin)
return AuthUserResponse(
id=row.id,
@@ -201,12 +348,15 @@ def get_auth_user(
email=row.email,
is_admin=is_admin,
has_write_access=has_write_access,
is_active=is_active,
role=role,
created_at=getattr(row, 'created_at', None),
updated_at=getattr(row, 'updated_at', None),
)
except HTTPException:
raise
except Exception as e:
auth_db.rollback()
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Failed to get auth user: {str(e)}",
@@ -248,52 +398,94 @@ def update_auth_user(
detail=f"User with email '{request.email}' already exists",
)
# Update all fields (all are required)
# Check if role column exists
has_role_column = _check_column_exists(auth_db, "users", "role")
# Update all fields (all are required, is_active and role are optional)
dialect = auth_db.bind.dialect.name if auth_db.bind else 'postgresql'
supports_returning = dialect == 'postgresql'
# Determine is_admin value - if role is provided and role column doesn't exist, use role to set is_admin
is_admin_value = request.is_admin
if request.role is not None and not has_role_column:
# Role column doesn't exist, derive is_admin from role
is_admin_value = _get_is_admin_from_role(request.role)
# Build UPDATE query - include is_active and role if provided and column exists
update_fields = ["email = :email", "name = :name", "is_admin = :is_admin", "has_write_access = :has_write_access"]
update_params = {
"user_id": user_id,
"email": request.email,
"name": request.name,
"is_admin": is_admin_value,
"has_write_access": request.has_write_access,
}
if request.is_active is not None:
update_fields.append("is_active = :is_active")
update_params["is_active"] = request.is_active
if request.role is not None and has_role_column:
# Only update role if the column exists
update_fields.append("role = :role")
update_params["role"] = request.role
update_sql = f"""
UPDATE users
SET {', '.join(update_fields)}
WHERE id = :user_id
"""
if supports_returning:
result = auth_db.execute(text("""
UPDATE users
SET email = :email,
name = :name,
is_admin = :is_admin,
has_write_access = :has_write_access
WHERE id = :user_id
RETURNING id, email, name, is_admin, has_write_access, created_at, updated_at
"""), {
"user_id": user_id,
"email": request.email,
"name": request.name,
"is_admin": request.is_admin,
"has_write_access": request.has_write_access,
})
# Build select fields - try to include optional columns
select_fields = "id, email, name, is_admin, has_write_access"
if request.is_active is not None or _check_column_exists(auth_db, "users", "is_active"):
select_fields += ", is_active"
if has_role_column:
select_fields += ", role"
select_fields += ", created_at, updated_at"
result = auth_db.execute(text(f"""
{update_sql}
RETURNING {select_fields}
"""), update_params)
auth_db.commit()
row = result.first()
else:
# SQLite - update then select
auth_db.execute(text("""
UPDATE users
SET email = :email,
name = :name,
is_admin = :is_admin,
has_write_access = :has_write_access
WHERE id = :user_id
"""), {
"user_id": user_id,
"email": request.email,
"name": request.name,
"is_admin": request.is_admin,
"has_write_access": request.has_write_access,
})
auth_db.execute(text(update_sql), update_params)
auth_db.commit()
# Get the updated row
result = auth_db.execute(text("""
SELECT id, email, name, is_admin, has_write_access, created_at, updated_at
FROM users
WHERE id = :user_id
"""), {"user_id": user_id})
row = result.first()
# Get the updated row - try to include optional columns
try:
if has_role_column:
result = auth_db.execute(text("""
SELECT id, email, name, is_admin, has_write_access, is_active, role, created_at, updated_at
FROM users
WHERE id = :user_id
"""), {"user_id": user_id})
else:
result = auth_db.execute(text("""
SELECT id, email, name, is_admin, has_write_access, is_active, created_at, updated_at
FROM users
WHERE id = :user_id
"""), {"user_id": user_id})
row = result.first()
except Exception:
auth_db.rollback()
try:
result = auth_db.execute(text("""
SELECT id, email, name, is_admin, has_write_access, is_active, created_at, updated_at
FROM users
WHERE id = :user_id
"""), {"user_id": user_id})
row = result.first()
except Exception:
auth_db.rollback()
result = auth_db.execute(text("""
SELECT id, email, name, is_admin, has_write_access, created_at, updated_at
FROM users
WHERE id = :user_id
"""), {"user_id": user_id})
row = result.first()
if not row:
raise HTTPException(
@@ -303,6 +495,18 @@ def update_auth_user(
is_admin = bool(row.is_admin)
has_write_access = bool(row.has_write_access)
is_active = getattr(row, 'is_active', None)
if is_active is not None:
is_active = bool(is_active)
else:
# NULL values should be treated as True (active)
is_active = True
# Get role - if column doesn't exist, derive from is_admin
if has_role_column:
role = getattr(row, 'role', None)
else:
role = _get_role_from_is_admin(auth_db, is_admin)
return AuthUserResponse(
id=row.id,
@@ -310,6 +514,8 @@ def update_auth_user(
email=row.email,
is_admin=is_admin,
has_write_access=has_write_access,
is_active=is_active,
role=role,
created_at=getattr(row, 'created_at', None),
updated_at=getattr(row, 'updated_at', None),
)
@@ -329,27 +535,146 @@ def delete_auth_user(
user_id: int,
auth_db: Session = Depends(get_auth_db),
) -> Response:
"""Delete an auth user - admin only."""
"""Delete an auth user - admin only.
If the user has linked data (pending_photos, pending_identifications,
inappropriate_photo_reports), the user will be set to inactive instead
of deleted. Admins will be notified via logging.
"""
try:
# Check if user exists
check_result = auth_db.execute(text("""
SELECT id FROM users WHERE id = :user_id
# Check if user exists and get user info
user_result = auth_db.execute(text("""
SELECT id, email, name FROM users WHERE id = :user_id
"""), {"user_id": user_id})
if not check_result.first():
user_row = user_result.first()
if not user_row:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Auth user with ID {user_id} not found",
)
# Delete user
auth_db.execute(text("""
DELETE FROM users WHERE id = :user_id
"""), {"user_id": user_id})
user_email = user_row.email
user_name = user_row.name or user_email
auth_db.commit()
# Check for linked data in auth database
pending_photos_count = auth_db.execute(text("""
SELECT COUNT(*) FROM pending_photos WHERE user_id = :user_id
"""), {"user_id": user_id}).scalar() or 0
return Response(status_code=status.HTTP_204_NO_CONTENT)
pending_identifications_count = auth_db.execute(text("""
SELECT COUNT(*) FROM pending_identifications WHERE user_id = :user_id
"""), {"user_id": user_id}).scalar() or 0
inappropriate_reports_count = auth_db.execute(text("""
SELECT COUNT(*) FROM inappropriate_photo_reports WHERE user_id = :user_id
"""), {"user_id": user_id}).scalar() or 0
has_linked_data = (
pending_photos_count > 0 or
pending_identifications_count > 0 or
inappropriate_reports_count > 0
)
if has_linked_data:
# Check if is_active column exists by trying to query it
dialect = auth_db.bind.dialect.name if auth_db.bind else "postgresql"
has_is_active_column = False
try:
# Try to select is_active column to check if it exists
test_result = auth_db.execute(text("""
SELECT is_active FROM users WHERE id = :user_id LIMIT 1
"""), {"user_id": user_id})
test_result.first()
has_is_active_column = True
except Exception:
# Column doesn't exist - this should have been added at startup
# but if it wasn't, we can't proceed
error_msg = "is_active column does not exist in auth database users table"
logger.error(
f"Cannot deactivate auth user '{user_name}' (ID: {user_id}): {error_msg}"
)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=(
f"Cannot delete user '{user_name}' because they have linked data "
f"({pending_photos_count} pending photo(s), "
f"{pending_identifications_count} pending identification(s), "
f"{inappropriate_reports_count} inappropriate photo report(s)) "
f"and the is_active column does not exist in the auth database users table. "
f"Please restart the server to add the column automatically, or contact "
f"your database administrator to add it manually: "
f"ALTER TABLE users ADD COLUMN is_active BOOLEAN DEFAULT TRUE"
),
)
# Set user inactive instead of deleting
if dialect == "postgresql":
auth_db.execute(text("""
UPDATE users SET is_active = FALSE WHERE id = :user_id
"""), {"user_id": user_id})
else:
# SQLite uses 0 for FALSE
auth_db.execute(text("""
UPDATE users SET is_active = 0 WHERE id = :user_id
"""), {"user_id": user_id})
auth_db.commit()
# Notify admins via logging
logger.warning(
f"Auth user '{user_name}' (ID: {user_id}, email: {user_email}) was set to inactive "
f"instead of deleted because they have linked data: {pending_photos_count} pending "
f"photo(s), {pending_identifications_count} pending identification(s), "
f"{inappropriate_reports_count} inappropriate photo report(s). "
f"Action performed by admin: {current_admin['username']}",
extra={
"user_id": user_id,
"user_email": user_email,
"user_name": user_name,
"pending_photos_count": pending_photos_count,
"pending_identifications_count": pending_identifications_count,
"inappropriate_reports_count": inappropriate_reports_count,
"admin_username": current_admin["username"],
}
)
# Return success but indicate user was deactivated
return JSONResponse(
status_code=status.HTTP_200_OK,
content={
"message": (
f"User '{user_name}' has been set to inactive because they have "
f"linked data ({pending_photos_count} pending photo(s), "
f"{pending_identifications_count} pending identification(s), "
f"{inappropriate_reports_count} inappropriate photo report(s))."
),
"deactivated": True,
"pending_photos_count": pending_photos_count,
"pending_identifications_count": pending_identifications_count,
"inappropriate_reports_count": inappropriate_reports_count,
}
)
else:
# No linked data - safe to delete
auth_db.execute(text("""
DELETE FROM users WHERE id = :user_id
"""), {"user_id": user_id})
auth_db.commit()
logger.info(
f"Auth user '{user_name}' (ID: {user_id}, email: {user_email}) was deleted. "
f"Action performed by admin: {current_admin['username']}",
extra={
"user_id": user_id,
"user_email": user_email,
"user_name": user_name,
"admin_username": current_admin["username"],
}
)
return Response(status_code=status.HTTP_204_NO_CONTENT)
except HTTPException:
raise
except Exception as e:
src/web/api/users.py
+72 -16
View File
@@ -6,6 +6,7 @@ import logging
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
from fastapi.responses import JSONResponse
from sqlalchemy import text
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
@@ -19,7 +20,7 @@ from src.web.constants.roles import (
is_admin_role,
)
from src.web.db.session import get_auth_db, get_db
from src.web.db.models import User
from src.web.db.models import Face, PhotoFavorite, PhotoPersonLinkage, User
from src.web.schemas.users import (
UserCreateRequest,
UserResponse,
@@ -442,6 +443,10 @@ def delete_user(
) -> Response:
"""Delete a user - admin only.
If the user has linked data (faces identified, video person linkages),
the user will be set to inactive instead of deleted, and favorites will
be removed. Admins will be notified via logging.
Prevents admin from deleting themselves.
"""
user = db.query(User).filter(User.id == user_id).first()
@@ -458,21 +463,72 @@ def delete_user(
detail="Cannot delete your own account",
)
try:
# Check for linked data (faces or photo_person_linkages identified by this user)
faces_count = db.query(Face).filter(Face.identified_by_user_id == user_id).count()
linkages_count = db.query(PhotoPersonLinkage).filter(
PhotoPersonLinkage.identified_by_user_id == user_id
).count()
has_linked_data = faces_count > 0 or linkages_count > 0
# Always delete favorites (they use username, not user_id)
favorites_deleted = db.query(PhotoFavorite).filter(
PhotoFavorite.username == user.username
).delete()
if has_linked_data:
# Set user inactive instead of deleting
user.is_active = False
db.add(user)
db.commit()
# Notify admins via logging
logger.warning(
f"User '{user.username}' (ID: {user_id}) was set to inactive instead of deleted "
f"because they have linked data: {faces_count} face(s) and {linkages_count} "
f"video person linkage(s). {favorites_deleted} favorite(s) were deleted. "
f"Action performed by admin: {current_admin['username']}",
extra={
"user_id": user_id,
"username": user.username,
"faces_count": faces_count,
"linkages_count": linkages_count,
"favorites_deleted": favorites_deleted,
"admin_username": current_admin["username"],
}
)
# Return success but indicate user was deactivated
return JSONResponse(
status_code=status.HTTP_200_OK,
content={
"message": (
f"User '{user.username}' has been set to inactive because they have "
f"linked data ({faces_count} face(s), {linkages_count} linkage(s)). "
f"{favorites_deleted} favorite(s) were deleted."
),
"deactivated": True,
"faces_count": faces_count,
"linkages_count": linkages_count,
"favorites_deleted": favorites_deleted,
}
)
else:
# No linked data - safe to delete
db.delete(user)
db.commit()
except IntegrityError as exc:
db.rollback()
constraint_name = "faces_identified_by_user_id_fkey"
if exc.orig and constraint_name in str(exc.orig):
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=(
"This user has identified faces and cannot be deleted. "
"Set the user inactive instead."
),
) from exc
raise
return Response(status_code=status.HTTP_204_NO_CONTENT)
logger.info(
f"User '{user.username}' (ID: {user_id}) was deleted. "
f"{favorites_deleted} favorite(s) were deleted. "
f"Action performed by admin: {current_admin['username']}",
extra={
"user_id": user_id,
"username": user.username,
"favorites_deleted": favorites_deleted,
"admin_username": current_admin["username"],
}
)
return Response(status_code=status.HTTP_204_NO_CONTENT)
src/web/app.py
+61 -1
View File
@@ -30,7 +30,7 @@ from src.web.api.version import router as version_router
from src.web.settings import APP_TITLE, APP_VERSION
from src.web.constants.roles import DEFAULT_ADMIN_ROLE, DEFAULT_USER_ROLE, ROLE_VALUES
from src.web.db.base import Base, engine
from src.web.db.session import database_url
from src.web.db.session import auth_engine, database_url
# Import models to ensure they're registered with Base.metadata
from src.web.db import models # noqa: F401
from src.web.db.models import RolePermission
@@ -475,6 +475,63 @@ def ensure_photo_person_linkage_table(inspector) -> None:
print("✅ Created photo_person_linkage table")
def ensure_auth_user_is_active_column() -> None:
"""Ensure auth database users table contains is_active column."""
if auth_engine is None:
# Auth database not configured
return
try:
from sqlalchemy import inspect as sqlalchemy_inspect
auth_inspector = sqlalchemy_inspect(auth_engine)
if "users" not in auth_inspector.get_table_names():
print("ℹ️ Auth database users table does not exist yet")
return
columns = {column["name"] for column in auth_inspector.get_columns("users")}
if "is_active" in columns:
print("ℹ️ is_active column already exists in auth database users table")
return
# Column doesn't exist - try to add it
print("🔄 Adding is_active column to auth database users table...")
dialect = auth_engine.dialect.name
try:
with auth_engine.connect() as connection:
with connection.begin():
if dialect == "postgresql":
connection.execute(
text("ALTER TABLE users ADD COLUMN IF NOT EXISTS is_active BOOLEAN DEFAULT TRUE")
)
else:
# SQLite
connection.execute(
text("ALTER TABLE users ADD COLUMN is_active BOOLEAN DEFAULT 1")
)
print("✅ Added is_active column to auth database users table")
except Exception as alter_exc:
# Check if it's a permission error
error_str = str(alter_exc)
if "permission" in error_str.lower() or "insufficient" in error_str.lower() or "owner" in error_str.lower():
print("⚠️ Cannot add is_active column: insufficient database privileges")
print(" The column will need to be added manually by a database administrator:")
if dialect == "postgresql":
print(" ALTER TABLE users ADD COLUMN is_active BOOLEAN DEFAULT TRUE;")
else:
print(" ALTER TABLE users ADD COLUMN is_active BOOLEAN DEFAULT 1;")
print(" Until then, users with linked data cannot be deleted.")
else:
# Some other error
print(f"⚠️ Failed to add is_active column to auth database users table: {alter_exc}")
except Exception as exc:
print(f"⚠️ Failed to check/add is_active column to auth database users table: {exc}")
# Don't raise - auth database might not be available or have permission issues
# The delete endpoint will handle this case gracefully
def ensure_role_permissions_table(inspector) -> None:
"""Ensure the role_permissions table exists for permission matrix."""
if "role_permissions" in inspector.get_table_names():
@@ -528,6 +585,9 @@ async def lifespan(app: FastAPI):
ensure_photo_person_linkage_table(inspector)
ensure_face_excluded_column(inspector)
ensure_role_permissions_table(inspector)
# Ensure auth database schema
ensure_auth_user_is_active_column()
except Exception as exc:
print(f"❌ Database initialization failed: {exc}")
raise
src/web/schemas/auth_users.py
+4
View File
@@ -18,6 +18,8 @@ class AuthUserResponse(BaseModel):
email: str
is_admin: Optional[bool] = None
has_write_access: Optional[bool] = None
is_active: Optional[bool] = None
role: Optional[str] = None
created_at: Optional[datetime] = None
updated_at: Optional[datetime] = None
@@ -43,6 +45,8 @@ class AuthUserUpdateRequest(BaseModel):
name: str = Field(..., min_length=1, max_length=200, description="Name (required)")
is_admin: bool = Field(..., description="Admin role (required)")
has_write_access: bool = Field(..., description="Write access (required)")
is_active: Optional[bool] = Field(None, description="Active status (optional)")
role: Optional[str] = Field(None, description="Role: 'Admin' or 'User' (optional)")
class AuthUsersListResponse(BaseModel):