feat: wire validation models into admin routes and deduplicate error handling
Replace manual validation error formatting with shared helper in both admin and manage profile routes. Add UsernameInput validation to invite route and GroupListInput validation to groups route. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
56c177c817
commit
72a93984f2
4 changed files with 205 additions and 49 deletions
|
|
@ -6,7 +6,7 @@ from pydantic import ValidationError
|
|||
|
||||
from porchlight.dependencies import get_session_user
|
||||
from porchlight.models import User
|
||||
from porchlight.validation import ProfileUpdate
|
||||
from porchlight.validation import GroupListInput, ProfileUpdate, UsernameInput, format_validation_errors
|
||||
|
||||
router = APIRouter(prefix="/admin", tags=["admin"])
|
||||
|
||||
|
|
@ -109,17 +109,21 @@ async def create_invite(
|
|||
if admin is None:
|
||||
return HTMLResponse("Forbidden", status_code=403)
|
||||
|
||||
username = username.strip()
|
||||
if not username:
|
||||
return HTMLResponse('<div role="alert">Username is required</div>')
|
||||
try:
|
||||
validated = UsernameInput(username=username)
|
||||
except ValidationError as exc:
|
||||
return HTMLResponse(format_validation_errors(exc))
|
||||
|
||||
magic_link_service = request.app.state.magic_link_service
|
||||
settings = request.app.state.settings
|
||||
link = await magic_link_service.create(username=username, created_by=admin.username, note="admin invite")
|
||||
link = await magic_link_service.create(
|
||||
username=validated.username, created_by=admin.username, note="admin invite"
|
||||
)
|
||||
url = f"{settings.issuer}/register/{link.token}"
|
||||
|
||||
return HTMLResponse(
|
||||
f'<div role="status">Invite created for <strong>{username}</strong>:</div><div class="invite-url">{url}</div>'
|
||||
f'<div role="status">Invite created for <strong>{validated.username}</strong>:</div>'
|
||||
f'<div class="invite-url">{url}</div>'
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -155,21 +159,7 @@ async def update_user_profile(
|
|||
locale=locale,
|
||||
)
|
||||
except ValidationError as exc:
|
||||
error = exc.errors()[0]
|
||||
field = error["loc"][-1] if error["loc"] else "input"
|
||||
msg = error["msg"]
|
||||
labels = {
|
||||
"given_name": "Given name",
|
||||
"family_name": "Family name",
|
||||
"preferred_username": "Display name",
|
||||
"email": "Email",
|
||||
"phone_number": "Phone number",
|
||||
"picture": "Picture URL",
|
||||
"locale": "Locale",
|
||||
}
|
||||
label = labels.get(str(field), str(field))
|
||||
display_msg = msg.removeprefix("Value error, ") if error["type"] == "value_error" else f"{label}: {msg}"
|
||||
return HTMLResponse(f'<div role="alert">{display_msg}</div>')
|
||||
return HTMLResponse(format_validation_errors(exc))
|
||||
|
||||
user_repo = request.app.state.user_repo
|
||||
user = await user_repo.get_by_userid(userid)
|
||||
|
|
@ -205,13 +195,17 @@ async def update_user_groups(
|
|||
if admin is None:
|
||||
return HTMLResponse("Forbidden", status_code=403)
|
||||
|
||||
try:
|
||||
validated = GroupListInput(groups=groups)
|
||||
except ValidationError as exc:
|
||||
return HTMLResponse(format_validation_errors(exc))
|
||||
|
||||
user_repo = request.app.state.user_repo
|
||||
user = await user_repo.get_by_userid(userid)
|
||||
if user is None:
|
||||
return HTMLResponse("User not found", status_code=404)
|
||||
|
||||
group_list = [g.strip() for g in groups.split(",") if g.strip()]
|
||||
updated = user.model_copy(update={"groups": group_list})
|
||||
updated = user.model_copy(update={"groups": validated.group_list})
|
||||
await user_repo.update(updated)
|
||||
return HTMLResponse('<div role="status">Groups updated</div>')
|
||||
|
||||
|
|
|
|||
|
|
@ -7,7 +7,7 @@ from pydantic import ValidationError
|
|||
|
||||
from porchlight.dependencies import get_session_user
|
||||
from porchlight.models import PasswordCredential, WebAuthnCredential
|
||||
from porchlight.validation import ProfileUpdate
|
||||
from porchlight.validation import PasswordChange, PasswordSet, ProfileUpdate, format_validation_errors
|
||||
|
||||
router = APIRouter(prefix="/manage", tags=["manage"])
|
||||
|
||||
|
|
@ -55,6 +55,7 @@ async def set_password(
|
|||
request: Request,
|
||||
password: str = Form(),
|
||||
confirm: str = Form(),
|
||||
current_password: str = Form(""),
|
||||
) -> Response:
|
||||
session_user = get_session_user(request)
|
||||
if session_user is None:
|
||||
|
|
@ -64,15 +65,30 @@ async def set_password(
|
|||
cred_repo = request.app.state.credential_repo
|
||||
password_service = request.app.state.password_service
|
||||
|
||||
if password != confirm:
|
||||
return HTMLResponse('<div role="alert">Passwords do not match</div>')
|
||||
|
||||
if len(password) < 8:
|
||||
return HTMLResponse('<div role="alert">Password must be at least 8 characters</div>')
|
||||
|
||||
password_hash = password_service.hash(password)
|
||||
|
||||
existing = await cred_repo.get_password_by_user(userid)
|
||||
has_password = existing is not None
|
||||
|
||||
# Validate input
|
||||
try:
|
||||
if has_password:
|
||||
validated = PasswordChange(
|
||||
current_password=current_password,
|
||||
password=password,
|
||||
confirm=confirm,
|
||||
)
|
||||
else:
|
||||
validated = PasswordSet(password=password, confirm=confirm)
|
||||
except ValidationError as exc:
|
||||
return HTMLResponse(format_validation_errors(exc))
|
||||
|
||||
# Verify current password if changing
|
||||
if has_password:
|
||||
if not password_service.verify(existing.password_hash, validated.current_password):
|
||||
return HTMLResponse('<div role="alert">Current password is incorrect</div>')
|
||||
|
||||
# Store new password
|
||||
password_hash = password_service.hash(validated.password)
|
||||
|
||||
if existing is not None:
|
||||
await cred_repo.delete_password(userid)
|
||||
|
||||
|
|
@ -225,23 +241,7 @@ async def update_profile(
|
|||
locale=locale,
|
||||
)
|
||||
except ValidationError as exc:
|
||||
error = exc.errors()[0]
|
||||
field = error["loc"][-1] if error["loc"] else "input"
|
||||
msg = error["msg"]
|
||||
# Produce user-friendly field labels
|
||||
labels = {
|
||||
"given_name": "Given name",
|
||||
"family_name": "Family name",
|
||||
"preferred_username": "Display name",
|
||||
"email": "Email",
|
||||
"phone_number": "Phone number",
|
||||
"picture": "Picture URL",
|
||||
"locale": "Locale",
|
||||
}
|
||||
label = labels.get(str(field), str(field))
|
||||
# Use custom message for value errors (e.g. picture URL), generic pydantic message otherwise
|
||||
display_msg = msg.removeprefix("Value error, ") if error["type"] == "value_error" else f"{label}: {msg}"
|
||||
return HTMLResponse(f'<div role="alert">{display_msg}</div>')
|
||||
return HTMLResponse(format_validation_errors(exc))
|
||||
|
||||
user_repo = request.app.state.user_repo
|
||||
user = await user_repo.get_by_userid(userid)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue