0
# Request/Response Handling
1
2
This document covers BlackSheep's HTTP message handling system, including Request and Response classes, content types, headers, cookies, URL parsing, and data binding.
3
4
## Request Class
5
6
The `Request` class represents incoming HTTP requests with methods for accessing headers, body content, and parsed data.
7
8
### Request Properties
9
10
```python { .api }
11
from blacksheep import Request, URL
12
from typing import Optional, Dict, List
13
14
# Request creation and properties
15
request = Request("GET", b"https://example.com/users/123?q=search", headers=[
16
(b"Content-Type", b"application/json"),
17
(b"Authorization", b"Bearer token123")
18
])
19
20
# Basic properties
21
method: str = request.method # "GET", "POST", etc.
22
url: URL = request.url # Parsed URL object
23
query: Dict[str, List[str]] = request.query # Query parameters
24
route_values: Optional[Dict[str, str]] = request.route_values # Route params
25
cookies: Dict[str, str] = request.cookies # Request cookies
26
content: Optional[Content] = request.content # Request body content
27
28
# URL components
29
scheme: str = request.scheme # "http", "https"
30
host: str = request.host # "example.com"
31
path: str = request.path # "/users/123"
32
base_path: str = request.base_path # Application base path
33
34
# Client information
35
client_ip: str = request.client_ip # Client IP address
36
original_client_ip: str = request.original_client_ip # With setter
37
```
38
39
### Request Headers
40
41
```python { .api }
42
from blacksheep.headers import Headers
43
44
# Header access methods
45
content_type: bytes = request.content_type()
46
etag: Optional[bytes] = request.etag
47
if_none_match: Optional[bytes] = request.if_none_match
48
49
# Generic header methods
50
first_header = request.get_first_header(b"Authorization")
51
all_auth_headers = request.get_headers(b"Authorization")
52
single_header = request.get_single_header(b"Content-Type")
53
has_auth = request.has_header(b"Authorization")
54
55
# Header manipulation
56
request.add_header(b"X-Custom", b"value")
57
request.set_header(b"Content-Type", b"application/json")
58
request.remove_header(b"X-Remove-Me")
59
60
# Content type checking
61
is_json = request.declares_json()
62
is_xml = request.declares_xml()
63
has_content_type = request.declares_content_type(b"application/json")
64
has_body = request.has_body()
65
```
66
67
### Request Body Parsing
68
69
```python { .api }
70
import json
71
from blacksheep.contents import FormPart
72
from typing import Any, Union
73
74
# Raw body access
75
body_bytes: Optional[bytes] = await request.read()
76
body_text: str = await request.text()
77
78
# Streaming body
79
async for chunk in request.stream():
80
process_chunk(chunk)
81
82
# Structured data parsing
83
json_data: Any = await request.json()
84
json_with_custom = await request.json(loads=custom_json_loads)
85
86
# Form data parsing
87
form_data: Union[Dict, None] = await request.form()
88
# Returns: {"field1": "value1", "field2": ["value2", "value3"]}
89
90
# Multipart form data
91
multipart_parts: List[FormPart] = await request.multipart()
92
for part in multipart_parts:
93
name = part.name.decode()
94
data = part.data
95
filename = part.file_name.decode() if part.file_name else None
96
content_type = part.content_type
97
98
# File uploads
99
uploaded_files: List[FormPart] = await request.files()
100
specific_files = await request.files("avatar") # Files with name "avatar"
101
102
for file_part in uploaded_files:
103
filename = file_part.file_name.decode()
104
content_type = file_part.content_type.decode()
105
file_data = file_part.data
106
```
107
108
### Request Cookies and Session
109
110
```python { .api }
111
# Cookie access
112
cookie_value = request.get_cookie("session_id")
113
all_cookies = request.cookies # Dict[str, str]
114
115
# Session access (when sessions are configured)
116
from blacksheep.sessions import Session
117
118
session: Session = request.session
119
session["user_id"] = 123
120
user_id = session.get("user_id")
121
122
# Set cookies on request (for processing)
123
request.set_cookie("temp_cookie", "temp_value")
124
```
125
126
### Request Context
127
128
```python { .api }
129
# Authentication context
130
from guardpost import Identity
131
132
identity: Optional[Identity] = request.identity
133
if identity:
134
user_id = identity.id
135
claims = identity.claims
136
is_authenticated = identity.is_authenticated()
137
138
# ASGI scope access
139
scope = request.scope
140
asgi_version = scope.get("asgi", {}).get("version")
141
```
142
143
### Request Utility Functions
144
145
BlackSheep provides utility functions for common request processing tasks.
146
147
```python { .api }
148
from blacksheep.messages import (
149
is_cors_request, is_cors_preflight_request,
150
get_request_absolute_url, get_absolute_url_to_path
151
)
152
153
# CORS request detection
154
def is_cors_request(request: Request) -> bool:
155
"""Check if request is a CORS request"""
156
pass
157
158
def is_cors_preflight_request(request: Request) -> bool:
159
"""Check if request is a CORS preflight request"""
160
pass
161
162
# URL utilities
163
def get_request_absolute_url(request: Request) -> URL:
164
"""Get the absolute URL of the request"""
165
pass
166
167
def get_absolute_url_to_path(request: Request, path: str) -> URL:
168
"""Convert a path to absolute URL using request context"""
169
pass
170
```
171
172
#### Usage Examples
173
174
```python
175
@app.middleware
176
async def cors_middleware(request: Request, handler):
177
if is_cors_preflight_request(request):
178
return Response(200, headers=[
179
(b"Access-Control-Allow-Origin", b"*"),
180
(b"Access-Control-Allow-Methods", b"GET, POST, PUT, DELETE"),
181
(b"Access-Control-Allow-Headers", b"Content-Type, Authorization")
182
])
183
184
if is_cors_request(request):
185
response = await handler(request)
186
response.headers.add(Header(b"Access-Control-Allow-Origin", b"*"))
187
return response
188
189
return await handler(request)
190
191
@app.get("/api/resource")
192
async def get_resource(request: Request):
193
# Get absolute URL for current request
194
current_url = get_request_absolute_url(request)
195
196
# Generate absolute URLs for related resources
197
related_url = get_absolute_url_to_path(request, "/api/related")
198
199
return json({
200
"current_url": current_url.value.decode(),
201
"related_url": related_url.value.decode()
202
})
203
```
204
205
## Response Class
206
207
The `Response` class represents HTTP responses with status codes, headers, cookies, and content.
208
209
### Response Creation
210
211
```python { .api }
212
from blacksheep import Response, Content, TextContent, JSONContent
213
from blacksheep.headers import Headers
214
215
# Basic response creation
216
response = Response(200) # Status only
217
response = Response(200, content=TextContent("Hello World"))
218
219
# With headers
220
response = Response(
221
status=200,
222
headers=[(b"Content-Type", b"application/json")],
223
content=JSONContent({"message": "success"})
224
)
225
226
# Response properties
227
status: int = response.status # HTTP status code
228
reason: str = response.reason # Status reason phrase
229
content: Optional[Content] = response.content # Response body
230
cookies: Cookies = response.cookies # Response cookies
231
```
232
233
### Response Headers
234
235
```python { .api }
236
# Header manipulation (inherits from Message)
237
response.add_header(b"X-API-Version", b"1.0")
238
response.set_header(b"Cache-Control", b"no-cache")
239
response.remove_header(b"X-Powered-By")
240
241
# Content type setting
242
response.set_header(b"Content-Type", b"application/json")
243
244
# Custom headers
245
response.headers.add(b"X-Response-Time", b"0.123")
246
response.headers.set(b"X-Request-ID", request_id.encode())
247
```
248
249
### Response Cookies
250
251
```python { .api }
252
from blacksheep.cookies import Cookie, CookieSameSiteMode
253
from datetime import datetime, timedelta
254
255
# Cookie creation
256
cookie = Cookie(
257
name="session_id",
258
value="abc123",
259
expires=datetime.utcnow() + timedelta(days=7),
260
domain="example.com",
261
path="/",
262
http_only=True,
263
secure=True,
264
max_age=604800, # 7 days in seconds
265
same_site=CookieSameSiteMode.LAX
266
)
267
268
# Set cookies on response
269
response.set_cookie(cookie)
270
response.set_cookies([cookie1, cookie2, cookie3])
271
272
# Remove cookies
273
response.unset_cookie("old_session")
274
response.remove_cookie("temp_cookie") # Alias for unset_cookie
275
276
# Get response cookies
277
all_cookies = response.get_cookies()
278
specific_cookie = response.get_cookie("session_id")
279
```
280
281
### Response Utilities
282
283
```python { .api }
284
# Response type checking
285
is_redirect: bool = response.is_redirect() # 3xx status codes
286
287
# Response modification
288
new_response = response.with_content(new_content)
289
```
290
291
### Response Helper Functions
292
293
BlackSheep provides convenient helper functions for creating common HTTP responses.
294
295
```python { .api }
296
from blacksheep.server.responses import (
297
ok, created, accepted, no_content,
298
bad_request, unauthorized, forbidden, not_found,
299
file, redirect, json, text, html,
300
ContentDispositionType
301
)
302
303
# Success responses
304
def ok(message: Any = None) -> Response:
305
"""Returns HTTP 200 OK response"""
306
pass
307
308
def created(message: Any = None, location: AnyStr = "") -> Response:
309
"""Returns HTTP 201 Created response with optional location header"""
310
pass
311
312
def accepted(message: Any = None) -> Response:
313
"""Returns HTTP 202 Accepted response"""
314
pass
315
316
def no_content() -> Response:
317
"""Returns HTTP 204 No Content response"""
318
pass
319
320
# Error responses
321
def bad_request(message: Any = None) -> Response:
322
"""Returns HTTP 400 Bad Request response"""
323
pass
324
325
def unauthorized(message: str = "Unauthorized") -> Response:
326
"""Returns HTTP 401 Unauthorized response"""
327
pass
328
329
def forbidden(message: str = "Forbidden") -> Response:
330
"""Returns HTTP 403 Forbidden response"""
331
pass
332
333
def not_found() -> Response:
334
"""Returns HTTP 404 Not Found response"""
335
pass
336
337
# File responses
338
def file(
339
value: FileInput,
340
content_type: str,
341
*,
342
file_name: str = None,
343
content_disposition: ContentDispositionType = ContentDispositionType.ATTACHMENT,
344
) -> Response:
345
"""Returns file response with content type and disposition"""
346
pass
347
348
# Content responses
349
def json(obj: Any) -> Response:
350
"""Returns JSON response"""
351
pass
352
353
def text(message: str) -> Response:
354
"""Returns plain text response"""
355
pass
356
357
def html(content: str) -> Response:
358
"""Returns HTML response"""
359
pass
360
361
def redirect(location: str, permanent: bool = False) -> Response:
362
"""Returns redirect response (302 or 301)"""
363
pass
364
```
365
366
#### Response Helper Examples
367
368
```python
369
# File download with custom filename
370
@app.get("/download/{file_id}")
371
async def download_file(file_id: int):
372
file_data = await get_file_data(file_id)
373
return file(
374
file_data,
375
"application/pdf",
376
file_name="document.pdf",
377
content_disposition=ContentDispositionType.ATTACHMENT
378
)
379
380
# File display inline
381
@app.get("/preview/{image_id}")
382
async def preview_image(image_id: int):
383
image_data = await get_image_data(image_id)
384
return file(
385
image_data,
386
"image/jpeg",
387
content_disposition=ContentDispositionType.INLINE
388
)
389
390
# Created response with location
391
@app.post("/users")
392
async def create_user(user_data: FromJSON[dict]):
393
user = await create_user_in_db(user_data.value)
394
return created(
395
{"id": user.id, "name": user.name},
396
location=f"/users/{user.id}"
397
)
398
```
399
400
## Content Types
401
402
BlackSheep provides various content types for request and response bodies.
403
404
### Base Content Class
405
406
```python { .api }
407
from blacksheep.contents import Content
408
409
# Content properties and methods
410
class CustomContent(Content):
411
def __init__(self, data: str):
412
content_type = b"application/custom"
413
encoded_data = data.encode('utf-8')
414
super().__init__(content_type, encoded_data)
415
416
async def read(self) -> bytes:
417
return self.body
418
419
def dispose(self):
420
# Cleanup resources
421
pass
422
423
# Content properties
424
content_type: bytes = content.type
425
content_body: bytes = content.body
426
content_length: int = content.length
427
```
428
429
### Text Content
430
431
```python { .api }
432
from blacksheep.contents import TextContent
433
434
# Plain text content
435
text_content = TextContent("Hello, World!")
436
# Content-Type: text/plain; charset=utf-8
437
438
# Custom encoding (defaults to UTF-8)
439
text_content = TextContent("Hello", encoding="latin1")
440
```
441
442
### HTML Content
443
444
```python { .api }
445
from blacksheep.contents import HTMLContent, HtmlContent
446
447
# HTML content (both aliases work)
448
html_content = HTMLContent("<h1>Hello</h1>")
449
html_content = HtmlContent("<p>Paragraph</p>")
450
# Content-Type: text/html; charset=utf-8
451
452
# Template rendering example
453
template = "<h1>Hello, {{name}}!</h1>"
454
rendered = template.replace("{{name}}", "Alice")
455
html_content = HTMLContent(rendered)
456
```
457
458
### JSON Content
459
460
```python { .api }
461
from blacksheep.contents import JSONContent
462
import json
463
464
# JSON content with default serializer
465
data = {"users": [{"id": 1, "name": "Alice"}]}
466
json_content = JSONContent(data)
467
# Content-Type: application/json
468
469
# Custom JSON serializer
470
def custom_json_dumps(obj):
471
return json.dumps(obj, indent=2, sort_keys=True)
472
473
json_content = JSONContent(data, dumps=custom_json_dumps)
474
475
# Complex data serialization
476
from dataclasses import dataclass, asdict
477
478
@dataclass
479
class User:
480
id: int
481
name: str
482
483
user = User(1, "Alice")
484
json_content = JSONContent(asdict(user))
485
```
486
487
### Form Content
488
489
```python { .api }
490
from blacksheep.contents import FormContent
491
from typing import Dict, List, Tuple, Union
492
493
# URL-encoded form data
494
form_data: Dict[str, str] = {
495
"username": "alice",
496
"password": "secret",
497
"remember": "on"
498
}
499
form_content = FormContent(form_data)
500
# Content-Type: application/x-www-form-urlencoded
501
502
# Multiple values for same key
503
form_data: List[Tuple[str, str]] = [
504
("tags", "python"),
505
("tags", "web"),
506
("tags", "framework"),
507
("title", "My Post")
508
]
509
form_content = FormContent(form_data)
510
511
# Parsing form data
512
from blacksheep.contents import parse_www_form
513
514
form_string = "name=Alice&tags=python&tags=web"
515
parsed: Dict[str, Union[str, List[str]]] = parse_www_form(form_string)
516
# {"name": "Alice", "tags": ["python", "web"]}
517
```
518
519
### Multipart Form Data
520
521
```python { .api }
522
from blacksheep.contents import MultiPartFormData, FormPart
523
524
# Create form parts
525
text_part = FormPart(
526
name=b"title",
527
data=b"My Blog Post",
528
content_type=b"text/plain"
529
)
530
531
file_part = FormPart(
532
name=b"avatar",
533
data=image_bytes,
534
content_type=b"image/jpeg",
535
file_name=b"profile.jpg",
536
charset=b"utf-8"
537
)
538
539
# Multipart form
540
multipart = MultiPartFormData([text_part, file_part])
541
# Content-Type: multipart/form-data; boundary=...
542
543
# Access multipart properties
544
parts = multipart.parts
545
boundary = multipart.boundary
546
```
547
548
### Streamed Content
549
550
```python { .api }
551
from blacksheep.contents import StreamedContent
552
from typing import AsyncIterable
553
554
# Streaming content for large responses
555
async def data_generator() -> AsyncIterable[bytes]:
556
for i in range(1000):
557
yield f"Data chunk {i}\n".encode()
558
559
streamed = StreamedContent(
560
content_type=b"text/plain",
561
data_provider=data_generator,
562
data_length=-1 # Unknown length
563
)
564
565
# Known length streaming
566
def file_streamer():
567
with open("large_file.txt", "rb") as f:
568
while chunk := f.read(8192):
569
yield chunk
570
571
file_size = os.path.getsize("large_file.txt")
572
streamed = StreamedContent(
573
content_type=b"application/octet-stream",
574
data_provider=file_streamer,
575
data_length=file_size
576
)
577
```
578
579
### ASGI Content
580
581
```python { .api }
582
from blacksheep.contents import ASGIContent
583
from typing import Callable, Dict, Any
584
585
# Content from ASGI receive callable
586
async def asgi_handler(scope: Dict[str, Any], receive: Callable, send: Callable):
587
# Create content from ASGI receive
588
content = ASGIContent(receive)
589
590
# Stream content
591
async for chunk in content.stream():
592
process_chunk(chunk)
593
594
# Or read all at once
595
body = await content.read()
596
```
597
598
## Headers
599
600
BlackSheep provides comprehensive header handling with case-insensitive access and manipulation.
601
602
### Headers Collection
603
604
```python { .api }
605
from blacksheep.headers import Headers, Header
606
from typing import List, Tuple, Optional, Dict
607
608
# Header type
609
HeaderType = Tuple[bytes, bytes]
610
611
# Create headers collection
612
headers = Headers([
613
(b"Content-Type", b"application/json"),
614
(b"Authorization", b"Bearer token123"),
615
(b"Accept", b"application/json"),
616
(b"Accept", b"application/xml") # Multiple values
617
])
618
619
# Empty headers
620
headers = Headers()
621
622
# Header access methods
623
content_type_headers: Tuple[HeaderType] = headers.get(b"Content-Type")
624
header_tuples: List[HeaderType] = headers.get_tuples(b"Accept")
625
first_auth: Optional[bytes] = headers.get_first(b"Authorization")
626
single_ct: bytes = headers.get_single(b"Content-Type") # Throws if multiple
627
628
# Header manipulation
629
headers.add(b"X-Custom", b"value1")
630
headers.add(b"X-Custom", b"value2") # Multiple values allowed
631
headers.set(b"Content-Type", b"text/html") # Replace all existing
632
headers.remove(b"X-Remove-Me") # Remove all with this name
633
634
# Dictionary-like access
635
headers[b"Content-Type"] = b"application/json" # Set header
636
value = headers[b"Authorization"] # Get first value
637
del headers[b"X-Temp"] # Remove header
638
has_header = b"Content-Type" in headers # Check existence
639
```
640
641
### Header Utilities
642
643
```python { .api }
644
# Header iteration
645
for name, value in headers.items():
646
print(f"{name.decode()}: {value.decode()}")
647
648
# Header keys
649
header_names = headers.keys()
650
651
# Header merging
652
new_headers = [(b"X-New", b"value")]
653
headers.merge(new_headers)
654
655
# Dictionary update
656
header_dict = {b"Cache-Control": b"no-cache", b"X-Frame-Options": b"DENY"}
657
headers.update(header_dict)
658
659
# Add multiple headers
660
headers.add_many([
661
(b"X-Custom-1", b"value1"),
662
(b"X-Custom-2", b"value2")
663
])
664
665
# Or from dictionary
666
headers.add_many({
667
b"X-API-Version": b"1.0",
668
b"X-Rate-Limit": b"100"
669
})
670
671
# Header cloning
672
cloned_headers = headers.clone()
673
674
# Header combination
675
combined = headers + other_headers
676
headers += more_headers
677
```
678
679
### Individual Header
680
681
```python { .api }
682
from blacksheep.headers import Header
683
684
# Create individual header
685
header = Header(b"Content-Type", b"application/json")
686
687
# Header properties
688
name: bytes = header.name # b"Content-Type"
689
value: bytes = header.value # b"application/json"
690
691
# Header iteration (name, then value)
692
for item in header:
693
print(item) # b"Content-Type", then b"application/json"
694
695
# Header equality
696
header1 = Header(b"Content-Type", b"application/json")
697
header2 = Header(b"content-type", b"application/json")
698
are_equal = header1 == header2 # Case-insensitive comparison
699
```
700
701
## Cookies
702
703
BlackSheep provides comprehensive cookie support with security features and expiration handling.
704
705
### Cookie Class
706
707
```python { .api }
708
from blacksheep.cookies import Cookie, CookieSameSiteMode
709
from datetime import datetime, timedelta
710
711
# Basic cookie
712
cookie = Cookie("session_id", "abc123def456")
713
714
# Full cookie configuration
715
cookie = Cookie(
716
name="user_session",
717
value="encrypted_session_data",
718
expires=datetime.utcnow() + timedelta(days=30),
719
domain="example.com",
720
path="/",
721
http_only=True, # Prevent JavaScript access
722
secure=True, # Require HTTPS
723
max_age=2592000, # 30 days in seconds
724
same_site=CookieSameSiteMode.LAX
725
)
726
727
# Cookie properties
728
name: str = cookie.name
729
value: str = cookie.value
730
expires: Optional[datetime] = cookie.expires
731
domain: Optional[str] = cookie.domain
732
path: Optional[str] = cookie.path
733
http_only: bool = cookie.http_only
734
secure: bool = cookie.secure
735
max_age: int = cookie.max_age
736
same_site: CookieSameSiteMode = cookie.same_site
737
```
738
739
### SameSite Modes
740
741
```python { .api }
742
from blacksheep.cookies import CookieSameSiteMode
743
744
# SameSite attribute values
745
CookieSameSiteMode.UNDEFINED # Not specified
746
CookieSameSiteMode.LAX # Lax mode (default for most browsers)
747
CookieSameSiteMode.STRICT # Strict mode (no cross-site requests)
748
CookieSameSiteMode.NONE # None mode (requires Secure=True)
749
750
# Usage examples
751
session_cookie = Cookie("session", "value", same_site=CookieSameSiteMode.LAX)
752
csrf_cookie = Cookie("csrf", "token", same_site=CookieSameSiteMode.STRICT)
753
tracking_cookie = Cookie("track", "id", same_site=CookieSameSiteMode.NONE, secure=True)
754
```
755
756
### Cookie Utilities
757
758
```python { .api }
759
from blacksheep.cookies import (
760
datetime_to_cookie_format,
761
datetime_from_cookie_format,
762
parse_cookie,
763
write_response_cookie
764
)
765
766
# DateTime formatting for cookies
767
expire_time = datetime.utcnow() + timedelta(hours=24)
768
cookie_date: bytes = datetime_to_cookie_format(expire_time)
769
parsed_date: datetime = datetime_from_cookie_format(cookie_date)
770
771
# Cookie parsing from header value
772
cookie_header = b"session_id=abc123; Path=/; HttpOnly"
773
parsed_cookie: Cookie = parse_cookie(cookie_header)
774
775
# Cookie serialization for Set-Cookie header
776
cookie = Cookie("session", "value123", http_only=True)
777
set_cookie_header: bytes = write_response_cookie(cookie)
778
# b"session=value123; HttpOnly"
779
780
# Cookie cloning and comparison
781
cloned_cookie = cookie.clone()
782
is_equal = cookie == "session" # Compare by name
783
is_equal = cookie == cookie2 # Full comparison
784
```
785
786
## URL Handling
787
788
BlackSheep provides comprehensive URL parsing, validation, and manipulation capabilities.
789
790
### URL Class
791
792
```python { .api }
793
from blacksheep.url import URL, InvalidURL
794
795
# URL creation and parsing
796
try:
797
url = URL(b"https://api.example.com:8080/users/123?q=search&tag=python#section")
798
except InvalidURL as e:
799
print(f"Invalid URL: {e}")
800
801
# URL components
802
schema: Optional[bytes] = url.schema # b"https"
803
host: Optional[bytes] = url.host # b"api.example.com"
804
port: int = url.port # 8080
805
path: bytes = url.path # b"/users/123"
806
query: bytes = url.query # b"q=search&tag=python"
807
fragment: Optional[bytes] = url.fragment # b"section"
808
value: bytes = url.value # Original URL bytes
809
810
# URL properties
811
is_absolute: bool = url.is_absolute
812
```
813
814
### URL Manipulation
815
816
```python { .api }
817
# URL modification
818
base_url = URL(b"https://api.example.com")
819
new_url = base_url.with_host(b"api2.example.com")
820
secure_url = base_url.with_scheme(b"https")
821
822
# URL joining
823
base = URL(b"https://api.example.com/v1")
824
relative = URL(b"users/123")
825
full_url = base.join(relative) # https://api.example.com/v1/users/123
826
827
# Get base URL (scheme + host + port)
828
base_only = url.base_url() # https://api.example.com:8080
829
830
# URL concatenation
831
url1 = URL(b"https://api.example.com")
832
url2 = url1 + b"/users" # URL + bytes
833
url3 = url1 + URL(b"/posts") # URL + URL
834
```
835
836
### URL Validation
837
838
```python { .api }
839
# URL validation
840
def validate_url(url_string: str) -> bool:
841
try:
842
URL(url_string.encode())
843
return True
844
except InvalidURL:
845
return False
846
847
# Safe URL creation
848
def safe_url(url_string: str) -> Optional[URL]:
849
try:
850
return URL(url_string.encode())
851
except InvalidURL:
852
return None
853
```
854
855
## Data Binding
856
857
BlackSheep provides powerful data binding capabilities to extract and convert request data into typed parameters.
858
859
### Request Data Binding Types
860
861
```python { .api }
862
from blacksheep.server.bindings import (
863
FromJSON, FromQuery, FromRoute, FromForm,
864
FromHeader, FromCookie, FromServices, FromFiles,
865
FromBytes, FromText,
866
ClientInfo, ServerInfo, RequestUser, RequestURL, RequestMethod
867
)
868
869
# JSON body binding
870
@app.post("/users")
871
async def create_user(user_data: FromJSON[dict]):
872
user = user_data.value # Parsed JSON dict
873
return json({"created": True, "user": user})
874
875
# Type-safe JSON binding with dataclass
876
from dataclasses import dataclass
877
878
@dataclass
879
class CreateUserRequest:
880
name: str
881
email: str
882
age: int
883
884
@app.post("/users")
885
async def create_user_typed(request: FromJSON[CreateUserRequest]):
886
user = request.value # CreateUserRequest instance
887
return json({"name": user.name, "email": user.email})
888
```
889
890
### Route Parameter Binding
891
892
```python { .api }
893
# Route parameter binding with type conversion
894
@app.get("/users/{user_id:int}")
895
async def get_user(user_id: FromRoute[int]):
896
user_id_value: int = user_id.value
897
return json({"user_id": user_id_value})
898
899
# Multiple route parameters
900
@app.get("/users/{user_id:int}/posts/{post_id:int}")
901
async def get_user_post(
902
user_id: FromRoute[int],
903
post_id: FromRoute[int]
904
):
905
return json({
906
"user_id": user_id.value,
907
"post_id": post_id.value
908
})
909
910
# String route parameters (default)
911
@app.get("/categories/{category_name}")
912
async def get_category(category_name: FromRoute[str]):
913
name: str = category_name.value
914
return json({"category": name})
915
```
916
917
### Query Parameter Binding
918
919
```python { .api }
920
# Single query parameter
921
@app.get("/search")
922
async def search(q: FromQuery[str]):
923
query: str = q.value
924
return json({"query": query})
925
926
# Optional query parameters
927
@app.get("/users")
928
async def list_users(
929
page: FromQuery[int] = FromQuery(1),
930
limit: FromQuery[int] = FromQuery(10)
931
):
932
return json({
933
"page": page.value,
934
"limit": limit.value
935
})
936
937
# Multiple values for same parameter
938
@app.get("/posts")
939
async def filter_posts(tags: FromQuery[List[str]]):
940
tag_list: List[str] = tags.value # ["python", "web", "api"]
941
return json({"tags": tag_list})
942
```
943
944
### Form Data Binding
945
946
```python { .api }
947
# Form data binding
948
@app.post("/contact")
949
async def contact_form(form_data: FromForm[dict]):
950
data: dict = form_data.value
951
name = data.get("name")
952
email = data.get("email")
953
return json({"received": True})
954
955
# Typed form binding
956
@dataclass
957
class ContactForm:
958
name: str
959
email: str
960
message: str
961
962
@app.post("/contact")
963
async def contact_typed(form: FromForm[ContactForm]):
964
contact: ContactForm = form.value
965
return json({
966
"name": contact.name,
967
"email": contact.email
968
})
969
```
970
971
### Header and Cookie Binding
972
973
```python { .api }
974
# Header binding
975
@app.get("/protected")
976
async def protected_endpoint(auth_header: FromHeader[str]):
977
# Header: Authorization: Bearer token123
978
token: str = auth_header.value # "Bearer token123"
979
return json({"authorized": True})
980
981
# Specific header name
982
@app.get("/api")
983
async def api_endpoint(api_key: FromHeader[str] = FromHeader(name="X-API-Key")):
984
key: str = api_key.value
985
return json({"api_key_received": True})
986
987
# Cookie binding
988
@app.get("/dashboard")
989
async def dashboard(session_id: FromCookie[str]):
990
session: str = session_id.value
991
return json({"session": session})
992
993
# Optional cookie with default
994
@app.get("/preferences")
995
async def preferences(
996
theme: FromCookie[str] = FromCookie("light", name="theme")
997
):
998
user_theme: str = theme.value
999
return json({"theme": user_theme})
1000
```
1001
1002
### File Upload Binding
1003
1004
```python { .api }
1005
# File upload binding
1006
@app.post("/upload")
1007
async def upload_file(files: FromFiles):
1008
uploaded_files = files.value # List[FormPart]
1009
1010
for file_part in uploaded_files:
1011
filename = file_part.file_name.decode() if file_part.file_name else "unknown"
1012
content_type = file_part.content_type.decode() if file_part.content_type else "unknown"
1013
file_size = len(file_part.data)
1014
1015
# Save file
1016
with open(f"uploads/{filename}", "wb") as f:
1017
f.write(file_part.data)
1018
1019
return json({"uploaded": len(uploaded_files)})
1020
1021
# Specific file field
1022
@app.post("/avatar")
1023
async def upload_avatar(avatar: FromFiles = FromFiles(name="avatar")):
1024
if avatar.value:
1025
file_part = avatar.value[0] # First file
1026
# Process avatar
1027
return json({"avatar_uploaded": True})
1028
```
1029
1030
### Raw Body Binding
1031
1032
```python { .api }
1033
# Raw bytes binding
1034
@app.post("/binary")
1035
async def handle_binary(body: FromBytes):
1036
raw_data: bytes = body.value
1037
return Response(200, content=TextContent(f"Received {len(raw_data)} bytes"))
1038
1039
# Text binding
1040
@app.post("/text")
1041
async def handle_text(text: FromText):
1042
content: str = text.value
1043
return json({"text_length": len(content)})
1044
```
1045
1046
### Dependency Injection Binding
1047
1048
```python { .api }
1049
# Service injection
1050
@app.get("/users/{user_id:int}")
1051
async def get_user(
1052
user_id: FromRoute[int],
1053
user_service: FromServices[UserService]
1054
):
1055
service: UserService = user_service.value
1056
user = await service.get_by_id(user_id.value)
1057
return json(user)
1058
1059
# Multiple services
1060
@app.post("/orders")
1061
async def create_order(
1062
order_data: FromJSON[dict],
1063
order_service: FromServices[OrderService],
1064
email_service: FromServices[EmailService]
1065
):
1066
order = await order_service.create(order_data.value)
1067
await email_service.send_confirmation(order.email)
1068
return json({"order_id": order.id})
1069
```
1070
1071
### Context Information Binding
1072
1073
BlackSheep provides several binding classes to access request context information like client details, server info, and user identity.
1074
1075
```python { .api }
1076
from blacksheep.server.bindings import (
1077
ClientInfo, ServerInfo, RequestUser, RequestURL, RequestMethod
1078
)
1079
from guardpost.authentication import Identity
1080
from blacksheep import URL
1081
1082
# Context binding classes
1083
class ClientInfo(BoundValue[Tuple[str, int]]):
1084
"""Client IP and port information obtained from request scope"""
1085
pass
1086
1087
class ServerInfo(BoundValue[Tuple[str, int]]):
1088
"""Server IP and port information obtained from request scope"""
1089
pass
1090
1091
class RequestUser(BoundValue[Identity]):
1092
"""Returns the identity of the authenticated user"""
1093
pass
1094
1095
class RequestURL(BoundValue[URL]):
1096
"""Returns the URL of the request"""
1097
pass
1098
1099
class RequestMethod(BoundValue[str]):
1100
"""Returns the HTTP Method of the request"""
1101
pass
1102
```
1103
1104
#### Context Binding Usage Examples
1105
1106
```python
1107
# Client connection information
1108
@app.get("/info")
1109
async def client_info(client: ClientInfo):
1110
ip, port = client.value
1111
return json({"client_ip": ip, "client_port": port})
1112
1113
# Server information
1114
@app.get("/server-info")
1115
async def server_info(server: ServerInfo):
1116
host, port = server.value
1117
return json({"server_host": host, "server_port": port})
1118
1119
# Authenticated user information
1120
@app.get("/profile")
1121
async def user_profile(user: RequestUser):
1122
identity: Identity = user.value
1123
return json({
1124
"user_id": identity.claims.get("sub"),
1125
"name": identity.claims.get("name"),
1126
"roles": identity.claims.get("roles", [])
1127
})
1128
1129
# Request URL information
1130
@app.get("/request-info")
1131
async def request_info(url: RequestURL, method: RequestMethod):
1132
request_url: URL = url.value
1133
http_method: str = method.value
1134
return json({
1135
"method": http_method,
1136
"path": request_url.path,
1137
"query": request_url.query,
1138
"scheme": request_url.scheme,
1139
"host": request_url.host
1140
})
1141
```
1142
1143
This comprehensive request/response handling system provides type-safe, efficient processing of HTTP messages with rich content type support and flexible data binding capabilities.