pypi-fastapi

1
# Request and Response Handling
2

3
FastAPI provides comprehensive request and response handling through Starlette's Request and Response objects, along with FastAPI-specific enhancements like UploadFile for file handling and various response classes for different content types.
4

5
## Capabilities
6

7
### Request Object
8

9
HTTP request object providing access to all request data including headers, cookies, query parameters, form data, and JSON body.
10

11
```python { .api }
12
class Request:
13
    """
14
    Starlette Request object with all HTTP request functionality.
15
    
16
    Key attributes and methods for accessing request data:
17
    """
18
    
19
    # Request metadata
20
    method: str          # HTTP method (GET, POST, etc.)
21
    url: URL            # Complete URL object
22
    headers: Headers    # HTTP headers
23
    query_params: QueryParams  # Query string parameters
24
    path_params: dict   # Path parameters from URL
25
    cookies: dict       # HTTP cookies
26
    client: Address     # Client connection info
27
    
28
    # Request body access
29
    async def body(self) -> bytes:
30
        """Get raw request body as bytes."""
31
        pass
32
    
33
    async def json(self) -> Any:
34
        """Parse request body as JSON."""
35
        pass
36
    
37
    async def form(self) -> FormData:
38
        """Parse request body as form data."""
39
        pass
40
    
41
    # State and context
42
    state: State        # Request-scoped state storage
43
    scope: dict         # ASGI scope dictionary
44
    
45
    # Utility methods
46
    def url_for(self, name: str, **path_params) -> str:
47
        """Generate URL for named route."""
48
        pass
49
```
50

51
### Response Classes
52

53
Base response class and specialized response classes for different content types and use cases.
54

55
```python { .api }
56
class Response:
57
    def __init__(
58
        self,
59
        content: Any = None,
60
        status_code: int = 200,
61
        headers: dict = None,
62
        media_type: str = None,
63
        background: BackgroundTask = None,
64
    ) -> None:
65
        """
66
        Base HTTP response class.
67

68
        Parameters:
69
        - content: Response content (string, bytes, or None)
70
        - status_code: HTTP status code
71
        - headers: Additional HTTP headers
72
        - media_type: Content-Type header value
73
        - background: Background task to run after response
74
        """
75
    
76
    # Response properties
77
    status_code: int    # HTTP status code
78
    headers: Headers    # Response headers
79
    media_type: str     # Content-Type
80
    body: bytes         # Response body
81
    background: BackgroundTask  # Background task
82

83
class JSONResponse(Response):
84
    def __init__(
85
        self,
86
        content: Any = None,
87
        status_code: int = 200,
88
        headers: dict = None,
89
        media_type: str = "application/json",
90
        background: BackgroundTask = None,
91
    ) -> None:
92
        """
93
        JSON response with automatic serialization.
94

95
        Parameters:
96
        - content: Python object to serialize as JSON
97
        - Other parameters same as Response
98
        """
99

100
class HTMLResponse(Response):
101
    def __init__(
102
        self,
103
        content: str = None,
104
        status_code: int = 200,
105
        headers: dict = None,
106
        media_type: str = "text/html",
107
        background: BackgroundTask = None,
108
    ) -> None:
109
        """HTML response for returning HTML content."""
110

111
class PlainTextResponse(Response):
112
    def __init__(
113
        self,
114
        content: str = None,
115
        status_code: int = 200,
116
        headers: dict = None,
117
        media_type: str = "text/plain",
118
        background: BackgroundTask = None,
119
    ) -> None:
120
        """Plain text response."""
121

122
class RedirectResponse(Response):
123
    def __init__(
124
        self,
125
        url: str,
126
        status_code: int = 307,
127
        headers: dict = None,
128
        background: BackgroundTask = None,
129
    ) -> None:
130
        """
131
        HTTP redirect response.
132

133
        Parameters:
134
        - url: Target URL for redirection
135
        - status_code: Redirect status code (307, 302, 301, etc.)
136
        """
137

138
class FileResponse(Response):
139
    def __init__(
140
        self,
141
        path: str = None,
142
        status_code: int = 200,
143
        headers: dict = None,
144
        media_type: str = None,
145
        background: BackgroundTask = None,
146
        filename: str = None,
147
        stat_result: os.stat_result = None,
148
        method: str = None,
149
    ) -> None:
150
        """
151
        File download response.
152

153
        Parameters:
154
        - path: File system path to serve
155
        - filename: Filename for Content-Disposition header
156
        - stat_result: Cached file stat result for performance
157
        - method: HTTP method for conditional logic
158
        """
159

160
class StreamingResponse(Response):
161
    def __init__(
162
        self,
163
        content: Any,
164
        status_code: int = 200,
165
        headers: dict = None,
166
        media_type: str = None,
167
        background: BackgroundTask = None,
168
    ) -> None:
169
        """
170
        Streaming response for large data or real-time content.
171

172
        Parameters:
173
        - content: Iterable or async iterable of bytes/strings
174
        - Other parameters same as Response
175
        """
176
```
177

178
### FastAPI-Specific Response Classes
179

180
Enhanced JSON response classes using high-performance JSON libraries.
181

182
```python { .api }
183
class UJSONResponse(JSONResponse):
184
    def render(self, content: Any) -> bytes:
185
        """
186
        Ultra-fast JSON response using ujson library.
187
        
188
        Requires: pip install ujson
189
        Provides faster JSON serialization than standard library.
190
        """
191

192
class ORJSONResponse(JSONResponse):
193
    def render(self, content: Any) -> bytes:
194
        """
195
        Fast JSON response using orjson library.
196
        
197
        Requires: pip install orjson
198
        Fastest JSON serialization with additional features.
199
        """
200
```
201

202
### File Upload Handling
203

204
UploadFile class for handling multipart file uploads with async file operations.
205

206
```python { .api }
207
class UploadFile:
208
    def __init__(
209
        self,
210
        file: BinaryIO,
211
        *,
212
        size: int = None,
213
        filename: str = None,
214
        headers: Headers = None,
215
    ) -> None: ...
216
    
217
    # File metadata
218
    filename: Optional[str]      # Original filename
219
    content_type: Optional[str]  # MIME content type
220
    headers: Headers             # File-specific headers
221
    size: Optional[int]          # File size in bytes
222
    file: BinaryIO              # Underlying file object
223
    
224
    # Async file operations
225
    async def read(self, size: int = -1) -> bytes:
226
        """Read data from uploaded file."""
227
    
228
    async def readline(self, size: int = -1) -> bytes:
229
        """Read a line from uploaded file."""
230
    
231
    async def readlines(self) -> List[bytes]:
232
        """Read all lines from uploaded file."""
233
    
234
    async def write(self, data: bytes) -> None:
235
        """Write data to uploaded file."""
236
    
237
    async def seek(self, offset: int) -> None:
238
        """Seek to position in uploaded file."""
239
    
240
    async def close(self) -> None:
241
        """Close the uploaded file."""
242
```
243

244
## Usage Examples
245

246
### Accessing Request Data
247

248
```python
249
from fastapi import FastAPI, Request
250

251
app = FastAPI()
252

253
@app.post("/analyze-request")
254
async def analyze_request(request: Request):
255
    return {
256
        "method": request.method,
257
        "url": str(request.url),
258
        "headers": dict(request.headers),
259
        "query_params": dict(request.query_params),
260
        "path_params": request.path_params,
261
        "cookies": request.cookies,
262
        "client": f"{request.client.host}:{request.client.port}",
263
        "body": (await request.body()).decode()
264
    }
265

266
@app.get("/items/{item_id}")
267
async def get_item(item_id: int, request: Request):
268
    # Access path parameters
269
    path_params = request.path_params
270
    
271
    # Access query parameters
272
    query_params = request.query_params
273
    
274
    # Access headers
275
    user_agent = request.headers.get("user-agent")
276
    
277
    return {
278
        "item_id": item_id,
279
        "path_params": path_params,
280
        "query_params": dict(query_params),
281
        "user_agent": user_agent
282
    }
283
```
284

285
### Custom Response Types
286

287
```python
288
from fastapi import FastAPI
289
from fastapi.responses import HTMLResponse, PlainTextResponse, RedirectResponse
290

291
app = FastAPI()
292

293
@app.get("/html", response_class=HTMLResponse)
294
def get_html():
295
    html_content = """
296
    <html>
297
        <head><title>FastAPI HTML</title></head>
298
        <body><h1>Hello, HTML!</h1></body>
299
    </html>
300
    """
301
    return html_content
302

303
@app.get("/text", response_class=PlainTextResponse)
304
def get_text():
305
    return "Hello, plain text!"
306

307
@app.get("/redirect")
308
def redirect_to_docs():
309
    return RedirectResponse(url="/docs")
310

311
@app.get("/permanent-redirect")
312
def permanent_redirect():
313
    return RedirectResponse(url="/new-location", status_code=301)
314
```
315

316
### File Downloads
317

318
```python
319
from fastapi import FastAPI
320
from fastapi.responses import FileResponse
321
import tempfile
322
import os
323

324
app = FastAPI()
325

326
@app.get("/download-file")
327
def download_file():
328
    # Create a temporary file
329
    with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".txt") as f:
330
        f.write("This is a downloadable file content.")
331
        temp_file_path = f.name
332
    
333
    return FileResponse(
334
        path=temp_file_path,
335
        filename="download.txt",
336
        media_type="text/plain"
337
    )
338

339
@app.get("/download-image")
340
def download_image():
341
    # Serve an existing file
342
    file_path = "/path/to/image.jpg"
343
    return FileResponse(
344
        path=file_path,
345
        filename="image.jpg",
346
        media_type="image/jpeg"
347
    )
348
```
349

350
### Streaming Responses
351

352
```python
353
from fastapi import FastAPI
354
from fastapi.responses import StreamingResponse
355
import io
356
import json
357

358
app = FastAPI()
359

360
@app.get("/stream-data")
361
def stream_data():
362
    def generate_data():
363
        for i in range(1000):
364
            data = {"item": i, "value": f"data_{i}"}
365
            yield f"data: {json.dumps(data)}\n"
366
    
367
    return StreamingResponse(
368
        generate_data(),
369
        media_type="text/plain"
370
    )
371

372
@app.get("/stream-csv")
373
def stream_csv():
374
    def generate_csv():
375
        yield "id,name,email\n"
376
        for i in range(1000):
377
            yield f"{i},user_{i},user_{i}@example.com\n"
378
    
379
    return StreamingResponse(
380
        generate_csv(),
381
        media_type="text/csv",
382
        headers={"Content-Disposition": "attachment; filename=users.csv"}
383
    )
384

385
@app.get("/stream-file")
386
def stream_large_file():
387
    def iterfile(file_path: str):
388
        with open(file_path, mode="rb") as file_like:
389
            while True:
390
                chunk = file_like.read(1024)
391
                if not chunk:
392
                    break
393
                yield chunk
394
    
395
    return StreamingResponse(
396
        iterfile("/path/to/large/file.bin"),
397
        media_type="application/octet-stream"
398
    )
399
```
400

401
### File Upload Handling
402

403
```python
404
from fastapi import FastAPI, File, UploadFile, HTTPException
405
from typing import List
406
import aiofiles
407
import os
408

409
app = FastAPI()
410

411
@app.post("/upload-file/")
412
async def upload_file(file: UploadFile = File(...)):
413
    # Validate file type
414
    if not file.content_type.startswith("image/"):
415
        raise HTTPException(400, "File must be an image")
416
    
417
    # Read file content
418
    content = await file.read()
419
    
420
    # Save file
421
    file_path = f"uploads/{file.filename}"
422
    async with aiofiles.open(file_path, "wb") as f:
423
        await f.write(content)
424
    
425
    return {
426
        "filename": file.filename,
427
        "content_type": file.content_type,
428
        "size": len(content),
429
        "saved_to": file_path
430
    }
431

432
@app.post("/upload-multiple/")
433
async def upload_multiple_files(files: List[UploadFile] = File(...)):
434
    uploaded_files = []
435
    
436
    for file in files:
437
        content = await file.read()
438
        file_info = {
439
            "filename": file.filename,
440
            "content_type": file.content_type,
441
            "size": len(content)
442
        }
443
        uploaded_files.append(file_info)
444
        
445
        # Reset file pointer if you need to read again
446
        await file.seek(0)
447
    
448
    return {"uploaded_files": uploaded_files}
449

450
@app.post("/process-file/")
451
async def process_file(file: UploadFile = File(...)):
452
    # Process file line by line for large files
453
    processed_lines = []
454
    
455
    # Read file line by line
456
    content = await file.read()
457
    lines = content.decode().split('\n')
458
    
459
    for i, line in enumerate(lines):
460
        if line.strip():  # Skip empty lines
461
            processed_lines.append(f"Line {i+1}: {line.strip()}")
462
    
463
    await file.close()
464
    
465
    return {
466
        "filename": file.filename,
467
        "total_lines": len(processed_lines),
468
        "processed_lines": processed_lines[:10]  # Return first 10 lines
469
    }
470
```
471

472
### Custom Response with Headers
473

474
```python
475
from fastapi import FastAPI, Response
476
from fastapi.responses import JSONResponse
477

478
app = FastAPI()
479

480
@app.get("/custom-headers")
481
def custom_headers():
482
    content = {"message": "Custom headers response"}
483
    headers = {
484
        "X-Custom-Header": "Custom Value",
485
        "X-Processing-Time": "0.123",
486
        "Cache-Control": "no-cache"
487
    }
488
    return JSONResponse(content=content, headers=headers)
489

490
@app.get("/set-cookie")
491
def set_cookie(response: Response):
492
    content = {"message": "Cookie set"}
493
    response.set_cookie(
494
        key="session_id",
495
        value="abc123",
496
        max_age=3600,
497
        httponly=True,
498
        secure=True,
499
        samesite="lax"
500
    )
501
    return content
502

503
@app.get("/custom-status")
504
def custom_status():
505
    return JSONResponse(
506
        content={"message": "Created successfully"},
507
        status_code=201,
508
        headers={"Location": "/items/123"}
509
    )
510
```
511

512
### Response Model with Custom Response Class
513

514
```python
515
from fastapi import FastAPI
516
from fastapi.responses import ORJSONResponse
517
from pydantic import BaseModel
518
from typing import List
519

520
app = FastAPI(default_response_class=ORJSONResponse)
521

522
class Item(BaseModel):
523
    id: int
524
    name: str
525
    price: float
526

527
@app.get("/items", response_model=List[Item])
528
def get_items():
529
    # FastAPI will use ORJSONResponse for serialization
530
    return [
531
        {"id": 1, "name": "Item 1", "price": 10.5},
532
        {"id": 2, "name": "Item 2", "price": 20.0}
533
    ]
534

535
@app.get("/custom-json", response_class=ORJSONResponse)
536
def get_custom_json():
537
    # Explicitly use ORJSONResponse
538
    return {"message": "Fast JSON response"}
539
```