PowerOn/gateway
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

dyn options in api

Browse source
This commit is contained in:
ValueOn AG 2026-01-22 18:52:04 +01:00
parent 362080791a
commit f02ebead7c
7 changed files with 187 additions and 397 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
modules/datamodels/datamodelUam.py
View file

@ -100,11 +100,11 @@ registerModelLabels(
class UserConnection(BaseModel):
id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="Unique ID of the connection", json_schema_extra={"frontend_type": "text", "frontend_readonly": True, "frontend_required": False})
userId: str = Field(description="ID of the user this connection belongs to", json_schema_extra={"frontend_type": "text", "frontend_readonly": True, "frontend_required": False})
authority: AuthAuthority = Field(description="Authentication authority", json_schema_extra={"frontend_type": "select", "frontend_readonly": True, "frontend_required": False, "frontend_options": "auth.authority"})
authority: AuthAuthority = Field(description="Authentication authority", json_schema_extra={"frontend_type": "select", "frontend_readonly": True, "frontend_required": False, "frontend_options": "/api/connections/authorities/options"})
externalId: str = Field(description="User ID in the external system", json_schema_extra={"frontend_type": "text", "frontend_readonly": True, "frontend_required": False})
externalUsername: str = Field(description="Username in the external system", json_schema_extra={"frontend_type": "text", "frontend_readonly": False, "frontend_required": False})
externalEmail: Optional[EmailStr] = Field(None, description="Email in the external system", json_schema_extra={"frontend_type": "email", "frontend_readonly": False, "frontend_required": False})
status: ConnectionStatus = Field(default=ConnectionStatus.ACTIVE, description="Connection status", json_schema_extra={"frontend_type": "select", "frontend_readonly": False, "frontend_required": False, "frontend_options": "connection.status"})
status: ConnectionStatus = Field(default=ConnectionStatus.ACTIVE, description="Connection status", json_schema_extra={"frontend_type": "select", "frontend_readonly": False, "frontend_required": False, "frontend_options": "/api/connections/statuses/options"})
connectedAt: float = Field(default_factory=getUtcTimestamp, description="When the connection was established (UTC timestamp in seconds)", json_schema_extra={"frontend_type": "timestamp", "frontend_readonly": True, "frontend_required": False})
lastChecked: float = Field(default_factory=getUtcTimestamp, description="When the connection was last verified (UTC timestamp in seconds)", json_schema_extra={"frontend_type": "timestamp", "frontend_readonly": True, "frontend_required": False})
expiresAt: Optional[float] = Field(None, description="When the connection expires (UTC timestamp in seconds)", json_schema_extra={"frontend_type": "timestamp", "frontend_readonly": True, "frontend_required": False})
@ -198,7 +198,7 @@ class User(BaseModel):
authenticationAuthority: AuthAuthority = Field(
default=AuthAuthority.LOCAL,
description="Primary authentication authority",
json_schema_extra={"frontend_type": "select", "frontend_readonly": True, "frontend_required": False, "frontend_options": "auth.authority"}
json_schema_extra={"frontend_type": "select", "frontend_readonly": True, "frontend_required": False, "frontend_options": "/api/connections/authorities/options"}
)

26
modules/features/trustee/datamodelFeatureTrustee.py
View file

@ -138,7 +138,7 @@ class TrusteeAccess(BaseModel):
"frontend_type": "select",
"frontend_readonly": False,
"frontend_required": True,
"frontend_options": "TrusteeOrganisation"
"frontend_options": "/api/trustee/{instanceId}/organisations/options"
}
)
roleId: str = Field(
@ -147,7 +147,7 @@ class TrusteeAccess(BaseModel):
"frontend_type": "select",
"frontend_readonly": False,
"frontend_required": True,
"frontend_options": "TrusteeRole"
"frontend_options": "/api/trustee/{instanceId}/roles/options"
}
)
userId: str = Field(
@ -156,7 +156,7 @@ class TrusteeAccess(BaseModel):
"frontend_type": "select",
"frontend_readonly": False,
"frontend_required": True,
"frontend_options": "User"
"frontend_options": "/api/users/options"
}
)
contractId: Optional[str] = Field(
@ -166,7 +166,7 @@ class TrusteeAccess(BaseModel):
"frontend_type": "select",
"frontend_readonly": False,
"frontend_required": False,
"frontend_options": "TrusteeContract",
"frontend_options": "/api/trustee/{instanceId}/contracts/options",
"frontend_depends_on": "organisationId"
}
)
@ -223,7 +223,7 @@ class TrusteeContract(BaseModel):
"frontend_type": "select",
"frontend_readonly": False, # Editable at creation, then readonly
"frontend_required": True,
"frontend_options": "TrusteeOrganisation"
"frontend_options": "/api/trustee/{instanceId}/organisations/options"
}
)
label: str = Field(
@ -295,7 +295,7 @@ class TrusteeDocument(BaseModel):
"frontend_type": "select",
"frontend_readonly": False,
"frontend_required": True,
"frontend_options": "TrusteeOrganisation"
"frontend_options": "/api/trustee/{instanceId}/organisations/options"
}
)
contractId: str = Field(
@ -304,7 +304,7 @@ class TrusteeDocument(BaseModel):
"frontend_type": "select",
"frontend_readonly": False,
"frontend_required": True,
"frontend_options": "TrusteeContract",
"frontend_options": "/api/trustee/{instanceId}/contracts/options",
"frontend_depends_on": "organisationId"
}
)
@ -394,7 +394,7 @@ class TrusteePosition(BaseModel):
"frontend_type": "select",
"frontend_readonly": False,
"frontend_required": True,
"frontend_options": "TrusteeOrganisation"
"frontend_options": "/api/trustee/{instanceId}/organisations/options"
}
)
contractId: str = Field(
@ -403,7 +403,7 @@ class TrusteePosition(BaseModel):
"frontend_type": "select",
"frontend_readonly": False,
"frontend_required": True,
"frontend_options": "TrusteeContract",
"frontend_options": "/api/trustee/{instanceId}/contracts/options",
"frontend_depends_on": "organisationId"
}
)
@ -580,7 +580,7 @@ class TrusteePositionDocument(BaseModel):
"frontend_type": "select",
"frontend_readonly": False,
"frontend_required": True,
"frontend_options": "TrusteeOrganisation"
"frontend_options": "/api/trustee/{instanceId}/organisations/options"
}
)
contractId: str = Field(
@ -589,7 +589,7 @@ class TrusteePositionDocument(BaseModel):
"frontend_type": "select",
"frontend_readonly": False,
"frontend_required": True,
"frontend_options": "TrusteeContract",
"frontend_options": "/api/trustee/{instanceId}/contracts/options",
"frontend_depends_on": "organisationId"
}
)
@ -599,7 +599,7 @@ class TrusteePositionDocument(BaseModel):
"frontend_type": "select",
"frontend_readonly": False,
"frontend_required": True,
"frontend_options": "TrusteeDocument",
"frontend_options": "/api/trustee/{instanceId}/documents/options",
"frontend_depends_on": "contractId"
}
)
@ -609,7 +609,7 @@ class TrusteePositionDocument(BaseModel):
"frontend_type": "select",
"frontend_readonly": False,
"frontend_required": True,
"frontend_options": "TrusteePosition",
"frontend_options": "/api/trustee/{instanceId}/positions/options",
"frontend_depends_on": "contractId"
}
)

83
modules/features/trustee/routeFeatureTrustee.py
View file

@ -115,6 +115,89 @@ async def _validateInstanceAccess(instanceId: str, context: RequestContext) -> s
return str(instance.mandateId)
# ============================================================================
# OPTIONS ENDPOINTS (for dropdowns)
# ============================================================================
@router.get("/{instanceId}/organisations/options", response_model=List[Dict[str, Any]])
@limiter.limit("60/minute")
async def getOrganisationOptions(
request: Request,
instanceId: str = Path(..., description="Feature Instance ID"),
context: RequestContext = Depends(getRequestContext)
) -> List[Dict[str, Any]]:
"""Get organisation options for select dropdowns. Returns: [{ value, label }]"""
mandateId = await _validateInstanceAccess(instanceId, context)
interface = getInterface(context.user, mandateId=mandateId, featureInstanceId=instanceId)
result = interface.getAllOrganisations(None)
items = result.items if hasattr(result, 'items') else result
return [{"value": org.id, "label": org.label or org.id} for org in items]
@router.get("/{instanceId}/roles/options", response_model=List[Dict[str, Any]])
@limiter.limit("60/minute")
async def getRoleOptions(
request: Request,
instanceId: str = Path(..., description="Feature Instance ID"),
context: RequestContext = Depends(getRequestContext)
) -> List[Dict[str, Any]]:
"""Get role options for select dropdowns. Returns: [{ value, label }]"""
mandateId = await _validateInstanceAccess(instanceId, context)
interface = getInterface(context.user, mandateId=mandateId, featureInstanceId=instanceId)
result = interface.getAllRoles(None)
items = result.items if hasattr(result, 'items') else result
return [{"value": role.id, "label": role.desc or role.id} for role in items]
@router.get("/{instanceId}/contracts/options", response_model=List[Dict[str, Any]])
@limiter.limit("60/minute")
async def getContractOptions(
request: Request,
instanceId: str = Path(..., description="Feature Instance ID"),
context: RequestContext = Depends(getRequestContext)
) -> List[Dict[str, Any]]:
"""Get contract options for select dropdowns. Returns: [{ value, label }]"""
mandateId = await _validateInstanceAccess(instanceId, context)
interface = getInterface(context.user, mandateId=mandateId, featureInstanceId=instanceId)
result = interface.getAllContracts(None)
items = result.items if hasattr(result, 'items') else result
return [{"value": c.id, "label": c.label or c.name or c.id} for c in items]
@router.get("/{instanceId}/documents/options", response_model=List[Dict[str, Any]])
@limiter.limit("60/minute")
async def getDocumentOptions(
request: Request,
instanceId: str = Path(..., description="Feature Instance ID"),
context: RequestContext = Depends(getRequestContext)
) -> List[Dict[str, Any]]:
"""Get document options for select dropdowns. Returns: [{ value, label }]"""
mandateId = await _validateInstanceAccess(instanceId, context)
interface = getInterface(context.user, mandateId=mandateId, featureInstanceId=instanceId)
result = interface.getAllDocuments(None)
items = result.items if hasattr(result, 'items') else result
return [{"value": d.id, "label": d.name or d.id} for d in items]
@router.get("/{instanceId}/positions/options", response_model=List[Dict[str, Any]])
@limiter.limit("60/minute")
async def getPositionOptions(
request: Request,
instanceId: str = Path(..., description="Feature Instance ID"),
context: RequestContext = Depends(getRequestContext)
) -> List[Dict[str, Any]]:
"""Get position options for select dropdowns. Returns: [{ value, label }]"""
mandateId = await _validateInstanceAccess(instanceId, context)
interface = getInterface(context.user, mandateId=mandateId, featureInstanceId=instanceId)
result = interface.getAllPositions(None)
items = result.items if hasattr(result, 'items') else result
return [{"value": p.id, "label": p.title or p.id} for p in items]
# ============================================================================
# CRUD ENDPOINTS
# ============================================================================
# ===== Organisation Routes =====
@router.get("/{instanceId}/organisations", response_model=PaginatedResponse[TrusteeOrganisation])

46
modules/routes/routeDataConnections.py
View file

@ -92,6 +92,52 @@ router = APIRouter(
responses={404: {"description": "Not found"}}
)
# ============================================================================
# OPTIONS ENDPOINTS (for dropdowns)
# ============================================================================
@router.get("/statuses/options", response_model=List[Dict[str, Any]])
@limiter.limit("60/minute")
async def getConnectionStatusOptions(
request: Request,
currentUser: User = Depends(getCurrentUser)
) -> List[Dict[str, Any]]:
"""
Get connection status options for select dropdowns.
Returns standardized format: [{ value, label }]
"""
return [
{"value": status.value, "label": status.value.capitalize()}
for status in ConnectionStatus
]
@router.get("/authorities/options", response_model=List[Dict[str, Any]])
@limiter.limit("60/minute")
async def getAuthAuthorityOptions(
request: Request,
currentUser: User = Depends(getCurrentUser)
) -> List[Dict[str, Any]]:
"""
Get authentication authority options for select dropdowns.
Returns standardized format: [{ value, label }]
"""
authorityLabels = {
"local": "Local",
"google": "Google",
"msft": "Microsoft"
}
return [
{"value": auth.value, "label": authorityLabels.get(auth.value, auth.value)}
for auth in AuthAuthority
]
# ============================================================================
# CRUD ENDPOINTS
# ============================================================================
@router.get("/", response_model=PaginatedResponse[UserConnection])
@limiter.limit("30/minute")
async def get_connections(

42
modules/routes/routeDataUsers.py
View file

@ -145,6 +145,48 @@ router = APIRouter(
responses={404: {"description": "Not found"}}
)
# ============================================================================
# OPTIONS ENDPOINTS (for dropdowns)
# ============================================================================
@router.get("/options", response_model=List[Dict[str, Any]])
@limiter.limit("60/minute")
async def getUserOptions(
request: Request,
context: RequestContext = Depends(getRequestContext)
) -> List[Dict[str, Any]]:
"""
Get user options for select dropdowns.
MULTI-TENANT: mandateId from X-Mandate-Id header determines scope.
Returns standardized format: [{ value, label }]
"""
try:
appInterface = interfaceDbApp.getInterface(context.user)
if context.mandateId:
result = appInterface.getUsersByMandate(str(context.mandateId), None)
users = result.items if hasattr(result, 'items') else result
elif context.isSysAdmin:
users = appInterface.getAllUsers()
else:
raise HTTPException(status_code=403, detail="Access denied")
return [
{"value": user.id, "label": user.fullName or user.username or user.email or user.id}
for user in users
]
except HTTPException:
raise
except Exception as e:
logger.error(f"Error getting user options: {str(e)}")
raise HTTPException(status_code=500, detail=f"Failed to get user options: {str(e)}")
# ============================================================================
# CRUD ENDPOINTS
# ============================================================================
@router.get("/", response_model=PaginatedResponse[User])
@limiter.limit("30/minute")
async def get_users(

138
modules/shared/frontendOptionsTypes.py
View file

@ -1,138 +0,0 @@
# Copyright (c) 2025 Patrick Motsch
# All rights reserved.
"""
Type definitions and utilities for frontend_options attribute.
The frontend_options attribute supports two formats:
1. Static List: A list of option dictionaries for static options
2. String Reference: A string identifier that references dynamic options from /api/options/{optionsName}
"""
from typing import List, Dict, Any, Union
try:
from typing import TypeAlias # Python 3.10+
except ImportError:
from typing_extensions import TypeAlias # Python < 3.10
# Type definition for a single option item
OptionItem: TypeAlias = Dict[str, Any]
"""
Single option item format:
{
"value": str, # The value to be stored/returned
"label": { # Multilingual labels
"en": str,
"fr": str,
...
}
}
"""
# Type definition for frontend_options - can be either a list or string reference
FrontendOptions: TypeAlias = Union[List[OptionItem], str]
"""
frontend_options can be either:
1. List[OptionItem]: Static list of options
Example: [{"value": "a", "label": {"en": "All", "fr": "Tous"}}]
2. str: String reference to dynamic options API
Example: "user.role" -> Frontend fetches from /api/options/user.role
"""
def isStringReference(frontendOptions: FrontendOptions) -> bool:
"""
Check if frontend_options is a string reference (dynamic) or a list (static).
Args:
frontendOptions: The frontend_options value to check
Returns:
True if it's a string reference, False if it's a list
"""
return isinstance(frontendOptions, str)
def isStaticList(frontendOptions: FrontendOptions) -> bool:
"""
Check if frontend_options is a static list or a string reference.
Args:
frontendOptions: The frontend_options value to check
Returns:
True if it's a static list, False if it's a string reference
"""
return isinstance(frontendOptions, list)
def validateFrontendOptions(frontendOptions: FrontendOptions) -> bool:
"""
Validate that frontend_options is in the correct format.
Args:
frontendOptions: The frontend_options value to validate
Returns:
True if valid, False otherwise
"""
if isinstance(frontendOptions, str):
# String reference: should be a non-empty string
return bool(frontendOptions.strip())
elif isinstance(frontendOptions, list):
# Static list: should contain option dictionaries
if not frontendOptions:
return True # Empty list is valid (no options)
for option in frontendOptions:
if not isinstance(option, dict):
return False
if "value" not in option:
return False
if "label" not in option:
return False
if not isinstance(option["label"], dict):
return False
return True
else:
return False
def getOptionsName(frontendOptions: FrontendOptions) -> str:
"""
Get the options name from a string reference.
Args:
frontendOptions: The frontend_options value (must be a string reference)
Returns:
The options name (e.g., "user.role")
Raises:
ValueError: If frontendOptions is not a string reference
"""
if not isStringReference(frontendOptions):
raise ValueError(f"frontend_options is not a string reference: {type(frontendOptions)}")
return frontendOptions
def getStaticOptions(frontendOptions: FrontendOptions) -> List[OptionItem]:
"""
Get the static options list.
Args:
frontendOptions: The frontend_options value (must be a static list)
Returns:
The list of option items
Raises:
ValueError: If frontendOptions is not a static list
"""
if not isStaticList(frontendOptions):
raise ValueError(f"frontend_options is not a static list: {type(frontendOptions)}")
return frontendOptions

243
tests/integration/options/test_options_api.py
View file

@ -1,243 +0,0 @@
# Copyright (c) 2025 Patrick Motsch
# All rights reserved.
"""
Integration tests for Options API endpoints.
Tests the actual API endpoints with real database connections.
"""
import pytest
import secrets
from fastapi.testclient import TestClient
from modules.datamodels.datamodelUam import User
from modules.interfaces.interfaceDbApp import getRootInterface
@pytest.fixture
def app():
"""Create FastAPI app instance for testing."""
from app import app as fastapi_app
return fastapi_app
@pytest.fixture
def testClient(app):
"""Create test client for API testing."""
return TestClient(app)
@pytest.fixture
def csrfToken():
"""Generate a valid CSRF token for testing."""
# Generate a hex string between 16-64 characters (CSRF validation requirement)
return secrets.token_hex(16) # 32 character hex string
@pytest.fixture
def testUser() -> User:
"""Create a test user for API testing."""
# Use getRootInterface for system operations like user creation
# The root interface automatically uses the root mandate
rootInterface = getRootInterface()
user = rootInterface.createUser(
username="testuser_options",
email="testuser_options@example.com",
password="testpass123",
roleLabels=["user"]
)
return user
class TestOptionsAPI:
"""Test Options API endpoints."""
def testGetOptionsUserRole(self, testClient, testUser, csrfToken):
"""Test GET /api/options/user.role endpoint."""
# Get auth token (stored in cookie)
response = testClient.post(
"/api/local/login",
data={"username": testUser.username, "password": "testpass123"},
headers={"X-CSRF-Token": csrfToken}
)
assert response.status_code == 200
# Extract token from cookie for Bearer header
token = response.cookies.get("auth_token")
assert token is not None
# Get options
response = testClient.get(
"/api/options/user.role",
headers={"Authorization": f"Bearer {token}"}
)
assert response.status_code == 200
options = response.json()
assert isinstance(options, list)
assert len(options) >= 4 # At least sysadmin, admin, user, viewer
# Check structure
for option in options:
assert "value" in option
assert "label" in option
assert isinstance(option["label"], dict)
# Check specific values
values = [opt["value"] for opt in options]
assert "sysadmin" in values
assert "admin" in values
assert "user" in values
assert "viewer" in values
def testGetOptionsAuthAuthority(self, testClient, testUser, csrfToken):
"""Test GET /api/options/auth.authority endpoint."""
# Get auth token (stored in cookie)
response = testClient.post(
"/api/local/login",
data={"username": testUser.username, "password": "testpass123"},
headers={"X-CSRF-Token": csrfToken}
)
assert response.status_code == 200
# Extract token from cookie for Bearer header
token = response.cookies.get("auth_token")
assert token is not None
# Get options
response = testClient.get(
"/api/options/auth.authority",
headers={"Authorization": f"Bearer {token}"}
)
assert response.status_code == 200
options = response.json()
assert isinstance(options, list)
assert len(options) == 3 # local, google, msft
# Check structure
for option in options:
assert "value" in option
assert "label" in option
# Check specific values
values = [opt["value"] for opt in options]
assert "local" in values
assert "google" in values
assert "msft" in values
def testGetOptionsConnectionStatus(self, testClient, testUser, csrfToken):
"""Test GET /api/options/connection.status endpoint."""
# Get auth token (stored in cookie)
response = testClient.post(
"/api/local/login",
data={"username": testUser.username, "password": "testpass123"},
headers={"X-CSRF-Token": csrfToken}
)
assert response.status_code == 200
# Extract token from cookie for Bearer header
token = response.cookies.get("auth_token")
assert token is not None
# Get options
response = testClient.get(
"/api/options/connection.status",
headers={"Authorization": f"Bearer {token}"}
)
assert response.status_code == 200
options = response.json()
assert isinstance(options, list)
assert len(options) >= 4 # active, inactive, expired, pending, revoked, error
# Check structure
for option in options:
assert "value" in option
assert "label" in option
def testGetOptionsUserConnection(self, testClient, testUser, csrfToken):
"""Test GET /api/options/user.connection endpoint (context-aware)."""
# Get auth token (stored in cookie)
response = testClient.post(
"/api/local/login",
data={"username": testUser.username, "password": "testpass123"},
headers={"X-CSRF-Token": csrfToken}
)
assert response.status_code == 200
# Extract token from cookie for Bearer header
token = response.cookies.get("auth_token")
assert token is not None
# Get options (should return empty list if no connections)
response = testClient.get(
"/api/options/user.connection",
headers={"Authorization": f"Bearer {token}"}
)
assert response.status_code == 200
options = response.json()
# Should return a list (may be empty)
assert isinstance(options, list)
def testGetOptionsList(self, testClient, testUser, csrfToken):
"""Test GET /api/options/ endpoint (list all available options)."""
# Get auth token (stored in cookie)
response = testClient.post(
"/api/local/login",
data={"username": testUser.username, "password": "testpass123"},
headers={"X-CSRF-Token": csrfToken}
)
assert response.status_code == 200
# Extract token from cookie for Bearer header
token = response.cookies.get("auth_token")
assert token is not None
# Get available options names
response = testClient.get(
"/api/options/",
headers={"Authorization": f"Bearer {token}"}
)
assert response.status_code == 200
optionsNames = response.json()
assert isinstance(optionsNames, list)
assert "user.role" in optionsNames
assert "auth.authority" in optionsNames
assert "connection.status" in optionsNames
assert "user.connection" in optionsNames
def testGetOptionsUnknown(self, testClient, testUser, csrfToken):
"""Test GET /api/options/unknown.options endpoint (should return 400)."""
# Get auth token (stored in cookie)
response = testClient.post(
"/api/local/login",
data={"username": testUser.username, "password": "testpass123"},
headers={"X-CSRF-Token": csrfToken}
)
assert response.status_code == 200
# Extract token from cookie for Bearer header
token = response.cookies.get("auth_token")
assert token is not None
# Get unknown options (should return error)
response = testClient.get(
"/api/options/unknown.options",
headers={"Authorization": f"Bearer {token}"}
)
assert response.status_code == 400
def testGetOptionsUnauthorized(self, testClient):
"""Test GET /api/options/user.role without authentication."""
# Try to get options without auth token
response = testClient.get("/api/options/user.role")
# Should require authentication
assert response.status_code == 401