216 lines
No EOL
7.2 KiB
Python
216 lines
No EOL
7.2 KiB
Python
from fastapi import APIRouter, HTTPException, Depends, Body, Path
|
|
from typing import List, Dict, Any
|
|
from fastapi import status
|
|
from datetime import datetime
|
|
from dataclasses import dataclass
|
|
|
|
# Import interfaces
|
|
from modules.auth import get_current_active_user, get_user_context
|
|
from modules.gateway_interface import get_gateway_interface
|
|
from modules.gateway_model import Mandate
|
|
|
|
# Determine all attributes of the model (except internal/special attributes)
|
|
def get_model_attributes(model_class):
|
|
return [attr for attr in dir(model_class)
|
|
if not callable(getattr(model_class, attr))
|
|
and not attr.startswith('_')
|
|
and attr != 'metadata'
|
|
and attr != 'query'
|
|
and attr != 'query_class'
|
|
and attr != 'label'
|
|
and attr != 'field_labels']
|
|
|
|
# Model attributes for Mandate
|
|
mandate_attributes = get_model_attributes(Mandate)
|
|
|
|
@dataclass
|
|
class AppContext:
|
|
"""Context object for all required connections and user information"""
|
|
mandate_id: int
|
|
user_id: int
|
|
interface_data: Any # Gateway Interface
|
|
|
|
async def get_context(current_user: Dict[str, Any]) -> AppContext:
|
|
"""
|
|
Creates a central context object with all required interfaces
|
|
|
|
Args:
|
|
current_user: Current user from authentication
|
|
|
|
Returns:
|
|
AppContext object with all required connections
|
|
"""
|
|
mandate_id, user_id = await get_user_context(current_user)
|
|
interface_data = get_gateway_interface(mandate_id, user_id)
|
|
|
|
return AppContext(
|
|
mandate_id=mandate_id,
|
|
user_id=user_id,
|
|
interface_data=interface_data
|
|
)
|
|
|
|
# Create router for mandate endpoints
|
|
router = APIRouter(
|
|
prefix="/api/mandates",
|
|
tags=["Mandates"],
|
|
responses={404: {"description": "Not found"}}
|
|
)
|
|
|
|
@router.get("", response_model=List[Dict[str, Any]])
|
|
async def get_mandates(current_user: Dict[str, Any] = Depends(get_current_active_user)):
|
|
"""Get all available mandates (only for SysAdmin users)"""
|
|
context = await get_context(current_user)
|
|
|
|
# Permission check
|
|
if current_user.get("privilege") != "sysadmin":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Only system administrators can access all mandates"
|
|
)
|
|
|
|
# Get mandates generically
|
|
return context.interface_data.get_all_mandates()
|
|
|
|
|
|
@router.post("", response_model=Dict[str, Any])
|
|
async def create_mandate(
|
|
mandate: Dict[str, Any] = Body(...),
|
|
current_user: Dict[str, Any] = Depends(get_current_active_user)
|
|
):
|
|
"""Create a new mandate (only for SysAdmin users)"""
|
|
context = await get_context(current_user)
|
|
|
|
# Permission check
|
|
if current_user.get("privilege") != "sysadmin":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Only system administrators can create mandates"
|
|
)
|
|
|
|
# Set attributes from the request dynamically
|
|
mandate_data = {}
|
|
for attr in mandate_attributes:
|
|
if attr in mandate:
|
|
mandate_data[attr] = mandate[attr]
|
|
|
|
# Default values for missing fields
|
|
if "name" not in mandate_data:
|
|
mandate_data["name"] = "New Mandate"
|
|
if "language" not in mandate_data:
|
|
mandate_data["language"] = "de"
|
|
|
|
# Create mandate
|
|
new_mandate = context.interface_data.create_mandate(**mandate_data)
|
|
|
|
return new_mandate
|
|
|
|
|
|
@router.get("/{mandate_id}", response_model=Dict[str, Any])
|
|
async def get_mandate(
|
|
mandate_id: int,
|
|
current_user: Dict[str, Any] = Depends(get_current_active_user)
|
|
):
|
|
"""Get a specific mandate"""
|
|
context = await get_context(current_user)
|
|
|
|
# Permission check
|
|
# Admin can only see their own mandate, SysAdmin can see all
|
|
is_admin = current_user.get("privilege") == "admin"
|
|
is_sysadmin = current_user.get("privilege") == "sysadmin"
|
|
is_own_mandate = context.mandate_id == mandate_id
|
|
|
|
if (is_admin and not is_own_mandate) and not is_sysadmin:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="No permission to access this mandate"
|
|
)
|
|
|
|
# Get mandate generically
|
|
mandate = context.interface_data.get_mandate(mandate_id)
|
|
if not mandate:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Mandate with ID {mandate_id} not found"
|
|
)
|
|
|
|
return mandate
|
|
|
|
@router.put("/{mandate_id}", response_model=Dict[str, Any])
|
|
async def update_mandate(
|
|
mandate_id: int = Path(..., description="ID of the mandate to update"),
|
|
mandate_data: Dict[str, Any] = Body(..., description="Updated mandate data"),
|
|
current_user: Dict[str, Any] = Depends(get_current_active_user)
|
|
):
|
|
"""Update an existing mandate"""
|
|
context = await get_context(current_user)
|
|
|
|
# Mandate exists?
|
|
mandate = context.interface_data.get_mandate(mandate_id)
|
|
if not mandate:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Mandate with ID {mandate_id} not found"
|
|
)
|
|
|
|
# Permission check
|
|
is_admin = current_user.get("privilege") == "admin"
|
|
is_sysadmin = current_user.get("privilege") == "sysadmin"
|
|
is_own_mandate = context.mandate_id == mandate_id
|
|
|
|
if (is_admin and not is_own_mandate) and not is_sysadmin:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="No permission to update this mandate"
|
|
)
|
|
|
|
# Dynamically filter attributes from the request into update_data
|
|
update_data = {}
|
|
for attr in mandate_attributes:
|
|
if attr in mandate_data:
|
|
update_data[attr] = mandate_data[attr]
|
|
|
|
# Update mandate
|
|
updated_mandate = context.interface_data.update_mandate(
|
|
mandate_id=mandate_id,
|
|
mandate_data=update_data
|
|
)
|
|
|
|
return updated_mandate
|
|
|
|
@router.delete("/{mandate_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def delete_mandate(
|
|
mandate_id: int = Path(..., description="ID of the mandate to delete"),
|
|
current_user: Dict[str, Any] = Depends(get_current_active_user)
|
|
):
|
|
"""Delete a mandate, including all associated users and referenced objects"""
|
|
context = await get_context(current_user)
|
|
|
|
# Mandate exists?
|
|
mandate = context.interface_data.get_mandate(mandate_id)
|
|
if not mandate:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Mandate with ID {mandate_id} not found"
|
|
)
|
|
|
|
# Permission check
|
|
is_admin = current_user.get("privilege") == "admin"
|
|
is_sysadmin = current_user.get("privilege") == "sysadmin"
|
|
is_own_mandate = context.mandate_id == mandate_id
|
|
|
|
if (is_admin and not is_own_mandate) and not is_sysadmin:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="No permission to delete this mandate"
|
|
)
|
|
|
|
# Delete mandate
|
|
success = context.interface_data.delete_mandate(mandate_id)
|
|
if not success:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Error deleting mandate with ID {mandate_id}"
|
|
)
|
|
|
|
# Return no content on successful deletion
|
|
return None |