714 lines
21 KiB
Python
714 lines
21 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.audit.logging import audit_event
|
|
from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope
|
|
from govoplan_core.core.views import VIEW_SURFACE_CONTRACT_VERSION, ViewSurface
|
|
from govoplan_core.db.session import get_session
|
|
from govoplan_views.backend.manifest import (
|
|
ASSIGNMENT_READ_SCOPE,
|
|
ASSIGNMENT_WRITE_SCOPE,
|
|
DEFINITION_READ_SCOPE,
|
|
DEFINITION_WRITE_SCOPE,
|
|
SELECTION_READ_SCOPE,
|
|
SELECTION_WRITE_SCOPE,
|
|
SYSTEM_ASSIGNMENT_READ_SCOPE,
|
|
SYSTEM_ASSIGNMENT_WRITE_SCOPE,
|
|
SYSTEM_DEFINITION_READ_SCOPE,
|
|
SYSTEM_DEFINITION_WRITE_SCOPE,
|
|
)
|
|
from govoplan_views.backend.runtime import get_registry
|
|
from govoplan_views.backend.schemas import (
|
|
EffectiveViewOptionResponse,
|
|
EffectiveViewResponse,
|
|
ViewDiagnosticResponse,
|
|
ViewAssignmentCreateRequest,
|
|
ViewAssignmentListResponse,
|
|
ViewAssignmentResponse,
|
|
ViewAssignmentUpdateRequest,
|
|
ViewDefinitionCreateRequest,
|
|
ViewDefinitionListResponse,
|
|
ViewDefinitionResponse,
|
|
ViewDefinitionUpdateRequest,
|
|
ViewProvenanceResponse,
|
|
ViewRevisionCreateRequest,
|
|
ViewRevisionResponse,
|
|
ViewSelectionRequest,
|
|
ViewSurfaceCatalogueResponse,
|
|
ViewSurfaceResponse,
|
|
)
|
|
from govoplan_views.backend.service import (
|
|
EffectiveViewState,
|
|
ViewsConflictError,
|
|
ViewsError,
|
|
ViewsNotFoundError,
|
|
ViewsValidationError,
|
|
archive_definition,
|
|
assignment_payload,
|
|
create_assignment,
|
|
create_definition,
|
|
create_revision,
|
|
definition_payload,
|
|
definition_revisions,
|
|
delete_assignment,
|
|
get_assignment,
|
|
get_definition,
|
|
get_revision,
|
|
list_assignments,
|
|
list_definitions,
|
|
publish_revision,
|
|
resolve_effective_view,
|
|
select_view,
|
|
update_assignment,
|
|
update_definition,
|
|
)
|
|
|
|
|
|
router = APIRouter(prefix="/views", tags=["views"])
|
|
|
|
|
|
def _catalogue() -> tuple[ViewSurface, ...]:
|
|
return get_registry().view_surfaces()
|
|
|
|
|
|
def _require_any_scope(principal: ApiPrincipal, *scopes: str) -> None:
|
|
if any(has_scope(principal, scope) for scope in scopes):
|
|
return
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=f"Requires one of: {', '.join(scopes)}",
|
|
)
|
|
|
|
|
|
def _require_definition_read(principal: ApiPrincipal, scope_type: str) -> None:
|
|
if scope_type == "system":
|
|
_require_any_scope(
|
|
principal,
|
|
SYSTEM_DEFINITION_READ_SCOPE,
|
|
SYSTEM_DEFINITION_WRITE_SCOPE,
|
|
)
|
|
return
|
|
_require_any_scope(
|
|
principal,
|
|
DEFINITION_READ_SCOPE,
|
|
DEFINITION_WRITE_SCOPE,
|
|
)
|
|
|
|
|
|
def _require_definition_write(principal: ApiPrincipal, scope_type: str) -> None:
|
|
_require_any_scope(
|
|
principal,
|
|
SYSTEM_DEFINITION_WRITE_SCOPE
|
|
if scope_type == "system"
|
|
else DEFINITION_WRITE_SCOPE,
|
|
)
|
|
|
|
|
|
def _require_assignment_read(principal: ApiPrincipal, scope_type: str) -> None:
|
|
if scope_type == "system":
|
|
_require_any_scope(
|
|
principal,
|
|
SYSTEM_ASSIGNMENT_READ_SCOPE,
|
|
SYSTEM_ASSIGNMENT_WRITE_SCOPE,
|
|
)
|
|
return
|
|
_require_any_scope(
|
|
principal,
|
|
ASSIGNMENT_READ_SCOPE,
|
|
ASSIGNMENT_WRITE_SCOPE,
|
|
)
|
|
|
|
|
|
def _require_assignment_write(principal: ApiPrincipal, scope_type: str) -> None:
|
|
_require_any_scope(
|
|
principal,
|
|
SYSTEM_ASSIGNMENT_WRITE_SCOPE
|
|
if scope_type == "system"
|
|
else ASSIGNMENT_WRITE_SCOPE,
|
|
)
|
|
|
|
|
|
def _actor_id(principal: ApiPrincipal) -> str:
|
|
return principal.account_id
|
|
|
|
|
|
def _http_error(exc: ViewsError) -> HTTPException:
|
|
if isinstance(exc, ViewsNotFoundError):
|
|
return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
|
|
if isinstance(exc, ViewsConflictError):
|
|
return HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc))
|
|
if isinstance(exc, ViewsValidationError):
|
|
return HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
|
detail=str(exc),
|
|
)
|
|
return HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=str(exc),
|
|
)
|
|
|
|
|
|
def _audit(
|
|
session: Session,
|
|
principal: ApiPrincipal,
|
|
*,
|
|
action: str,
|
|
object_type: str,
|
|
object_id: str,
|
|
details: dict[str, object],
|
|
) -> None:
|
|
audit_event(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
user_id=getattr(principal.user, "id", None),
|
|
api_key_id=principal.api_key_id,
|
|
action=action,
|
|
object_type=object_type,
|
|
object_id=object_id,
|
|
details=details,
|
|
)
|
|
|
|
|
|
def _effective_response(state: EffectiveViewState) -> EffectiveViewResponse:
|
|
effective = state.effective
|
|
return EffectiveViewResponse(
|
|
active_view_id=effective.view_id,
|
|
active_revision_id=effective.revision_id,
|
|
active_view_name=effective.name,
|
|
visible_surface_ids=sorted(effective.visible_surface_ids),
|
|
locked=effective.locked,
|
|
available_views=[
|
|
EffectiveViewOptionResponse(
|
|
id=option.id,
|
|
name=option.name,
|
|
description=option.description,
|
|
revision_id=option.revision_id,
|
|
)
|
|
for option in state.available_views
|
|
],
|
|
provenance=[
|
|
ViewProvenanceResponse.model_validate(item) for item in effective.provenance
|
|
],
|
|
diagnostics=[
|
|
ViewDiagnosticResponse(
|
|
severity=item.severity,
|
|
code=item.code,
|
|
message=item.message,
|
|
surface_ids=list(item.surface_ids),
|
|
)
|
|
for item in state.diagnostics
|
|
],
|
|
)
|
|
|
|
|
|
def _definition_response(
|
|
session: Session,
|
|
definition,
|
|
*,
|
|
readonly: bool,
|
|
) -> ViewDefinitionResponse:
|
|
return ViewDefinitionResponse.model_validate(
|
|
definition_payload(
|
|
session,
|
|
definition,
|
|
readonly=readonly,
|
|
catalogue=_catalogue(),
|
|
)
|
|
)
|
|
|
|
|
|
@router.get("/effective", response_model=EffectiveViewResponse)
|
|
def api_effective_view(
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> EffectiveViewResponse:
|
|
_require_any_scope(
|
|
principal,
|
|
SELECTION_READ_SCOPE,
|
|
SELECTION_WRITE_SCOPE,
|
|
)
|
|
return _effective_response(
|
|
resolve_effective_view(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
account_id=principal.account_id,
|
|
group_ids=principal.group_ids,
|
|
catalogue=_catalogue(),
|
|
)
|
|
)
|
|
|
|
|
|
@router.put("/selection", response_model=EffectiveViewResponse)
|
|
def api_select_view(
|
|
payload: ViewSelectionRequest,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> EffectiveViewResponse:
|
|
_require_any_scope(principal, SELECTION_WRITE_SCOPE)
|
|
try:
|
|
state = select_view(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
account_id=principal.account_id,
|
|
group_ids=principal.group_ids,
|
|
view_id=payload.view_id,
|
|
catalogue=_catalogue(),
|
|
)
|
|
_audit(
|
|
session,
|
|
principal,
|
|
action="views.selection.update",
|
|
object_type="view_preference",
|
|
object_id=principal.account_id,
|
|
details={"view_id": payload.view_id},
|
|
)
|
|
session.commit()
|
|
return _effective_response(state)
|
|
except ViewsError as exc:
|
|
session.rollback()
|
|
raise _http_error(exc) from exc
|
|
|
|
|
|
@router.get("/surfaces", response_model=ViewSurfaceCatalogueResponse)
|
|
def api_view_surfaces(
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> ViewSurfaceCatalogueResponse:
|
|
_require_any_scope(
|
|
principal,
|
|
DEFINITION_READ_SCOPE,
|
|
DEFINITION_WRITE_SCOPE,
|
|
SYSTEM_DEFINITION_READ_SCOPE,
|
|
SYSTEM_DEFINITION_WRITE_SCOPE,
|
|
)
|
|
lockout_ids = {
|
|
"access.module",
|
|
"access.nav.admin",
|
|
"access.route.admin",
|
|
"views.module",
|
|
"views.selector",
|
|
"views.admin.system",
|
|
"views.admin.tenant",
|
|
}
|
|
return ViewSurfaceCatalogueResponse(
|
|
contract_version=VIEW_SURFACE_CONTRACT_VERSION,
|
|
surfaces=[
|
|
ViewSurfaceResponse(
|
|
id=surface.id,
|
|
module_id=surface.module_id,
|
|
kind=surface.kind,
|
|
label=surface.label,
|
|
parent_id=surface.parent_id,
|
|
description=surface.description,
|
|
order=surface.order,
|
|
default_visible=surface.default_visible,
|
|
required=surface.required,
|
|
required_for_locked_view=surface.id in lockout_ids,
|
|
)
|
|
for surface in _catalogue()
|
|
],
|
|
)
|
|
|
|
|
|
@router.get("/definitions", response_model=ViewDefinitionListResponse)
|
|
def api_list_definitions(
|
|
scope_type: str = Query(default="tenant", pattern="^(system|tenant)$"),
|
|
include_inherited: bool = True,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> ViewDefinitionListResponse:
|
|
_require_definition_read(principal, scope_type)
|
|
definitions = list_definitions(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
scope_type=scope_type,
|
|
include_inherited=include_inherited,
|
|
)
|
|
return ViewDefinitionListResponse(
|
|
definitions=[
|
|
_definition_response(
|
|
session,
|
|
definition,
|
|
readonly=(scope_type == "tenant" and definition.scope_type == "system"),
|
|
)
|
|
for definition in definitions
|
|
]
|
|
)
|
|
|
|
|
|
@router.post(
|
|
"/definitions",
|
|
response_model=ViewDefinitionResponse,
|
|
status_code=status.HTTP_201_CREATED,
|
|
)
|
|
def api_create_definition(
|
|
payload: ViewDefinitionCreateRequest,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> ViewDefinitionResponse:
|
|
_require_definition_write(principal, payload.scope_type)
|
|
try:
|
|
definition = create_definition(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
scope_type=payload.scope_type,
|
|
definition_key=payload.definition_key,
|
|
name=payload.name,
|
|
description=payload.description,
|
|
visible_surface_ids=payload.visible_surface_ids,
|
|
catalogue=_catalogue(),
|
|
actor_id=_actor_id(principal),
|
|
)
|
|
_audit(
|
|
session,
|
|
principal,
|
|
action="views.definition.create",
|
|
object_type="view_definition",
|
|
object_id=definition.id,
|
|
details={"scope_type": definition.scope_type},
|
|
)
|
|
session.commit()
|
|
return _definition_response(session, definition, readonly=False)
|
|
except ViewsError as exc:
|
|
session.rollback()
|
|
raise _http_error(exc) from exc
|
|
|
|
|
|
@router.patch(
|
|
"/definitions/{definition_id}",
|
|
response_model=ViewDefinitionResponse,
|
|
)
|
|
def api_update_definition(
|
|
definition_id: str,
|
|
payload: ViewDefinitionUpdateRequest,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> ViewDefinitionResponse:
|
|
try:
|
|
definition = get_definition(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
definition_id=definition_id,
|
|
)
|
|
_require_definition_write(principal, definition.scope_type)
|
|
update_definition(
|
|
session,
|
|
definition,
|
|
name=payload.name,
|
|
description=payload.description,
|
|
actor_id=_actor_id(principal),
|
|
fields_set=payload.model_fields_set,
|
|
)
|
|
_audit(
|
|
session,
|
|
principal,
|
|
action="views.definition.update",
|
|
object_type="view_definition",
|
|
object_id=definition.id,
|
|
details={"fields": sorted(payload.model_fields_set)},
|
|
)
|
|
session.commit()
|
|
return _definition_response(session, definition, readonly=False)
|
|
except ViewsError as exc:
|
|
session.rollback()
|
|
raise _http_error(exc) from exc
|
|
|
|
|
|
@router.get(
|
|
"/definitions/{definition_id}/revisions",
|
|
response_model=list[ViewRevisionResponse],
|
|
)
|
|
def api_list_revisions(
|
|
definition_id: str,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> list[ViewRevisionResponse]:
|
|
try:
|
|
definition = get_definition(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
definition_id=definition_id,
|
|
)
|
|
_require_definition_read(principal, definition.scope_type)
|
|
return [
|
|
ViewRevisionResponse.model_validate(revision)
|
|
for revision in definition_revisions(
|
|
session,
|
|
definition_id=definition.id,
|
|
)
|
|
]
|
|
except ViewsError as exc:
|
|
raise _http_error(exc) from exc
|
|
|
|
|
|
@router.post(
|
|
"/definitions/{definition_id}/revisions",
|
|
response_model=ViewDefinitionResponse,
|
|
)
|
|
def api_create_revision(
|
|
definition_id: str,
|
|
payload: ViewRevisionCreateRequest,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> ViewDefinitionResponse:
|
|
try:
|
|
definition = get_definition(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
definition_id=definition_id,
|
|
)
|
|
_require_definition_write(principal, definition.scope_type)
|
|
revision = create_revision(
|
|
session,
|
|
definition,
|
|
visible_surface_ids=payload.visible_surface_ids,
|
|
catalogue=_catalogue(),
|
|
actor_id=_actor_id(principal),
|
|
)
|
|
_audit(
|
|
session,
|
|
principal,
|
|
action="views.revision.create",
|
|
object_type="view_definition",
|
|
object_id=definition.id,
|
|
details={
|
|
"revision": revision.revision,
|
|
"content_hash": revision.content_hash,
|
|
},
|
|
)
|
|
session.commit()
|
|
return _definition_response(session, definition, readonly=False)
|
|
except ViewsError as exc:
|
|
session.rollback()
|
|
raise _http_error(exc) from exc
|
|
|
|
|
|
@router.post(
|
|
"/definitions/{definition_id}/revisions/{revision_id}/publish",
|
|
response_model=ViewDefinitionResponse,
|
|
)
|
|
def api_publish_revision(
|
|
definition_id: str,
|
|
revision_id: str,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> ViewDefinitionResponse:
|
|
try:
|
|
definition = get_definition(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
definition_id=definition_id,
|
|
)
|
|
_require_definition_write(principal, definition.scope_type)
|
|
revision = get_revision(
|
|
session,
|
|
definition_id=definition.id,
|
|
revision_id=revision_id,
|
|
)
|
|
publish_revision(
|
|
session,
|
|
definition,
|
|
revision,
|
|
catalogue=_catalogue(),
|
|
actor_id=_actor_id(principal),
|
|
)
|
|
_audit(
|
|
session,
|
|
principal,
|
|
action="views.revision.publish",
|
|
object_type="view_definition",
|
|
object_id=definition.id,
|
|
details={"revision": revision.revision, "revision_id": revision.id},
|
|
)
|
|
session.commit()
|
|
return _definition_response(session, definition, readonly=False)
|
|
except ViewsError as exc:
|
|
session.rollback()
|
|
raise _http_error(exc) from exc
|
|
|
|
|
|
@router.delete(
|
|
"/definitions/{definition_id}",
|
|
response_model=ViewDefinitionResponse,
|
|
)
|
|
def api_archive_definition(
|
|
definition_id: str,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> ViewDefinitionResponse:
|
|
try:
|
|
definition = get_definition(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
definition_id=definition_id,
|
|
)
|
|
_require_definition_write(principal, definition.scope_type)
|
|
archive_definition(
|
|
session,
|
|
definition,
|
|
actor_id=_actor_id(principal),
|
|
)
|
|
_audit(
|
|
session,
|
|
principal,
|
|
action="views.definition.archive",
|
|
object_type="view_definition",
|
|
object_id=definition.id,
|
|
details={},
|
|
)
|
|
session.commit()
|
|
return _definition_response(session, definition, readonly=False)
|
|
except ViewsError as exc:
|
|
session.rollback()
|
|
raise _http_error(exc) from exc
|
|
|
|
|
|
@router.get("/assignments", response_model=ViewAssignmentListResponse)
|
|
def api_list_assignments(
|
|
scope_type: str = Query(default="tenant", pattern="^(system|tenant)$"),
|
|
include_inherited: bool = True,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> ViewAssignmentListResponse:
|
|
_require_assignment_read(principal, scope_type)
|
|
assignments = list_assignments(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
scope_type=scope_type,
|
|
include_inherited=include_inherited,
|
|
)
|
|
return ViewAssignmentListResponse(
|
|
assignments=[
|
|
ViewAssignmentResponse.model_validate(assignment_payload(assignment))
|
|
for assignment in assignments
|
|
]
|
|
)
|
|
|
|
|
|
@router.post(
|
|
"/assignments",
|
|
response_model=ViewAssignmentResponse,
|
|
status_code=status.HTTP_201_CREATED,
|
|
)
|
|
def api_create_assignment(
|
|
payload: ViewAssignmentCreateRequest,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> ViewAssignmentResponse:
|
|
_require_assignment_write(principal, payload.scope_type)
|
|
try:
|
|
definition = get_definition(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
definition_id=payload.definition_id,
|
|
)
|
|
assignment = create_assignment(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
scope_type=payload.scope_type,
|
|
scope_id=payload.scope_id,
|
|
definition=definition,
|
|
revision_id=payload.revision_id,
|
|
mode=payload.mode,
|
|
priority=payload.priority,
|
|
is_active=payload.is_active,
|
|
metadata=payload.metadata,
|
|
catalogue=_catalogue(),
|
|
actor_id=_actor_id(principal),
|
|
)
|
|
_audit(
|
|
session,
|
|
principal,
|
|
action="views.assignment.create",
|
|
object_type="view_assignment",
|
|
object_id=assignment.id,
|
|
details={
|
|
"scope_type": assignment.scope_type,
|
|
"scope_id": assignment.scope_id,
|
|
"definition_id": assignment.definition_id,
|
|
"mode": assignment.mode,
|
|
},
|
|
)
|
|
session.commit()
|
|
return ViewAssignmentResponse.model_validate(assignment_payload(assignment))
|
|
except ViewsError as exc:
|
|
session.rollback()
|
|
raise _http_error(exc) from exc
|
|
|
|
|
|
@router.patch(
|
|
"/assignments/{assignment_id}",
|
|
response_model=ViewAssignmentResponse,
|
|
)
|
|
def api_update_assignment(
|
|
assignment_id: str,
|
|
payload: ViewAssignmentUpdateRequest,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> ViewAssignmentResponse:
|
|
try:
|
|
assignment = get_assignment(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
assignment_id=assignment_id,
|
|
)
|
|
_require_assignment_write(principal, assignment.scope_type)
|
|
update_assignment(
|
|
session,
|
|
assignment,
|
|
updates=payload.model_dump(exclude_unset=True),
|
|
catalogue=_catalogue(),
|
|
actor_id=_actor_id(principal),
|
|
)
|
|
_audit(
|
|
session,
|
|
principal,
|
|
action="views.assignment.update",
|
|
object_type="view_assignment",
|
|
object_id=assignment.id,
|
|
details={"fields": sorted(payload.model_fields_set)},
|
|
)
|
|
session.commit()
|
|
return ViewAssignmentResponse.model_validate(assignment_payload(assignment))
|
|
except ViewsError as exc:
|
|
session.rollback()
|
|
raise _http_error(exc) from exc
|
|
|
|
|
|
@router.delete(
|
|
"/assignments/{assignment_id}",
|
|
status_code=status.HTTP_204_NO_CONTENT,
|
|
)
|
|
def api_delete_assignment(
|
|
assignment_id: str,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(get_api_principal),
|
|
) -> None:
|
|
try:
|
|
assignment = get_assignment(
|
|
session,
|
|
tenant_id=principal.tenant_id,
|
|
assignment_id=assignment_id,
|
|
)
|
|
_require_assignment_write(principal, assignment.scope_type)
|
|
assignment_details = {
|
|
"scope_type": assignment.scope_type,
|
|
"scope_id": assignment.scope_id,
|
|
"definition_id": assignment.definition_id,
|
|
"mode": assignment.mode,
|
|
}
|
|
delete_assignment(session, assignment)
|
|
_audit(
|
|
session,
|
|
principal,
|
|
action="views.assignment.delete",
|
|
object_type="view_assignment",
|
|
object_id=assignment_id,
|
|
details=assignment_details,
|
|
)
|
|
session.commit()
|
|
except ViewsError as exc:
|
|
session.rollback()
|
|
raise _http_error(exc) from exc
|