145 lines
5.0 KiB
Python
145 lines
5.0 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Awaitable, Callable, Iterable
|
|
from typing import Any
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from api_server.html5_access import (
|
|
render_html5_access_page,
|
|
render_html5_access_profile_apply_result,
|
|
render_html5_access_profile_preview,
|
|
render_html5_access_publish_plan,
|
|
render_html5_access_publish_result,
|
|
render_html5_access_user_detail,
|
|
)
|
|
from api_server.html5_forms import form_value, html5_csv_values
|
|
|
|
|
|
def html5_access_page(
|
|
*,
|
|
project_id: str,
|
|
profile: str | None,
|
|
project_summaries: Callable[[], Iterable[object]],
|
|
normalized_project: Callable[[str], object],
|
|
access_profile_by_name: Callable[[object, str], object | None],
|
|
access_publish_plan: Callable[[object, object], object],
|
|
) -> str:
|
|
try:
|
|
normalized = normalized_project(project_id)
|
|
selected = _selected_profile(normalized, profile, access_profile_by_name)
|
|
plan = access_publish_plan(normalized, selected) if selected is not None else None
|
|
return render_html5_access_page(
|
|
project_id=project_id,
|
|
projects=project_summaries(),
|
|
normalized=normalized,
|
|
selected_profile=profile,
|
|
plan=plan,
|
|
)
|
|
except HTTPException as error:
|
|
return render_html5_access_page(
|
|
project_id=project_id,
|
|
projects=project_summaries(),
|
|
normalized=None,
|
|
error=str(error.detail),
|
|
)
|
|
|
|
|
|
def html5_access_publish_plan(
|
|
*,
|
|
project_id: str,
|
|
profile_name: str,
|
|
normalized_project: Callable[[str], object],
|
|
access_profile_by_name: Callable[[object, str], object | None],
|
|
access_publish_plan: Callable[[object, object], object],
|
|
) -> str:
|
|
normalized = normalized_project(project_id)
|
|
profile = access_profile_by_name(normalized, profile_name)
|
|
if profile is None:
|
|
raise HTTPException(status_code=404, detail="Access profile not found")
|
|
return render_html5_access_publish_plan(
|
|
project_id=project_id,
|
|
profile=profile,
|
|
plan=access_publish_plan(normalized, profile),
|
|
)
|
|
|
|
|
|
async def html5_access_publish_dry_run(
|
|
*,
|
|
project_id: str,
|
|
profile_name: str,
|
|
publish_dry_run: Callable[[str, str], Awaitable[Any]],
|
|
) -> str:
|
|
result = await publish_dry_run(project_id, profile_name)
|
|
return render_html5_access_publish_result(project_id=project_id, result=result)
|
|
|
|
|
|
async def html5_access_profile_preview(
|
|
*,
|
|
project_id: str,
|
|
form: dict[str, list[str]],
|
|
preview_profile: Callable[[str, object], Awaitable[Any]],
|
|
draft_request: Callable[..., object],
|
|
) -> str:
|
|
draft = await preview_profile(project_id, _draft_request_from_form(form, draft_request))
|
|
return render_html5_access_profile_preview(draft=draft)
|
|
|
|
|
|
async def html5_access_profile_apply(
|
|
*,
|
|
project_id: str,
|
|
form: dict[str, list[str]],
|
|
apply_profile: Callable[[str, object], Awaitable[Any]],
|
|
apply_request: Callable[..., object],
|
|
normalized_project: Callable[[str], object],
|
|
access_profile_by_name: Callable[[object, str], object | None],
|
|
access_publish_plan: Callable[[object, object], object],
|
|
) -> str:
|
|
request = _draft_request_from_form(form, apply_request, author="html5")
|
|
response = await apply_profile(project_id, request)
|
|
profile_name = str(response.profile.get("name") or response.profile.get("qualified_name") or "")
|
|
plan = None
|
|
if profile_name:
|
|
normalized = normalized_project(project_id)
|
|
profile = access_profile_by_name(normalized, profile_name)
|
|
if profile is not None:
|
|
plan = access_publish_plan(normalized, profile)
|
|
return render_html5_access_profile_apply_result(project_id=project_id, response=response, plan=plan)
|
|
|
|
|
|
async def html5_access_user_detail(
|
|
*,
|
|
project_id: str,
|
|
user_name: str,
|
|
access_user: Callable[[str, str], Awaitable[dict]],
|
|
) -> str:
|
|
return render_html5_access_user_detail(project_id=project_id, user_payload=await access_user(project_id, user_name))
|
|
|
|
|
|
def _selected_profile(
|
|
normalized: object,
|
|
profile_name: str | None,
|
|
access_profile_by_name: Callable[[object, str], object | None],
|
|
) -> object | None:
|
|
access = getattr(normalized, "access", None)
|
|
profiles = list(getattr(access, "profiles", []) or [])
|
|
if not profiles:
|
|
return None
|
|
if not profile_name:
|
|
return profiles[0]
|
|
return access_profile_by_name(normalized, profile_name) or profiles[0]
|
|
|
|
|
|
def _draft_request_from_form(form: dict[str, list[str]], request_factory: Callable[..., object], **extra: object) -> object:
|
|
name = form_value(form, "name")
|
|
if not name:
|
|
raise HTTPException(status_code=400, detail="Access profile name is required")
|
|
payload = {
|
|
"name": name,
|
|
"target_objects": html5_csv_values(form_value(form, "target_objects") or ""),
|
|
"permissions": html5_csv_values(form_value(form, "permissions") or "read"),
|
|
"source_user": form_value(form, "source_user"),
|
|
**extra,
|
|
}
|
|
return request_factory(**payload)
|