or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

Files

docs

additional.mdauth.mdclient.mdcore-server.mdindex.mdrequest-response.mdtesting.mdwebsockets.md

request-response.mddocs/

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.