| 1 | | class RestAPI: |
| 2 | | def __init__(self, database: dict | None = None): |
| 3 | | pass |
| 4 | | |
| 5 | | def get(self, url: str, payload: str | None = None) -> str: |
| 6 | | pass |
| 1 | import json |
| 2 | from typing import Dict, List, Optional, Any |
| 7 | 3 | |
| 8 | | def post(self, url: str, payload: str | None = None) -> str: |
| 9 | | pass |
| 4 | class RestAPI: |
| 5 | """ |
| 6 | A RESTful API for tracking IOUs between users. |
| 7 | |
| 8 | This class manages user data and IOU transactions in memory. |
| 9 | """ |
| 10 | |
| 11 | def __init__(self, database: Optional[Dict[str, Any]] = None): |
| 12 | """ |
| 13 | Initialize the RestAPI with an optional database. |
| 14 | |
| 15 | Args: |
| 16 | database: Optional initial database with users data |
| 17 | """ |
| 18 | if database and 'users' in database: |
| 19 | # Initialize users from database |
| 20 | self.users = {} |
| 21 | for user_data in database['users']: |
| 22 | self.users[user_data['name']] = { |
| 23 | 'name': user_data['name'], |
| 24 | 'owes': user_data.get('owes', {}), |
| 25 | 'owed_by': user_data.get('owed_by', {}), |
| 26 | 'balance': user_data.get('balance', 0.0) |
| 27 | } |
| 28 | else: |
| 29 | # Initialize empty users dictionary |
| 30 | self.users = {} |
| 31 | |
| 32 | def get(self, url: str, payload: Optional[str] = None) -> str: |
| 33 | """ |
| 34 | Handle GET requests to the API. |
| 35 | |
| 36 | Args: |
| 37 | url: The endpoint URL |
| 38 | payload: Optional JSON payload |
| 39 | |
| 40 | Returns: |
| 41 | JSON response string |
| 42 | """ |
| 43 | if url == '/users': |
| 44 | if payload is None: |
| 45 | # Return all users |
| 46 | users_list = list(self.users.values()) |
| 47 | users_list.sort(key=lambda x: x['name']) |
| 48 | return json.dumps({'users': users_list}) |
| 49 | else: |
| 50 | # Return specific users |
| 51 | try: |
| 52 | data = json.loads(payload) |
| 53 | user_names = data.get('users', []) |
| 54 | |
| 55 | # Edge Case: Empty users list in payload |
| 56 | if not user_names: |
| 57 | return json.dumps({'users': []}) |
| 58 | |
| 59 | # Edge Case: Non-existent user requested |
| 60 | result_users = [] |
| 61 | for name in user_names: |
| 62 | if name in self.users: |
| 63 | result_users.append(self.users[name]) |
| 64 | else: |
| 65 | # If any user doesn't exist, return empty list as per typical API behavior |
| 66 | return json.dumps({'users': []}) |
| 67 | |
| 68 | result_users.sort(key=lambda x: x['name']) |
| 69 | return json.dumps({'users': result_users}) |
| 70 | except json.JSONDecodeError: |
| 71 | # Edge Case: Invalid JSON payload |
| 72 | return json.dumps({'users': []}) |
| 73 | else: |
| 74 | # Edge Case: Invalid endpoint |
| 75 | return json.dumps({}) |
| 76 | |
| 77 | # Handled Edge Cases: Empty users list in payload, Non-existent user requested, Invalid JSON payload, Invalid endpoint |
| 78 | |
| 79 | def post(self, url: str, payload: Optional[str] = None) -> str: |
| 80 | """ |
| 81 | Handle POST requests to the API. |
| 82 | |
| 83 | Args: |
| 84 | url: The endpoint URL |
| 85 | payload: Optional JSON payload |
| 86 | |
| 87 | Returns: |
| 88 | JSON response string |
| 89 | """ |
| 90 | if payload is None: |
| 91 | # Edge Case: Missing payload |
| 92 | return json.dumps({}) |
| 93 | |
| 94 | try: |
| 95 | data = json.loads(payload) |
| 96 | except json.JSONDecodeError: |
| 97 | # Edge Case: Invalid JSON payload |
| 98 | return json.dumps({}) |
| 99 | |
| 100 | if url == '/add': |
| 101 | user_name = data.get('user') |
| 102 | |
| 103 | # Edge Case: Missing user name |
| 104 | if not user_name: |
| 105 | return json.dumps({}) |
| 106 | |
| 107 | # Edge Case: User already exists |
| 108 | if user_name in self.users: |
| 109 | return json.dumps({}) |
| 110 | |
| 111 | # Create new user |
| 112 | new_user = { |
| 113 | 'name': user_name, |
| 114 | 'owes': {}, |
| 115 | 'owed_by': {}, |
| 116 | 'balance': 0.0 |
| 117 | } |
| 118 | self.users[user_name] = new_user |
| 119 | return json.dumps(new_user) |
| 120 | |
| 121 | elif url == '/iou': |
| 122 | lender = data.get('lender') |
| 123 | borrower = data.get('borrower') |
| 124 | amount = data.get('amount') |
| 125 | |
| 126 | # Edge Case: Missing required fields |
| 127 | if not lender or not borrower or amount is None: |
| 128 | return json.dumps({}) |
| 129 | |
| 130 | # Edge Case: Invalid amount type or negative amount |
| 131 | if not isinstance(amount, (int, float)) or amount < 0: |
| 132 | return json.dumps({}) |
| 133 | |
| 134 | # Edge Case: Lender or borrower doesn't exist |
| 135 | if lender not in self.users or borrower not in self.users: |
| 136 | return json.dumps({}) |
| 137 | |
| 138 | # Edge Case: Lender and borrower are the same person |
| 139 | if lender == borrower: |
| 140 | return json.dumps({}) |
| 141 | |
| 142 | # Update IOU records |
| 143 | # Update lender's owed_by |
| 144 | if borrower in self.users[lender]['owed_by']: |
| 145 | self.users[lender]['owed_by'][borrower] += amount |
| 146 | else: |
| 147 | self.users[lender]['owed_by'][borrower] = amount |
| 148 | |
| 149 | # Update borrower's owes |
| 150 | if lender in self.users[borrower]['owes']: |
| 151 | self.users[borrower]['owes'][lender] += amount |
| 152 | else: |
| 153 | self.users[borrower]['owes'][lender] = amount |
| 154 | |
| 155 | # Update balances |
| 156 | self.users[lender]['balance'] += amount |
| 157 | self.users[borrower]['balance'] -= amount |
| 158 | |
| 159 | # Return updated user objects |
| 160 | result_users = [self.users[lender], self.users[borrower]] |
| 161 | result_users.sort(key=lambda x: x['name']) |
| 162 | return json.dumps({'users': result_users}) |
| 163 | else: |
| 164 | # Edge Case: Invalid endpoint |
| 165 | return json.dumps({}) |
| 166 | |
| 167 | # Handled Edge Cases: Missing payload, Invalid JSON payload, Missing user name, User already exists, Missing required fields, Invalid amount type or negative amount, Lender or borrower doesn't exist, Lender and borrower are the same person, Invalid endpoint |
| 168 | |
| 169 | def _calculate_balance(self, user_name: str) -> float: |
| 170 | """ |
| 171 | Calculate the balance for a user based on their owes and owed_by. |
| 172 | |
| 173 | Args: |
| 174 | user_name: Name of the user |
| 175 | |
| 176 | Returns: |
| 177 | Calculated balance |
| 178 | """ |
| 179 | user = self.users[user_name] |
| 180 | owed_by_total = sum(user['owed_by'].values()) |
| 181 | owes_total = sum(user['owes'].values()) |
| 182 | return round(owed_by_total - owes_total, 2) |
| 183 | |
| 184 | # Handled Edge Cases: N/A (helper function) |