836 lines
30 KiB
Python
836 lines
30 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""
|
|
Interface to Trustee database.
|
|
Manages trustee organisations, roles, access, contracts, documents, and positions.
|
|
"""
|
|
|
|
import logging
|
|
import math
|
|
from typing import Dict, Any, List, Optional
|
|
|
|
from modules.connectors.connectorDbPostgre import DatabaseConnector
|
|
from modules.shared.configuration import APP_CONFIG
|
|
from modules.interfaces.interfaceRbac import getRecordsetWithRBAC
|
|
from modules.security.rbac import RbacClass
|
|
from modules.datamodels.datamodelUam import User, AccessLevel
|
|
from modules.datamodels.datamodelRbac import AccessRuleContext
|
|
from modules.datamodels.datamodelTrustee import (
|
|
TrusteeOrganisation,
|
|
TrusteeRole,
|
|
TrusteeAccess,
|
|
TrusteeContract,
|
|
TrusteeDocument,
|
|
TrusteePosition,
|
|
TrusteePositionDocument,
|
|
)
|
|
from modules.datamodels.datamodelPagination import PaginationParams, PaginatedResult
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Singleton factory for TrusteeObjects instances per context
|
|
_trusteeInterfaces = {}
|
|
|
|
|
|
def getInterface(currentUser: User) -> "TrusteeObjects":
|
|
"""Get or create a TrusteeObjects instance for the given user context."""
|
|
global _trusteeInterfaces
|
|
|
|
if not currentUser or not currentUser.id:
|
|
raise ValueError("Valid user context required")
|
|
|
|
cacheKey = f"{currentUser.id}_{currentUser.mandateId}"
|
|
|
|
if cacheKey not in _trusteeInterfaces:
|
|
_trusteeInterfaces[cacheKey] = TrusteeObjects(currentUser)
|
|
else:
|
|
# Update user context if needed
|
|
_trusteeInterfaces[cacheKey].setUserContext(currentUser)
|
|
|
|
return _trusteeInterfaces[cacheKey]
|
|
|
|
|
|
class TrusteeObjects:
|
|
"""
|
|
Interface to Trustee database.
|
|
Manages trustee organisations, roles, access, contracts, documents, and positions.
|
|
"""
|
|
|
|
def __init__(self, currentUser: Optional[User] = None):
|
|
"""Initializes the Trustee Interface."""
|
|
self.currentUser = currentUser
|
|
self.userId = currentUser.id if currentUser else None
|
|
self.mandateId = currentUser.mandateId if currentUser else None
|
|
self.rbac = None
|
|
|
|
# Initialize database
|
|
self._initializeDatabase()
|
|
|
|
# Set user context if provided
|
|
if currentUser:
|
|
self.setUserContext(currentUser)
|
|
|
|
def setUserContext(self, currentUser: User):
|
|
"""Sets the user context for the interface."""
|
|
if not currentUser:
|
|
logger.info("Initializing interface without user context")
|
|
return
|
|
|
|
self.currentUser = currentUser
|
|
self.userId = currentUser.id
|
|
self.mandateId = currentUser.mandateId
|
|
|
|
if not self.userId or not self.mandateId:
|
|
raise ValueError("Invalid user context: id and mandateId are required")
|
|
|
|
self.userLanguage = currentUser.language
|
|
|
|
# Initialize RBAC interface
|
|
from modules.security.rootAccess import getRootDbAppConnector
|
|
dbApp = getRootDbAppConnector()
|
|
self.rbac = RbacClass(self.db, dbApp=dbApp)
|
|
|
|
# Update database context
|
|
self.db.updateContext(self.userId)
|
|
|
|
def __del__(self):
|
|
"""Cleanup method to close database connection."""
|
|
if hasattr(self, "db") and self.db is not None:
|
|
try:
|
|
self.db.close()
|
|
except Exception as e:
|
|
logger.error(f"Error closing database connection: {e}")
|
|
|
|
def _initializeDatabase(self):
|
|
"""Initializes the database connection directly."""
|
|
try:
|
|
dbHost = APP_CONFIG.get("DB_TRUSTEE_HOST", APP_CONFIG.get("DB_CHAT_HOST", "_no_config_default_data"))
|
|
dbDatabase = APP_CONFIG.get("DB_TRUSTEE_DATABASE", "trustee")
|
|
dbUser = APP_CONFIG.get("DB_TRUSTEE_USER", APP_CONFIG.get("DB_CHAT_USER"))
|
|
dbPassword = APP_CONFIG.get("DB_TRUSTEE_PASSWORD_SECRET", APP_CONFIG.get("DB_CHAT_PASSWORD_SECRET"))
|
|
dbPort = int(APP_CONFIG.get("DB_TRUSTEE_PORT", APP_CONFIG.get("DB_CHAT_PORT", 5432)))
|
|
|
|
self.db = DatabaseConnector(
|
|
dbHost=dbHost,
|
|
dbDatabase=dbDatabase,
|
|
dbUser=dbUser,
|
|
dbPassword=dbPassword,
|
|
dbPort=dbPort,
|
|
userId=self.userId,
|
|
)
|
|
|
|
self.db.initDbSystem()
|
|
logger.info(f"Trustee database initialized successfully for user {self.userId}")
|
|
except Exception as e:
|
|
logger.error(f"Failed to initialize Trustee database: {str(e)}")
|
|
raise
|
|
|
|
def checkRbacPermission(
|
|
self,
|
|
modelClass: type,
|
|
operation: str,
|
|
recordId: Optional[str] = None
|
|
) -> bool:
|
|
"""Check RBAC permission for a specific operation on a table."""
|
|
if not self.rbac or not self.currentUser:
|
|
return False
|
|
|
|
tableName = modelClass.__name__
|
|
permissions = self.rbac.getUserPermissions(
|
|
self.currentUser,
|
|
AccessRuleContext.DATA,
|
|
tableName
|
|
)
|
|
|
|
if not permissions.view:
|
|
return False
|
|
|
|
permLevel = getattr(permissions, operation, AccessLevel.NONE)
|
|
if permLevel == AccessLevel.NONE:
|
|
return False
|
|
|
|
return True
|
|
|
|
# ===== Organisation CRUD =====
|
|
|
|
def createOrganisation(self, data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
"""Create a new organisation."""
|
|
if not self.checkRbacPermission(TrusteeOrganisation, "create"):
|
|
logger.warning(f"User {self.userId} lacks permission to create organisation")
|
|
return None
|
|
|
|
# Set mandateId from current user
|
|
data["mandateId"] = self.mandateId
|
|
|
|
# Validate ID format (alphanumeric, hyphens, underscores, 3-50 chars)
|
|
orgId = data.get("id", "")
|
|
if not orgId or len(orgId) < 3 or len(orgId) > 50:
|
|
logger.error(f"Invalid organisation ID length: {len(orgId)}")
|
|
return None
|
|
|
|
import re
|
|
if not re.match(r'^[a-zA-Z0-9_-]+$', orgId):
|
|
logger.error(f"Invalid organisation ID format: {orgId}")
|
|
return None
|
|
|
|
success = self.db.saveRecord(TrusteeOrganisation, orgId, data)
|
|
if success:
|
|
return self.db.getRecord(TrusteeOrganisation, orgId)
|
|
return None
|
|
|
|
def getOrganisation(self, orgId: str) -> Optional[Dict[str, Any]]:
|
|
"""Get a single organisation by ID."""
|
|
return self.db.getRecord(TrusteeOrganisation, orgId)
|
|
|
|
def getAllOrganisations(self, params: Optional[PaginationParams] = None) -> PaginatedResult:
|
|
"""Get all organisations with RBAC filtering."""
|
|
records = getRecordsetWithRBAC(
|
|
connector=self.db,
|
|
modelClass=TrusteeOrganisation,
|
|
currentUser=self.currentUser,
|
|
recordFilter=None,
|
|
orderBy="id"
|
|
)
|
|
|
|
# Apply pagination
|
|
totalItems = len(records)
|
|
if params:
|
|
pageSize = params.pageSize or 20
|
|
page = params.page or 1
|
|
startIdx = (page - 1) * pageSize
|
|
endIdx = startIdx + pageSize
|
|
items = records[startIdx:endIdx]
|
|
totalPages = math.ceil(totalItems / pageSize) if pageSize > 0 else 1
|
|
else:
|
|
items = records
|
|
totalPages = 1
|
|
page = 1
|
|
pageSize = totalItems
|
|
|
|
return PaginatedResult(
|
|
items=items,
|
|
totalItems=totalItems,
|
|
totalPages=totalPages
|
|
)
|
|
|
|
def updateOrganisation(self, orgId: str, data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
"""Update an organisation."""
|
|
if not self.checkRbacPermission(TrusteeOrganisation, "update"):
|
|
logger.warning(f"User {self.userId} lacks permission to update organisation")
|
|
return None
|
|
|
|
# ID cannot be changed after creation
|
|
if "id" in data and data["id"] != orgId:
|
|
logger.error("Organisation ID cannot be changed")
|
|
return None
|
|
|
|
data["id"] = orgId
|
|
success = self.db.saveRecord(TrusteeOrganisation, orgId, data)
|
|
if success:
|
|
return self.db.getRecord(TrusteeOrganisation, orgId)
|
|
return None
|
|
|
|
def deleteOrganisation(self, orgId: str) -> bool:
|
|
"""Delete an organisation."""
|
|
if not self.checkRbacPermission(TrusteeOrganisation, "delete"):
|
|
logger.warning(f"User {self.userId} lacks permission to delete organisation")
|
|
return False
|
|
|
|
return self.db.deleteRecord(TrusteeOrganisation, orgId)
|
|
|
|
# ===== Role CRUD =====
|
|
|
|
def createRole(self, data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
"""Create a new role (sysadmin only)."""
|
|
if not self.checkRbacPermission(TrusteeRole, "create"):
|
|
logger.warning(f"User {self.userId} lacks permission to create role")
|
|
return None
|
|
|
|
data["mandateId"] = self.mandateId
|
|
roleId = data.get("id", "")
|
|
|
|
if not roleId:
|
|
logger.error("Role ID is required")
|
|
return None
|
|
|
|
success = self.db.saveRecord(TrusteeRole, roleId, data)
|
|
if success:
|
|
return self.db.getRecord(TrusteeRole, roleId)
|
|
return None
|
|
|
|
def getRole(self, roleId: str) -> Optional[Dict[str, Any]]:
|
|
"""Get a single role by ID."""
|
|
return self.db.getRecord(TrusteeRole, roleId)
|
|
|
|
def getAllRoles(self, params: Optional[PaginationParams] = None) -> PaginatedResult:
|
|
"""Get all roles with RBAC filtering."""
|
|
records = getRecordsetWithRBAC(
|
|
connector=self.db,
|
|
modelClass=TrusteeRole,
|
|
currentUser=self.currentUser,
|
|
recordFilter=None,
|
|
orderBy="id"
|
|
)
|
|
|
|
totalItems = len(records)
|
|
if params:
|
|
pageSize = params.pageSize or 20
|
|
page = params.page or 1
|
|
startIdx = (page - 1) * pageSize
|
|
endIdx = startIdx + pageSize
|
|
items = records[startIdx:endIdx]
|
|
totalPages = math.ceil(totalItems / pageSize) if pageSize > 0 else 1
|
|
else:
|
|
items = records
|
|
totalPages = 1
|
|
page = 1
|
|
pageSize = totalItems
|
|
|
|
return PaginatedResult(
|
|
items=items,
|
|
totalItems=totalItems,
|
|
totalPages=totalPages
|
|
)
|
|
|
|
def updateRole(self, roleId: str, data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
"""Update a role (sysadmin only)."""
|
|
if not self.checkRbacPermission(TrusteeRole, "update"):
|
|
logger.warning(f"User {self.userId} lacks permission to update role")
|
|
return None
|
|
|
|
data["id"] = roleId
|
|
success = self.db.saveRecord(TrusteeRole, roleId, data)
|
|
if success:
|
|
return self.db.getRecord(TrusteeRole, roleId)
|
|
return None
|
|
|
|
def deleteRole(self, roleId: str) -> bool:
|
|
"""Delete a role (sysadmin only, not if in use)."""
|
|
if not self.checkRbacPermission(TrusteeRole, "delete"):
|
|
logger.warning(f"User {self.userId} lacks permission to delete role")
|
|
return False
|
|
|
|
# Check if role is in use
|
|
accessRecords = self.db.getRecordset(TrusteeAccess, {"roleId": roleId})
|
|
if accessRecords:
|
|
logger.error(f"Cannot delete role {roleId}: still in use")
|
|
return False
|
|
|
|
return self.db.deleteRecord(TrusteeRole, roleId)
|
|
|
|
# ===== Access CRUD =====
|
|
|
|
def createAccess(self, data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
"""Create a new access record."""
|
|
if not self.checkRbacPermission(TrusteeAccess, "create"):
|
|
logger.warning(f"User {self.userId} lacks permission to create access")
|
|
return None
|
|
|
|
data["mandateId"] = self.mandateId
|
|
|
|
import uuid
|
|
accessId = data.get("id") or str(uuid.uuid4())
|
|
data["id"] = accessId
|
|
|
|
success = self.db.saveRecord(TrusteeAccess, accessId, data)
|
|
if success:
|
|
return self.db.getRecord(TrusteeAccess, accessId)
|
|
return None
|
|
|
|
def getAccess(self, accessId: str) -> Optional[Dict[str, Any]]:
|
|
"""Get a single access record by ID."""
|
|
return self.db.getRecord(TrusteeAccess, accessId)
|
|
|
|
def getAllAccess(self, params: Optional[PaginationParams] = None) -> PaginatedResult:
|
|
"""Get all access records with RBAC filtering."""
|
|
records = getRecordsetWithRBAC(
|
|
connector=self.db,
|
|
modelClass=TrusteeAccess,
|
|
currentUser=self.currentUser,
|
|
recordFilter=None,
|
|
orderBy="id"
|
|
)
|
|
|
|
totalItems = len(records)
|
|
if params:
|
|
pageSize = params.pageSize or 20
|
|
page = params.page or 1
|
|
startIdx = (page - 1) * pageSize
|
|
endIdx = startIdx + pageSize
|
|
items = records[startIdx:endIdx]
|
|
totalPages = math.ceil(totalItems / pageSize) if pageSize > 0 else 1
|
|
else:
|
|
items = records
|
|
totalPages = 1
|
|
page = 1
|
|
pageSize = totalItems
|
|
|
|
return PaginatedResult(
|
|
items=items,
|
|
totalItems=totalItems,
|
|
totalPages=totalPages
|
|
)
|
|
|
|
def getAccessByOrganisation(self, organisationId: str) -> List[Dict[str, Any]]:
|
|
"""Get all access records for a specific organisation."""
|
|
return getRecordsetWithRBAC(
|
|
connector=self.db,
|
|
modelClass=TrusteeAccess,
|
|
currentUser=self.currentUser,
|
|
recordFilter={"organisationId": organisationId},
|
|
orderBy="id"
|
|
)
|
|
|
|
def getAccessByUser(self, userId: str) -> List[Dict[str, Any]]:
|
|
"""Get all access records for a specific user."""
|
|
return getRecordsetWithRBAC(
|
|
connector=self.db,
|
|
modelClass=TrusteeAccess,
|
|
currentUser=self.currentUser,
|
|
recordFilter={"userId": userId},
|
|
orderBy="id"
|
|
)
|
|
|
|
def updateAccess(self, accessId: str, data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
"""Update an access record."""
|
|
if not self.checkRbacPermission(TrusteeAccess, "update"):
|
|
logger.warning(f"User {self.userId} lacks permission to update access")
|
|
return None
|
|
|
|
data["id"] = accessId
|
|
success = self.db.saveRecord(TrusteeAccess, accessId, data)
|
|
if success:
|
|
return self.db.getRecord(TrusteeAccess, accessId)
|
|
return None
|
|
|
|
def deleteAccess(self, accessId: str) -> bool:
|
|
"""Delete an access record."""
|
|
if not self.checkRbacPermission(TrusteeAccess, "delete"):
|
|
logger.warning(f"User {self.userId} lacks permission to delete access")
|
|
return False
|
|
|
|
return self.db.deleteRecord(TrusteeAccess, accessId)
|
|
|
|
# ===== Contract CRUD =====
|
|
|
|
def createContract(self, data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
"""Create a new contract."""
|
|
if not self.checkRbacPermission(TrusteeContract, "create"):
|
|
logger.warning(f"User {self.userId} lacks permission to create contract")
|
|
return None
|
|
|
|
data["mandateId"] = self.mandateId
|
|
|
|
import uuid
|
|
contractId = data.get("id") or str(uuid.uuid4())
|
|
data["id"] = contractId
|
|
|
|
success = self.db.saveRecord(TrusteeContract, contractId, data)
|
|
if success:
|
|
return self.db.getRecord(TrusteeContract, contractId)
|
|
return None
|
|
|
|
def getContract(self, contractId: str) -> Optional[Dict[str, Any]]:
|
|
"""Get a single contract by ID."""
|
|
return self.db.getRecord(TrusteeContract, contractId)
|
|
|
|
def getAllContracts(self, params: Optional[PaginationParams] = None) -> PaginatedResult:
|
|
"""Get all contracts with RBAC filtering."""
|
|
records = getRecordsetWithRBAC(
|
|
connector=self.db,
|
|
modelClass=TrusteeContract,
|
|
currentUser=self.currentUser,
|
|
recordFilter=None,
|
|
orderBy="id"
|
|
)
|
|
|
|
totalItems = len(records)
|
|
if params:
|
|
pageSize = params.pageSize or 20
|
|
page = params.page or 1
|
|
startIdx = (page - 1) * pageSize
|
|
endIdx = startIdx + pageSize
|
|
items = records[startIdx:endIdx]
|
|
totalPages = math.ceil(totalItems / pageSize) if pageSize > 0 else 1
|
|
else:
|
|
items = records
|
|
totalPages = 1
|
|
page = 1
|
|
pageSize = totalItems
|
|
|
|
return PaginatedResult(
|
|
items=items,
|
|
totalItems=totalItems,
|
|
totalPages=totalPages
|
|
)
|
|
|
|
def getContractsByOrganisation(self, organisationId: str) -> List[Dict[str, Any]]:
|
|
"""Get all contracts for a specific organisation."""
|
|
return getRecordsetWithRBAC(
|
|
connector=self.db,
|
|
modelClass=TrusteeContract,
|
|
currentUser=self.currentUser,
|
|
recordFilter={"organisationId": organisationId},
|
|
orderBy="label"
|
|
)
|
|
|
|
def updateContract(self, contractId: str, data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
"""Update a contract (organisationId is immutable)."""
|
|
if not self.checkRbacPermission(TrusteeContract, "update"):
|
|
logger.warning(f"User {self.userId} lacks permission to update contract")
|
|
return None
|
|
|
|
# Check if organisationId is being changed
|
|
existing = self.db.getRecord(TrusteeContract, contractId)
|
|
if existing and "organisationId" in data:
|
|
if data["organisationId"] != existing.get("organisationId"):
|
|
logger.error("Contract organisationId cannot be changed after creation")
|
|
return None
|
|
|
|
data["id"] = contractId
|
|
success = self.db.saveRecord(TrusteeContract, contractId, data)
|
|
if success:
|
|
return self.db.getRecord(TrusteeContract, contractId)
|
|
return None
|
|
|
|
def deleteContract(self, contractId: str) -> bool:
|
|
"""Delete a contract."""
|
|
if not self.checkRbacPermission(TrusteeContract, "delete"):
|
|
logger.warning(f"User {self.userId} lacks permission to delete contract")
|
|
return False
|
|
|
|
return self.db.deleteRecord(TrusteeContract, contractId)
|
|
|
|
# ===== Document CRUD =====
|
|
|
|
def createDocument(self, data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
"""Create a new document."""
|
|
if not self.checkRbacPermission(TrusteeDocument, "create"):
|
|
logger.warning(f"User {self.userId} lacks permission to create document")
|
|
return None
|
|
|
|
data["mandateId"] = self.mandateId
|
|
|
|
import uuid
|
|
documentId = data.get("id") or str(uuid.uuid4())
|
|
data["id"] = documentId
|
|
|
|
success = self.db.saveRecord(TrusteeDocument, documentId, data)
|
|
if success:
|
|
return self.db.getRecord(TrusteeDocument, documentId)
|
|
return None
|
|
|
|
def getDocument(self, documentId: str) -> Optional[Dict[str, Any]]:
|
|
"""Get a single document by ID (metadata only)."""
|
|
record = self.db.getRecord(TrusteeDocument, documentId)
|
|
if record:
|
|
# Remove binary data from response
|
|
record.pop("documentData", None)
|
|
return record
|
|
|
|
def getDocumentData(self, documentId: str) -> Optional[bytes]:
|
|
"""Get document binary data."""
|
|
record = self.db.getRecord(TrusteeDocument, documentId)
|
|
if record:
|
|
return record.get("documentData")
|
|
return None
|
|
|
|
def getAllDocuments(self, params: Optional[PaginationParams] = None) -> PaginatedResult:
|
|
"""Get all documents with RBAC filtering (metadata only)."""
|
|
records = getRecordsetWithRBAC(
|
|
connector=self.db,
|
|
modelClass=TrusteeDocument,
|
|
currentUser=self.currentUser,
|
|
recordFilter=None,
|
|
orderBy="documentName"
|
|
)
|
|
|
|
# Remove binary data from responses
|
|
for record in records:
|
|
record.pop("documentData", None)
|
|
|
|
totalItems = len(records)
|
|
if params:
|
|
pageSize = params.pageSize or 20
|
|
page = params.page or 1
|
|
startIdx = (page - 1) * pageSize
|
|
endIdx = startIdx + pageSize
|
|
items = records[startIdx:endIdx]
|
|
totalPages = math.ceil(totalItems / pageSize) if pageSize > 0 else 1
|
|
else:
|
|
items = records
|
|
totalPages = 1
|
|
page = 1
|
|
pageSize = totalItems
|
|
|
|
return PaginatedResult(
|
|
items=items,
|
|
totalItems=totalItems,
|
|
totalPages=totalPages
|
|
)
|
|
|
|
def getDocumentsByContract(self, contractId: str) -> List[Dict[str, Any]]:
|
|
"""Get all documents for a specific contract."""
|
|
records = getRecordsetWithRBAC(
|
|
connector=self.db,
|
|
modelClass=TrusteeDocument,
|
|
currentUser=self.currentUser,
|
|
recordFilter={"contractId": contractId},
|
|
orderBy="documentName"
|
|
)
|
|
for record in records:
|
|
record.pop("documentData", None)
|
|
return records
|
|
|
|
def updateDocument(self, documentId: str, data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
"""Update a document."""
|
|
if not self.checkRbacPermission(TrusteeDocument, "update"):
|
|
logger.warning(f"User {self.userId} lacks permission to update document")
|
|
return None
|
|
|
|
data["id"] = documentId
|
|
success = self.db.saveRecord(TrusteeDocument, documentId, data)
|
|
if success:
|
|
result = self.db.getRecord(TrusteeDocument, documentId)
|
|
if result:
|
|
result.pop("documentData", None)
|
|
return result
|
|
return None
|
|
|
|
def deleteDocument(self, documentId: str) -> bool:
|
|
"""Delete a document."""
|
|
if not self.checkRbacPermission(TrusteeDocument, "delete"):
|
|
logger.warning(f"User {self.userId} lacks permission to delete document")
|
|
return False
|
|
|
|
return self.db.deleteRecord(TrusteeDocument, documentId)
|
|
|
|
# ===== Position CRUD =====
|
|
|
|
def createPosition(self, data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
"""Create a new position."""
|
|
if not self.checkRbacPermission(TrusteePosition, "create"):
|
|
logger.warning(f"User {self.userId} lacks permission to create position")
|
|
return None
|
|
|
|
data["mandateId"] = self.mandateId
|
|
|
|
# Calculate VAT amount if not provided
|
|
if "vatAmount" not in data or data.get("vatAmount") == 0:
|
|
bookingAmount = data.get("bookingAmount", 0)
|
|
vatPercentage = data.get("vatPercentage", 0)
|
|
data["vatAmount"] = bookingAmount * vatPercentage / 100
|
|
|
|
import uuid
|
|
positionId = data.get("id") or str(uuid.uuid4())
|
|
data["id"] = positionId
|
|
|
|
success = self.db.saveRecord(TrusteePosition, positionId, data)
|
|
if success:
|
|
return self.db.getRecord(TrusteePosition, positionId)
|
|
return None
|
|
|
|
def getPosition(self, positionId: str) -> Optional[Dict[str, Any]]:
|
|
"""Get a single position by ID."""
|
|
return self.db.getRecord(TrusteePosition, positionId)
|
|
|
|
def getAllPositions(self, params: Optional[PaginationParams] = None) -> PaginatedResult:
|
|
"""Get all positions with RBAC filtering."""
|
|
records = getRecordsetWithRBAC(
|
|
connector=self.db,
|
|
modelClass=TrusteePosition,
|
|
currentUser=self.currentUser,
|
|
recordFilter=None,
|
|
orderBy="valuta"
|
|
)
|
|
|
|
totalItems = len(records)
|
|
if params:
|
|
pageSize = params.pageSize or 20
|
|
page = params.page or 1
|
|
startIdx = (page - 1) * pageSize
|
|
endIdx = startIdx + pageSize
|
|
items = records[startIdx:endIdx]
|
|
totalPages = math.ceil(totalItems / pageSize) if pageSize > 0 else 1
|
|
else:
|
|
items = records
|
|
totalPages = 1
|
|
page = 1
|
|
pageSize = totalItems
|
|
|
|
return PaginatedResult(
|
|
items=items,
|
|
totalItems=totalItems,
|
|
totalPages=totalPages
|
|
)
|
|
|
|
def getPositionsByContract(self, contractId: str) -> List[Dict[str, Any]]:
|
|
"""Get all positions for a specific contract."""
|
|
return getRecordsetWithRBAC(
|
|
connector=self.db,
|
|
modelClass=TrusteePosition,
|
|
currentUser=self.currentUser,
|
|
recordFilter={"contractId": contractId},
|
|
orderBy="valuta"
|
|
)
|
|
|
|
def getPositionsByOrganisation(self, organisationId: str) -> List[Dict[str, Any]]:
|
|
"""Get all positions for a specific organisation."""
|
|
return getRecordsetWithRBAC(
|
|
connector=self.db,
|
|
modelClass=TrusteePosition,
|
|
currentUser=self.currentUser,
|
|
recordFilter={"organisationId": organisationId},
|
|
orderBy="valuta"
|
|
)
|
|
|
|
def updatePosition(self, positionId: str, data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
"""Update a position."""
|
|
if not self.checkRbacPermission(TrusteePosition, "update"):
|
|
logger.warning(f"User {self.userId} lacks permission to update position")
|
|
return None
|
|
|
|
data["id"] = positionId
|
|
success = self.db.saveRecord(TrusteePosition, positionId, data)
|
|
if success:
|
|
return self.db.getRecord(TrusteePosition, positionId)
|
|
return None
|
|
|
|
def deletePosition(self, positionId: str) -> bool:
|
|
"""Delete a position."""
|
|
if not self.checkRbacPermission(TrusteePosition, "delete"):
|
|
logger.warning(f"User {self.userId} lacks permission to delete position")
|
|
return False
|
|
|
|
return self.db.deleteRecord(TrusteePosition, positionId)
|
|
|
|
# ===== Position-Document Link CRUD =====
|
|
|
|
def createPositionDocument(self, data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
"""Create a new position-document link."""
|
|
if not self.checkRbacPermission(TrusteePositionDocument, "create"):
|
|
logger.warning(f"User {self.userId} lacks permission to create position-document link")
|
|
return None
|
|
|
|
data["mandateId"] = self.mandateId
|
|
|
|
import uuid
|
|
linkId = data.get("id") or str(uuid.uuid4())
|
|
data["id"] = linkId
|
|
|
|
success = self.db.saveRecord(TrusteePositionDocument, linkId, data)
|
|
if success:
|
|
return self.db.getRecord(TrusteePositionDocument, linkId)
|
|
return None
|
|
|
|
def getPositionDocument(self, linkId: str) -> Optional[Dict[str, Any]]:
|
|
"""Get a single position-document link by ID."""
|
|
return self.db.getRecord(TrusteePositionDocument, linkId)
|
|
|
|
def getAllPositionDocuments(self, params: Optional[PaginationParams] = None) -> PaginatedResult:
|
|
"""Get all position-document links with RBAC filtering."""
|
|
records = getRecordsetWithRBAC(
|
|
connector=self.db,
|
|
modelClass=TrusteePositionDocument,
|
|
currentUser=self.currentUser,
|
|
recordFilter=None,
|
|
orderBy="id"
|
|
)
|
|
|
|
totalItems = len(records)
|
|
if params:
|
|
pageSize = params.pageSize or 20
|
|
page = params.page or 1
|
|
startIdx = (page - 1) * pageSize
|
|
endIdx = startIdx + pageSize
|
|
items = records[startIdx:endIdx]
|
|
totalPages = math.ceil(totalItems / pageSize) if pageSize > 0 else 1
|
|
else:
|
|
items = records
|
|
totalPages = 1
|
|
page = 1
|
|
pageSize = totalItems
|
|
|
|
return PaginatedResult(
|
|
items=items,
|
|
totalItems=totalItems,
|
|
totalPages=totalPages
|
|
)
|
|
|
|
def getDocumentsForPosition(self, positionId: str) -> List[Dict[str, Any]]:
|
|
"""Get all documents linked to a position."""
|
|
links = getRecordsetWithRBAC(
|
|
connector=self.db,
|
|
modelClass=TrusteePositionDocument,
|
|
currentUser=self.currentUser,
|
|
recordFilter={"positionId": positionId},
|
|
orderBy="id"
|
|
)
|
|
return links
|
|
|
|
def getPositionsForDocument(self, documentId: str) -> List[Dict[str, Any]]:
|
|
"""Get all positions linked to a document."""
|
|
links = getRecordsetWithRBAC(
|
|
connector=self.db,
|
|
modelClass=TrusteePositionDocument,
|
|
currentUser=self.currentUser,
|
|
recordFilter={"documentId": documentId},
|
|
orderBy="id"
|
|
)
|
|
return links
|
|
|
|
def deletePositionDocument(self, linkId: str) -> bool:
|
|
"""Delete a position-document link."""
|
|
if not self.checkRbacPermission(TrusteePositionDocument, "delete"):
|
|
logger.warning(f"User {self.userId} lacks permission to delete position-document link")
|
|
return False
|
|
|
|
return self.db.deleteRecord(TrusteePositionDocument, linkId)
|
|
|
|
# ===== Trustee-specific Access Check =====
|
|
|
|
def getUserAccessForOrganisation(self, userId: str, organisationId: str) -> List[Dict[str, Any]]:
|
|
"""Get all access records for a user in a specific organisation."""
|
|
return self.db.getRecordset(
|
|
TrusteeAccess,
|
|
{"userId": userId, "organisationId": organisationId}
|
|
)
|
|
|
|
def checkUserTrusteePermission(
|
|
self,
|
|
userId: str,
|
|
organisationId: str,
|
|
requiredRole: str,
|
|
contractId: Optional[str] = None
|
|
) -> bool:
|
|
"""
|
|
Check if a user has a specific role for an organisation (and optionally contract).
|
|
|
|
Args:
|
|
userId: User ID to check
|
|
organisationId: Organisation ID
|
|
requiredRole: Required role (userreport, admin, operate)
|
|
contractId: Optional contract ID for contract-specific access
|
|
|
|
Returns:
|
|
True if user has the required role
|
|
"""
|
|
accessRecords = self.getUserAccessForOrganisation(userId, organisationId)
|
|
|
|
for access in accessRecords:
|
|
if access.get("roleId") == requiredRole:
|
|
accessContractId = access.get("contractId")
|
|
|
|
# If access has no contractId, it grants access to all contracts
|
|
if accessContractId is None:
|
|
return True
|
|
|
|
# If checking for specific contract, match it
|
|
if contractId and accessContractId == contractId:
|
|
return True
|
|
|
|
# If no specific contract requested and access is contract-specific, deny
|
|
if contractId is None:
|
|
continue
|
|
|
|
return False
|