Spaces:
Sleeping
Sleeping
File size: 6,103 Bytes
4e4664a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 |
"""
Validation Utilities
Common validation functions to reduce duplicate code across routes
"""
from fastapi import HTTPException, status
from typing import Dict, Any
import repo
def validate_property_exists(db, property_id: str) -> Dict[str, Any]:
"""
Validate that a property exists and return it
Args:
db: Database connection
property_id: Property ID to validate
Returns:
Property dict if found
Raises:
HTTPException: 404 if property not found
Usage:
property_obj = validate_property_exists(db, property_id)
"""
property_obj = repo.get_property_by_id(db, property_id)
if not property_obj:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Property not found"
)
return property_obj
def validate_wallet_exists(db, user_id: str) -> Dict[str, Any]:
"""
Validate that a user's wallet exists and return it
Args:
db: Database connection
user_id: User ID to get wallet for
Returns:
Wallet dict if found
Raises:
HTTPException: 404 if wallet not found
Usage:
wallet = validate_wallet_exists(db, current_user['id'])
"""
wallet = repo.get_wallet_by_user(db, user_id)
if not wallet:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Wallet not found"
)
return wallet
def validate_user_exists(db, user_id: str) -> Dict[str, Any]:
"""
Validate that a user exists and return them
Args:
db: Database connection
user_id: User ID to validate
Returns:
User dict if found
Raises:
HTTPException: 404 if user not found
Usage:
user = validate_user_exists(db, user_id)
"""
user = repo.get_user_by_id(db, user_id)
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found"
)
return user
def validate_property_active(property_obj: Dict[str, Any]) -> None:
"""
Validate that a property is active
Args:
property_obj: Property dict to validate
Raises:
HTTPException: 400 if property not active
Usage:
validate_property_active(property_obj)
"""
if not property_obj.get('is_active', False):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Property is not active"
)
def validate_property_funded(property_obj: Dict[str, Any]) -> None:
"""
Validate that a property is funded
Args:
property_obj: Property dict to validate
Raises:
HTTPException: 400 if property not funded
Usage:
validate_property_funded(property_obj)
"""
if property_obj.get('status') != 'funded':
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Property is not funded. Current status: {property_obj.get('status', 'unknown')}"
)
def validate_tokens_available(property_obj: Dict[str, Any], requested_tokens: int) -> None:
"""
Validate that enough tokens are available for purchase
Args:
property_obj: Property dict to check
requested_tokens: Number of tokens requested
Raises:
HTTPException: 400 if insufficient tokens available
Usage:
validate_tokens_available(property_obj, buy_request.amount_tokens)
"""
available_tokens = property_obj.get('available_tokens', 0)
if available_tokens < requested_tokens:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Insufficient tokens available. Requested: {requested_tokens}, Available: {available_tokens}"
)
def validate_wallet_balance(wallet: Dict[str, Any], required_amount: float) -> None:
"""
Validate that wallet has sufficient balance
Args:
wallet: Wallet dict to check
required_amount: Amount required
Raises:
HTTPException: 400 if insufficient balance
Usage:
validate_wallet_balance(wallet, total_cost)
"""
current_balance = wallet.get('balance', 0.0)
if current_balance < required_amount:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Insufficient wallet balance. Required: {required_amount:.2f} {wallet.get('currency', 'AED')}, Available: {current_balance:.2f}"
)
def validate_kyc_approved(user: Dict[str, Any]) -> None:
"""
Validate that user's KYC is approved
Args:
user: User dict to check
Raises:
HTTPException: 403 if KYC not approved
Usage:
validate_kyc_approved(current_user)
"""
kyc_status = user.get('kyc_status', 'pending')
if kyc_status != 'approved':
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"KYC verification required. Your current status: {kyc_status}. Please complete KYC verification."
)
def validate_admin_role(user: Dict[str, Any]) -> None:
"""
Validate that user has admin role
Args:
user: User dict to check
Raises:
HTTPException: 403 if not admin
Usage:
validate_admin_role(current_user)
"""
if user.get('role') != 'admin':
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Admin access required"
)
def validate_super_admin_role(user: Dict[str, Any]) -> None:
"""
Validate that user has super_admin role
Args:
user: User dict to check
Raises:
HTTPException: 403 if not super admin
Usage:
validate_super_admin_role(current_user)
"""
if user.get('role') != 'super_admin':
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Super admin access required"
)
|