0
# Exception Handling
1

2
Exception classes for handling HTTP and WebSocket errors with proper status codes and error messages.
3

4
## Capabilities
5

6
### HTTP Exceptions
7

8
HTTP exception class for handling web application errors with appropriate status codes.
9

10
```python { .api }
11
class HTTPException(Exception):
12
    def __init__(self, status_code: int, detail: str | None = None) -> None:
13
        """
14
        HTTP exception with status code and detail message.
15
        
16
        Args:
17
            status_code: HTTP status code (e.g., 404, 500)
18
            detail: Optional error detail message (defaults to standard HTTP phrase)
19
        """
20
    
21
    def __str__(self) -> str:
22
        """String representation of the exception."""
23
    
24
    def __repr__(self) -> str:
25
        """Detailed string representation for debugging."""
26
```
27

28
### WebSocket Exceptions
29

30
WebSocket exception class for handling WebSocket connection errors.
31

32
```python { .api }
33
class WebSocketException(Exception):
34
    def __init__(self, code: int, reason: str | None = None) -> None:
35
        """
36
        WebSocket exception with close code and reason.
37
        
38
        Args:
39
            code: WebSocket close code (e.g., 1000, 1001, 1002)
40
            reason: Optional close reason message
41
        """
42
    
43
    def __str__(self) -> str:
44
        """String representation of the exception."""
45
    
46
    def __repr__(self) -> str:
47
        """Detailed string representation for debugging."""
48
```
49

50
## Usage Examples
51

52
### HTTP Exception Handling
53

54
```python
55
from robyn import Robyn, status_codes
56
from robyn.exceptions import HTTPException
57

58
app = Robyn(__file__)
59

60
@app.get("/user/<user_id>")
61
def get_user(request):
62
    user_id = request.path_params["user_id"]
63
    
64
    # Validate user ID
65
    if not user_id.isdigit():
66
        raise HTTPException(
67
            status_code=status_codes.HTTP_400_BAD_REQUEST,
68
            detail="User ID must be a number"
69
        )
70
    
71
    user_id = int(user_id)
72
    
73
    # Check if user exists (example)
74
    if user_id < 1 or user_id > 1000:
75
        raise HTTPException(
76
            status_code=status_codes.HTTP_404_NOT_FOUND,
77
            detail=f"User {user_id} not found"
78
        )
79
    
80
    # Simulate authorization check
81
    auth_header = request.headers.get("Authorization")
82
    if not auth_header:
83
        raise HTTPException(
84
            status_code=status_codes.HTTP_401_UNAUTHORIZED,
85
            detail="Authentication required"
86
        )
87
    
88
    return {"user_id": user_id, "name": f"User {user_id}"}
89

90
@app.get("/admin/users")
91
def admin_users(request):
92
    # Check admin permissions
93
    user_role = request.headers.get("X-User-Role")
94
    if user_role != "admin":
95
        raise HTTPException(
96
            status_code=status_codes.HTTP_403_FORBIDDEN,
97
            detail="Admin access required"
98
        )
99
    
100
    return {"users": ["alice", "bob", "charlie"]}
101

102
app.start()
103
```
104

105
### Global Exception Handler
106

107
```python
108
from robyn import Robyn, Response, status_codes
109
from robyn.exceptions import HTTPException
110

111
app = Robyn(__file__)
112

113
@app.exception
114
def handle_exception(request, exception):
115
    """Global exception handler for all unhandled exceptions."""
116
    if isinstance(exception, HTTPException):
117
        return Response(
118
            status_code=exception.status_code,
119
            headers={"Content-Type": "application/json"},
120
            description=f'{{"error": "{exception.detail}"}}'
121
        )
122
    else:
123
        # Handle unexpected exceptions
124
        return Response(
125
            status_code=status_codes.HTTP_500_INTERNAL_SERVER_ERROR,
126
            headers={"Content-Type": "application/json"},
127
            description='{"error": "Internal server error"}'
128
        )
129

130
@app.get("/error-demo/<error_type>")
131
def error_demo(request):
132
    error_type = request.path_params["error_type"]
133
    
134
    if error_type == "not_found":
135
        raise HTTPException(404, "Resource not found")
136
    elif error_type == "bad_request":
137
        raise HTTPException(400, "Invalid request parameters")
138
    elif error_type == "unauthorized":
139
        raise HTTPException(401)  # Uses default message
140
    elif error_type == "server_error":
141
        # This will trigger the generic exception handler
142
        raise ValueError("Something went wrong")
143
    
144
    return {"message": "No error triggered"}
145

146
app.start()
147
```
148

149
### WebSocket Exception Handling
150

151
```python
152
from robyn import Robyn
153
from robyn.ws import WebSocket
154
from robyn.exceptions import WebSocketException
155

156
app = Robyn(__file__)
157
websocket = WebSocket(app, "/ws")
158

159
@websocket.on("connect")
160
def on_connect(websocket_connector):
161
    # Validate connection (example)
162
    auth_token = websocket_connector.query_params.get("token")
163
    if not auth_token:
164
        raise WebSocketException(
165
            code=1008,  # Policy violation
166
            reason="Authentication token required"
167
        )
168
    
169
    print(f"Client {websocket_connector.id} connected")
170

171
@websocket.on("message")
172
def on_message(websocket_connector, message):
173
    try:
174
        # Process message
175
        if message == "close":
176
            raise WebSocketException(
177
                code=1000,  # Normal closure
178
                reason="Client requested close"
179
            )
180
        
181
        # Echo message back
182
        websocket_connector.sync_send_to(websocket_connector.id, f"Echo: {message}")
183
        
184
    except Exception as e:
185
        # Handle processing errors
186
        raise WebSocketException(
187
            code=1011,  # Unexpected condition
188
            reason=f"Message processing error: {str(e)}"
189
        )
190

191
@websocket.on("close")
192
def on_close(websocket_connector):
193
    print(f"Client {websocket_connector.id} disconnected")
194

195
app.add_web_socket("/ws", websocket)
196
app.start()
197
```
198

199
### Custom Exception Classes
200

201
```python
202
from robyn import Robyn, status_codes
203
from robyn.exceptions import HTTPException
204

205
# Custom business logic exceptions
206
class UserNotFoundError(HTTPException):
207
    def __init__(self, user_id: int):
208
        super().__init__(
209
            status_code=status_codes.HTTP_404_NOT_FOUND,
210
            detail=f"User with ID {user_id} not found"
211
        )
212

213
class InsufficientPermissionsError(HTTPException):
214
    def __init__(self, required_role: str):
215
        super().__init__(
216
            status_code=status_codes.HTTP_403_FORBIDDEN,
217
            detail=f"Requires {required_role} role"
218
        )
219

220
class ValidationError(HTTPException):
221
    def __init__(self, field: str, message: str):
222
        super().__init__(
223
            status_code=status_codes.HTTP_422_UNPROCESSABLE_ENTITY,
224
            detail=f"Validation error in {field}: {message}"
225
        )
226

227
app = Robyn(__file__)
228

229
@app.post("/users")
230
def create_user(request):
231
    user_data = request.json()
232
    
233
    # Validation
234
    if not user_data.get("email"):
235
        raise ValidationError("email", "Email is required")
236
    
237
    if "@" not in user_data["email"]:
238
        raise ValidationError("email", "Invalid email format")
239
    
240
    # Create user logic here
241
    return {"message": "User created", "id": 123}
242

243
@app.get("/user/<user_id>/profile")
244
def get_user_profile(request):
245
    user_id = int(request.path_params["user_id"])
246
    
247
    # Check if user exists
248
    if user_id > 1000:  # Example check
249
        raise UserNotFoundError(user_id)
250
    
251
    # Check permissions
252
    user_role = request.headers.get("X-User-Role")
253
    if user_role not in ["admin", "user"]:
254
        raise InsufficientPermissionsError("user")
255
    
256
    return {"user_id": user_id, "profile": "User profile data"}
257

258
app.start()
259
```
260

261
### Exception Logging and Monitoring
262

263
```python
264
import logging
265
from robyn import Robyn, Response, status_codes
266
from robyn.exceptions import HTTPException, WebSocketException
267

268
# Configure logging
269
logging.basicConfig(level=logging.INFO)
270
logger = logging.getLogger(__name__)
271

272
app = Robyn(__file__)
273

274
@app.exception
275
def log_and_handle_exception(request, exception):
276
    """Exception handler with logging for monitoring."""
277
    
278
    if isinstance(exception, HTTPException):
279
        # Log HTTP exceptions
280
        logger.warning(
281
            f"HTTP Exception: {exception.status_code} - {exception.detail} "
282
            f"[{request.method} {request.url.path}]"
283
        )
284
        
285
        return Response(
286
            status_code=exception.status_code,
287
            headers={"Content-Type": "application/json"},
288
            description=f'{{"error": "{exception.detail}"}}'
289
        )
290
    
291
    else:
292
        # Log unexpected exceptions
293
        logger.error(
294
            f"Unhandled Exception: {type(exception).__name__} - {str(exception)} "
295
            f"[{request.method} {request.url.path}]",
296
            exc_info=True
297
        )
298
        
299
        return Response(
300
            status_code=status_codes.HTTP_500_INTERNAL_SERVER_ERROR,
301
            headers={"Content-Type": "application/json"},
302
            description='{"error": "Internal server error"}'
303
        )
304

305
@app.get("/test/<scenario>")
306
def test_exceptions(request):
307
    scenario = request.path_params["scenario"]
308
    
309
    scenarios = {
310
        "http_400": lambda: HTTPException(400, "Bad request test"),
311
        "http_404": lambda: HTTPException(404, "Not found test"),
312
        "http_500": lambda: HTTPException(500, "Server error test"),
313
        "runtime": lambda: RuntimeError("Runtime error test"),
314
        "value": lambda: ValueError("Value error test"),
315
    }
316
    
317
    if scenario in scenarios:
318
        raise scenarios[scenario]()
319
    else:
320
        return {"message": f"Unknown scenario: {scenario}"}
321

322
app.start()
323
```