0
# HTTP Client
1

2
BlackSheep provides a full-featured async HTTP client built for high performance and ease of use. The `ClientSession` class supports connection pooling, SSL configuration, redirects, timeouts, middleware, and cookie management.
3

4
## ClientSession
5

6
The main HTTP client class for making requests to web services and APIs.
7

8
### Basic Client Usage
9

10
```python { .api }
11
import asyncio
12
from blacksheep.client import ClientSession
13
from blacksheep import JSONContent, TextContent
14

15
# Basic client usage
16
async def basic_example():
17
    async with ClientSession() as client:
18
        # GET request
19
        response = await client.get("https://api.example.com/users")
20
        data = await response.json()
21
        print(f"Users: {data}")
22
        
23
        # POST request with JSON
24
        json_data = JSONContent({"name": "Alice", "email": "alice@example.com"})
25
        response = await client.post("https://api.example.com/users", content=json_data)
26
        result = await response.json()
27
        print(f"Created: {result}")
28

29
asyncio.run(basic_example())
30
```
31

32
### Client Configuration
33

34
```python { .api }
35
import ssl
36
from blacksheep.client import ClientSession, ConnectionPools, CookieJar
37
from blacksheep import URL, Headers
38

39
# Advanced client configuration
40
async def configured_client():
41
    # Custom SSL context
42
    ssl_context = ssl.create_default_context()
43
    ssl_context.check_hostname = False
44
    ssl_context.verify_mode = ssl.CERT_NONE
45
    
46
    # Custom headers
47
    default_headers = Headers([
48
        (b"User-Agent", b"MyApp/1.0"),
49
        (b"Accept", b"application/json")
50
    ])
51
    
52
    client = ClientSession(
53
        base_url="https://api.example.com",
54
        ssl=ssl_context,                    # SSL configuration
55
        default_headers=default_headers,    # Default headers for all requests
56
        follow_redirects=True,             # Follow HTTP redirects  
57
        connection_timeout=30.0,           # Connection timeout (seconds)
58
        request_timeout=60.0,              # Request timeout (seconds)
59
        maximum_redirects=10,              # Max redirect hops
60
        cookie_jar=CookieJar(),            # Cookie storage
61
        middlewares=[]                     # Client middleware
62
    )
63
    
64
    async with client:
65
        response = await client.get("/users/123")
66
        return await response.json()
67
```
68

69
### HTTP Methods
70

71
```python { .api }
72
from blacksheep import Headers, JSONContent, FormContent
73
from typing import Optional, Dict, Union, Iterable, Tuple
74

75
# Type aliases for client
76
URLType = Union[str, bytes, URL]
77
HeadersType = Union[Dict[str, str], Iterable[Tuple[str, str]]]
78
ParamsType = Union[Dict[str, str], Iterable[Tuple[str, str]]]
79

80
async def http_methods_example():
81
    async with ClientSession() as client:
82
        # GET request
83
        response = await client.get(
84
            url="https://api.example.com/users",
85
            headers={"Authorization": "Bearer token123"},
86
            params={"page": "1", "limit": "10"}
87
        )
88
        
89
        # POST request
90
        json_data = JSONContent({"name": "Alice"})
91
        response = await client.post(
92
            url="https://api.example.com/users",
93
            content=json_data,
94
            headers={"Content-Type": "application/json"}
95
        )
96
        
97
        # PUT request  
98
        update_data = JSONContent({"name": "Alice Updated"})
99
        response = await client.put(
100
            url="https://api.example.com/users/123",
101
            content=update_data
102
        )
103
        
104
        # DELETE request
105
        response = await client.delete("https://api.example.com/users/123")
106
        
107
        # PATCH request
108
        patch_data = JSONContent({"email": "newemail@example.com"})
109
        response = await client.patch(
110
            url="https://api.example.com/users/123",
111
            content=patch_data
112
        )
113
        
114
        # HEAD request (headers only)
115
        response = await client.head("https://api.example.com/users/123")
116
        
117
        # OPTIONS request
118
        response = await client.options("https://api.example.com/users")
119
        
120
        # TRACE request
121
        response = await client.trace("https://api.example.com/debug")
122
```
123

124
### Request Content Types
125

126
```python { .api }
127
from blacksheep import JSONContent, TextContent, FormContent, MultiPartFormData, FormPart
128

129
async def content_types_example():
130
    async with ClientSession() as client:
131
        # JSON content
132
        json_data = {"user": {"name": "Alice", "age": 30}}
133
        response = await client.post(
134
            "https://api.example.com/users",
135
            content=JSONContent(json_data)
136
        )
137
        
138
        # Text content  
139
        text_data = "Plain text content"
140
        response = await client.post(
141
            "https://api.example.com/text",
142
            content=TextContent(text_data)
143
        )
144
        
145
        # Form data (URL-encoded)
146
        form_data = {"username": "alice", "password": "secret"}
147
        response = await client.post(
148
            "https://api.example.com/login",
149
            content=FormContent(form_data)
150
        )
151
        
152
        # Multipart form data (file upload)
153
        text_part = FormPart(b"title", b"My Document")
154
        with open("document.pdf", "rb") as f:
155
            file_data = f.read()
156
        file_part = FormPart(
157
            name=b"document",
158
            data=file_data,
159
            content_type=b"application/pdf",
160
            file_name=b"document.pdf"
161
        )
162
        
163
        multipart = MultiPartFormData([text_part, file_part])
164
        response = await client.post(
165
            "https://api.example.com/upload",
166
            content=multipart
167
        )
168
```
169

170
## Response Handling
171

172
Process and parse HTTP responses from the client.
173

174
### Response Properties
175

176
```python { .api }
177
async def response_handling():
178
    async with ClientSession() as client:
179
        response = await client.get("https://api.example.com/users/123")
180
        
181
        # Basic response properties
182
        status_code: int = response.status
183
        reason_phrase: str = response.reason
184
        headers: Headers = response.headers
185
        
186
        # Check response status
187
        if response.status == 200:
188
            print("Success!")
189
        elif 400 <= response.status < 500:
190
            print("Client error")
191
        elif 500 <= response.status < 600:
192
            print("Server error")
193
            
194
        # Response content access
195
        raw_bytes: bytes = await response.read()
196
        text_content: str = await response.text()
197
        json_data: dict = await response.json()
198
        
199
        # Stream large responses
200
        async for chunk in response.stream():
201
            process_chunk(chunk)
202
```
203

204
### Response Content Parsing
205

206
```python { .api }
207
import json
208

209
async def response_parsing():
210
    async with ClientSession() as client:
211
        # JSON response
212
        response = await client.get("https://api.example.com/users")
213
        if response.declares_json():
214
            users = await response.json()
215
            
216
        # Custom JSON parsing
217
        response = await client.get("https://api.example.com/data")
218
        custom_data = await response.json(loads=json.loads)
219
        
220
        # Text response
221
        response = await client.get("https://api.example.com/readme")
222
        readme_text = await response.text()
223
        
224
        # Binary response
225
        response = await client.get("https://api.example.com/image.jpg")
226
        image_bytes = await response.read()
227
        
228
        # Form data response
229
        response = await client.get("https://api.example.com/form-data")
230
        if response.declares_content_type(b"application/x-www-form-urlencoded"):
231
            form_data = await response.form()
232
        
233
        # Check content type
234
        content_type = response.content_type()
235
        is_json = response.declares_json()
236
        is_xml = response.declares_xml()
237
        has_body = response.has_body()
238
```
239

240
## Connection Management
241

242
Manage HTTP connections, pooling, and SSL configuration.
243

244
### Connection Pools
245

246
```python { .api }
247
from blacksheep.client import ConnectionPools
248

249
# Custom connection pool configuration
250
async def connection_pooling():
251
    pools = ConnectionPools(
252
        max_connections=100,      # Total connection limit
253
        max_connections_per_host=20  # Per-host connection limit
254
    )
255
    
256
    client = ClientSession(pools=pools)
257
    
258
    async with client:
259
        # Multiple concurrent requests reuse connections
260
        tasks = [
261
            client.get(f"https://api.example.com/users/{i}")
262
            for i in range(10)
263
        ]
264
        responses = await asyncio.gather(*tasks)
265
        
266
        for response in responses:
267
            data = await response.json()
268
            print(data)
269
```
270

271
### SSL Configuration
272

273
```python { .api }
274
import ssl
275

276
async def ssl_configuration():
277
    # Custom SSL context
278
    ssl_context = ssl.create_default_context()
279
    
280
    # Client certificate authentication
281
    ssl_context.load_cert_chain("client-cert.pem", "client-key.pem")
282
    
283
    # Custom CA certificate
284
    ssl_context.load_verify_locations("custom-ca.pem")
285
    
286
    # SSL options
287
    ssl_context.check_hostname = True
288
    ssl_context.verify_mode = ssl.CERT_REQUIRED
289
    
290
    client = ClientSession(ssl=ssl_context)
291
    
292
    async with client:
293
        # HTTPS request with custom SSL
294
        response = await client.get("https://secure-api.example.com/data")
295
        
296
    # Disable SSL verification (not recommended for production)
297
    insecure_client = ClientSession(ssl=False)
298
    
299
    async with insecure_client:
300
        response = await client.get("https://self-signed.example.com/api")
301
```
302

303
## Redirects and Error Handling
304

305
Handle HTTP redirects and client errors effectively.
306

307
### Redirect Configuration
308

309
```python { .api }
310
from blacksheep.client import ClientSession
311

312
async def redirect_handling():
313
    # Follow redirects (default)
314
    client = ClientSession(follow_redirects=True, maximum_redirects=5)
315
    
316
    async with client:
317
        # Automatically follows redirects
318
        response = await client.get("https://example.com/redirect-me")
319
        final_url = response.url  # Final URL after redirects
320
        
321
    # Manual redirect handling
322
    manual_client = ClientSession(follow_redirects=False)
323
    
324
    async with manual_client:
325
        response = await client.get("https://example.com/redirect-me")
326
        
327
        if response.is_redirect():
328
            location = response.get_first_header(b"Location")
329
            if location:
330
                # Follow redirect manually
331
                next_response = await client.get(location.decode())
332
```
333

334
### Client Exceptions
335

336
```python { .api }
337
from blacksheep.client import (
338
    ConnectionTimeout, RequestTimeout, ConnectionClosedError,
339
    CircularRedirectError, MaximumRedirectsExceededError,
340
    MissingLocationForRedirect, UnsupportedRedirect
341
)
342

343
async def error_handling():
344
    async with ClientSession() as client:
345
        try:
346
            response = await client.get("https://slow-api.example.com/data")
347
            data = await response.json()
348
            
349
        except ConnectionTimeout:
350
            print("Connection timed out")
351
            
352
        except RequestTimeout:
353
            print("Request timed out")
354
            
355
        except ConnectionClosedError:
356
            print("Connection closed unexpectedly")
357
            
358
        except CircularRedirectError:
359
            print("Circular redirect detected")
360
            
361
        except MaximumRedirectsExceededError:
362
            print("Too many redirects")
363
            
364
        except MissingLocationForRedirect:
365
            print("Redirect response missing Location header")
366
            
367
        except UnsupportedRedirect:
368
            print("Unsupported redirect scheme")
369
```
370

371
## Cookies and Sessions
372

373
Handle cookies and maintain session state across requests.
374

375
### Cookie Management
376

377
```python { .api }
378
from blacksheep.client import CookieJar
379
from blacksheep.cookies import Cookie
380

381
async def cookie_handling():
382
    # Automatic cookie handling
383
    cookie_jar = CookieJar()
384
    client = ClientSession(cookie_jar=cookie_jar)
385
    
386
    async with client:
387
        # Login request that sets cookies
388
        login_data = FormContent({"username": "alice", "password": "secret"})
389
        response = await client.post("https://example.com/login", content=login_data)
390
        
391
        # Subsequent requests automatically include cookies
392
        profile_response = await client.get("https://example.com/profile")
393
        
394
        # Cookies are automatically stored and sent
395
        
396
    # Manual cookie handling
397
    manual_client = ClientSession(cookie_jar=False)  # Disable automatic cookies
398
    
399
    async with manual_client:
400
        # Set cookies manually
401
        headers = {"Cookie": "session_id=abc123; user_pref=dark_theme"}
402
        response = await client.get("https://example.com/dashboard", headers=headers)
403
```
404

405
## Client Middleware
406

407
Implement custom middleware for request/response processing.
408

409
### Custom Middleware
410

411
```python { .api }
412
import time
413
from typing import Callable, Awaitable
414

415
async def logging_middleware(
416
    request: Request, 
417
    handler: Callable[[Request], Awaitable[Response]]
418
) -> Response:
419
    """Log all client requests and responses"""
420
    print(f"Sending {request.method} to {request.url}")
421
    start_time = time.time()
422
    
423
    response = await handler(request)
424
    
425
    duration = time.time() - start_time
426
    print(f"Response: {response.status} ({duration:.3f}s)")
427
    
428
    return response
429

430
async def auth_middleware(
431
    request: Request,
432
    handler: Callable[[Request], Awaitable[Response]]
433
) -> Response:
434
    """Add authentication header to all requests"""
435
    token = get_auth_token()  # Get current auth token
436
    request.add_header(b"Authorization", f"Bearer {token}".encode())
437
    
438
    response = await handler(request)
439
    
440
    # Handle auth errors
441
    if response.status == 401:
442
        # Refresh token and retry
443
        new_token = await refresh_auth_token()
444
        request.set_header(b"Authorization", f"Bearer {new_token}".encode())
445
        response = await handler(request)
446
    
447
    return response
448

449
# Add middleware to client
450
async def client_with_middleware():
451
    client = ClientSession(middlewares=[logging_middleware, auth_middleware])
452
    
453
    async with client:
454
        # All requests will go through middleware
455
        response = await client.get("https://api.example.com/protected")
456
```
457

458
## Timeouts and Performance
459

460
Configure timeouts and optimize client performance.
461

462
### Timeout Configuration
463

464
```python { .api }
465
async def timeout_configuration():
466
    # Configure timeouts
467
    client = ClientSession(
468
        connection_timeout=10.0,  # 10 seconds to establish connection
469
        request_timeout=30.0      # 30 seconds for complete request
470
    )
471
    
472
    async with client:
473
        try:
474
            # Fast timeout for health check
475
            response = await client.get("https://api.example.com/health")
476
            
477
        except ConnectionTimeout:
478
            print("Could not connect within 10 seconds")
479
            
480
        except RequestTimeout:
481
            print("Request did not complete within 30 seconds")
482

483
# Per-request timeout override
484
async def per_request_timeout():
485
    async with ClientSession() as client:
486
        # Override default timeout for specific request
487
        headers = {"X-Timeout": "5"}  # Custom timeout hint
488
        response = await client.get(
489
            "https://api.example.com/slow-operation",
490
            headers=headers
491
        )
492
```
493

494
## Complete Client Example
495

496
A comprehensive example showing various client features:
497

498
```python { .api }
499
import asyncio
500
import ssl
501
from blacksheep.client import ClientSession, CookieJar
502
from blacksheep import JSONContent, FormContent, Headers
503
from typing import Optional, Dict, Any
504

505
class APIClient:
506
    """Example API client with authentication and error handling"""
507
    
508
    def __init__(self, base_url: str, api_key: Optional[str] = None):
509
        self.base_url = base_url
510
        self.api_key = api_key
511
        self._client: Optional[ClientSession] = None
512
    
513
    async def __aenter__(self) -> 'APIClient':
514
        # Configure client
515
        default_headers = Headers([
516
            (b"User-Agent", b"MyAPIClient/1.0"),
517
            (b"Accept", b"application/json")
518
        ])
519
        
520
        if self.api_key:
521
            default_headers.add(b"Authorization", f"Bearer {self.api_key}".encode())
522
        
523
        self._client = ClientSession(
524
            base_url=self.base_url,
525
            default_headers=default_headers,
526
            follow_redirects=True,
527
            connection_timeout=10.0,
528
            request_timeout=30.0,
529
            cookie_jar=CookieJar()
530
        )
531
        
532
        await self._client.__aenter__()
533
        return self
534
    
535
    async def __aexit__(self, exc_type, exc_val, exc_tb):
536
        if self._client:
537
            await self._client.__aexit__(exc_type, exc_val, exc_tb)
538
    
539
    async def get_user(self, user_id: int) -> Dict[str, Any]:
540
        """Get user by ID"""
541
        response = await self._client.get(f"/users/{user_id}")
542
        
543
        if response.status == 404:
544
            raise ValueError(f"User {user_id} not found")
545
        elif response.status != 200:
546
            raise RuntimeError(f"API error: {response.status}")
547
        
548
        return await response.json()
549
    
550
    async def create_user(self, user_data: Dict[str, Any]) -> Dict[str, Any]:
551
        """Create new user"""
552
        content = JSONContent(user_data)
553
        response = await self._client.post("/users", content=content)
554
        
555
        if response.status != 201:
556
            error = await response.json()
557
            raise RuntimeError(f"Failed to create user: {error}")
558
        
559
        return await response.json()
560
    
561
    async def upload_file(self, file_path: str, user_id: int) -> Dict[str, Any]:
562
        """Upload file for user"""
563
        with open(file_path, "rb") as f:
564
            file_data = f.read()
565
        
566
        # Create multipart form data
567
        from blacksheep import MultiPartFormData, FormPart
568
        
569
        file_part = FormPart(
570
            name=b"file",
571
            data=file_data,
572
            content_type=b"application/octet-stream",
573
            file_name=file_path.encode()
574
        )
575
        
576
        user_part = FormPart(b"user_id", str(user_id).encode())
577
        multipart = MultiPartFormData([file_part, user_part])
578
        
579
        response = await self._client.post(f"/users/{user_id}/files", content=multipart)
580
        return await response.json()
581

582
# Usage example
583
async def main():
584
    async with APIClient("https://api.example.com", "your-api-key") as client:
585
        # Get user
586
        user = await client.get_user(123)
587
        print(f"User: {user}")
588
        
589
        # Create user
590
        new_user = await client.create_user({
591
            "name": "Alice Johnson",
592
            "email": "alice@example.com"
593
        })
594
        print(f"Created: {new_user}")
595
        
596
        # Upload file
597
        result = await client.upload_file("document.pdf", new_user["id"])
598
        print(f"Upload: {result}")
599

600
if __name__ == "__main__":
601
    asyncio.run(main())
602
```
603

604
The BlackSheep HTTP client provides a powerful, flexible foundation for building HTTP-based integrations with excellent performance, comprehensive error handling, and rich configuration options.