tessl/pypi-blacksheep

Fast web framework for Python asyncio

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

HTTP Client

Name: tessl/pypi-blacksheep
Author: tessl

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.

ClientSession

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

Basic Client Usage

import asyncio
from blacksheep.client import ClientSession
from blacksheep import JSONContent, TextContent

# Basic client usage
async def basic_example():
    async with ClientSession() as client:
        # GET request
        response = await client.get("https://api.example.com/users")
        data = await response.json()
        print(f"Users: {data}")
        
        # POST request with JSON
        json_data = JSONContent({"name": "Alice", "email": "alice@example.com"})
        response = await client.post("https://api.example.com/users", content=json_data)
        result = await response.json()
        print(f"Created: {result}")

asyncio.run(basic_example())

Client Configuration

import ssl
from blacksheep.client import ClientSession, ConnectionPools, CookieJar
from blacksheep import URL, Headers

# Advanced client configuration
async def configured_client():
    # Custom SSL context
    ssl_context = ssl.create_default_context()
    ssl_context.check_hostname = False
    ssl_context.verify_mode = ssl.CERT_NONE
    
    # Custom headers
    default_headers = Headers([
        (b"User-Agent", b"MyApp/1.0"),
        (b"Accept", b"application/json")
    ])
    
    client = ClientSession(
        base_url="https://api.example.com",
        ssl=ssl_context,                    # SSL configuration
        default_headers=default_headers,    # Default headers for all requests
        follow_redirects=True,             # Follow HTTP redirects  
        connection_timeout=30.0,           # Connection timeout (seconds)
        request_timeout=60.0,              # Request timeout (seconds)
        maximum_redirects=10,              # Max redirect hops
        cookie_jar=CookieJar(),            # Cookie storage
        middlewares=[]                     # Client middleware
    )
    
    async with client:
        response = await client.get("/users/123")
        return await response.json()

HTTP Methods

from blacksheep import Headers, JSONContent, FormContent
from typing import Optional, Dict, Union, Iterable, Tuple

# Type aliases for client
URLType = Union[str, bytes, URL]
HeadersType = Union[Dict[str, str], Iterable[Tuple[str, str]]]
ParamsType = Union[Dict[str, str], Iterable[Tuple[str, str]]]

async def http_methods_example():
    async with ClientSession() as client:
        # GET request
        response = await client.get(
            url="https://api.example.com/users",
            headers={"Authorization": "Bearer token123"},
            params={"page": "1", "limit": "10"}
        )
        
        # POST request
        json_data = JSONContent({"name": "Alice"})
        response = await client.post(
            url="https://api.example.com/users",
            content=json_data,
            headers={"Content-Type": "application/json"}
        )
        
        # PUT request  
        update_data = JSONContent({"name": "Alice Updated"})
        response = await client.put(
            url="https://api.example.com/users/123",
            content=update_data
        )
        
        # DELETE request
        response = await client.delete("https://api.example.com/users/123")
        
        # PATCH request
        patch_data = JSONContent({"email": "newemail@example.com"})
        response = await client.patch(
            url="https://api.example.com/users/123",
            content=patch_data
        )
        
        # HEAD request (headers only)
        response = await client.head("https://api.example.com/users/123")
        
        # OPTIONS request
        response = await client.options("https://api.example.com/users")
        
        # TRACE request
        response = await client.trace("https://api.example.com/debug")

Request Content Types

from blacksheep import JSONContent, TextContent, FormContent, MultiPartFormData, FormPart

async def content_types_example():
    async with ClientSession() as client:
        # JSON content
        json_data = {"user": {"name": "Alice", "age": 30}}
        response = await client.post(
            "https://api.example.com/users",
            content=JSONContent(json_data)
        )
        
        # Text content  
        text_data = "Plain text content"
        response = await client.post(
            "https://api.example.com/text",
            content=TextContent(text_data)
        )
        
        # Form data (URL-encoded)
        form_data = {"username": "alice", "password": "secret"}
        response = await client.post(
            "https://api.example.com/login",
            content=FormContent(form_data)
        )
        
        # Multipart form data (file upload)
        text_part = FormPart(b"title", b"My Document")
        with open("document.pdf", "rb") as f:
            file_data = f.read()
        file_part = FormPart(
            name=b"document",
            data=file_data,
            content_type=b"application/pdf",
            file_name=b"document.pdf"
        )
        
        multipart = MultiPartFormData([text_part, file_part])
        response = await client.post(
            "https://api.example.com/upload",
            content=multipart
        )

Response Handling

Process and parse HTTP responses from the client.

Response Properties

async def response_handling():
    async with ClientSession() as client:
        response = await client.get("https://api.example.com/users/123")
        
        # Basic response properties
        status_code: int = response.status
        reason_phrase: str = response.reason
        headers: Headers = response.headers
        
        # Check response status
        if response.status == 200:
            print("Success!")
        elif 400 <= response.status < 500:
            print("Client error")
        elif 500 <= response.status < 600:
            print("Server error")
            
        # Response content access
        raw_bytes: bytes = await response.read()
        text_content: str = await response.text()
        json_data: dict = await response.json()
        
        # Stream large responses
        async for chunk in response.stream():
            process_chunk(chunk)

Response Content Parsing

import json

async def response_parsing():
    async with ClientSession() as client:
        # JSON response
        response = await client.get("https://api.example.com/users")
        if response.declares_json():
            users = await response.json()
            
        # Custom JSON parsing
        response = await client.get("https://api.example.com/data")
        custom_data = await response.json(loads=json.loads)
        
        # Text response
        response = await client.get("https://api.example.com/readme")
        readme_text = await response.text()
        
        # Binary response
        response = await client.get("https://api.example.com/image.jpg")
        image_bytes = await response.read()
        
        # Form data response
        response = await client.get("https://api.example.com/form-data")
        if response.declares_content_type(b"application/x-www-form-urlencoded"):
            form_data = await response.form()
        
        # Check content type
        content_type = response.content_type()
        is_json = response.declares_json()
        is_xml = response.declares_xml()
        has_body = response.has_body()

Connection Management

Manage HTTP connections, pooling, and SSL configuration.

Connection Pools

from blacksheep.client import ConnectionPools

# Custom connection pool configuration
async def connection_pooling():
    pools = ConnectionPools(
        max_connections=100,      # Total connection limit
        max_connections_per_host=20  # Per-host connection limit
    )
    
    client = ClientSession(pools=pools)
    
    async with client:
        # Multiple concurrent requests reuse connections
        tasks = [
            client.get(f"https://api.example.com/users/{i}")
            for i in range(10)
        ]
        responses = await asyncio.gather(*tasks)
        
        for response in responses:
            data = await response.json()
            print(data)

SSL Configuration

import ssl

async def ssl_configuration():
    # Custom SSL context
    ssl_context = ssl.create_default_context()
    
    # Client certificate authentication
    ssl_context.load_cert_chain("client-cert.pem", "client-key.pem")
    
    # Custom CA certificate
    ssl_context.load_verify_locations("custom-ca.pem")
    
    # SSL options
    ssl_context.check_hostname = True
    ssl_context.verify_mode = ssl.CERT_REQUIRED
    
    client = ClientSession(ssl=ssl_context)
    
    async with client:
        # HTTPS request with custom SSL
        response = await client.get("https://secure-api.example.com/data")
        
    # Disable SSL verification (not recommended for production)
    insecure_client = ClientSession(ssl=False)
    
    async with insecure_client:
        response = await client.get("https://self-signed.example.com/api")

Redirects and Error Handling

Handle HTTP redirects and client errors effectively.

Redirect Configuration

from blacksheep.client import ClientSession

async def redirect_handling():
    # Follow redirects (default)
    client = ClientSession(follow_redirects=True, maximum_redirects=5)
    
    async with client:
        # Automatically follows redirects
        response = await client.get("https://example.com/redirect-me")
        final_url = response.url  # Final URL after redirects
        
    # Manual redirect handling
    manual_client = ClientSession(follow_redirects=False)
    
    async with manual_client:
        response = await client.get("https://example.com/redirect-me")
        
        if response.is_redirect():
            location = response.get_first_header(b"Location")
            if location:
                # Follow redirect manually
                next_response = await client.get(location.decode())

Client Exceptions

from blacksheep.client import (
    ConnectionTimeout, RequestTimeout, ConnectionClosedError,
    CircularRedirectError, MaximumRedirectsExceededError,
    MissingLocationForRedirect, UnsupportedRedirect
)

async def error_handling():
    async with ClientSession() as client:
        try:
            response = await client.get("https://slow-api.example.com/data")
            data = await response.json()
            
        except ConnectionTimeout:
            print("Connection timed out")
            
        except RequestTimeout:
            print("Request timed out")
            
        except ConnectionClosedError:
            print("Connection closed unexpectedly")
            
        except CircularRedirectError:
            print("Circular redirect detected")
            
        except MaximumRedirectsExceededError:
            print("Too many redirects")
            
        except MissingLocationForRedirect:
            print("Redirect response missing Location header")
            
        except UnsupportedRedirect:
            print("Unsupported redirect scheme")

Cookies and Sessions

Handle cookies and maintain session state across requests.

Cookie Management

from blacksheep.client import CookieJar
from blacksheep.cookies import Cookie

async def cookie_handling():
    # Automatic cookie handling
    cookie_jar = CookieJar()
    client = ClientSession(cookie_jar=cookie_jar)
    
    async with client:
        # Login request that sets cookies
        login_data = FormContent({"username": "alice", "password": "secret"})
        response = await client.post("https://example.com/login", content=login_data)
        
        # Subsequent requests automatically include cookies
        profile_response = await client.get("https://example.com/profile")
        
        # Cookies are automatically stored and sent
        
    # Manual cookie handling
    manual_client = ClientSession(cookie_jar=False)  # Disable automatic cookies
    
    async with manual_client:
        # Set cookies manually
        headers = {"Cookie": "session_id=abc123; user_pref=dark_theme"}
        response = await client.get("https://example.com/dashboard", headers=headers)

Client Middleware

Implement custom middleware for request/response processing.

Custom Middleware

import time
from typing import Callable, Awaitable

async def logging_middleware(
    request: Request, 
    handler: Callable[[Request], Awaitable[Response]]
) -> Response:
    """Log all client requests and responses"""
    print(f"Sending {request.method} to {request.url}")
    start_time = time.time()
    
    response = await handler(request)
    
    duration = time.time() - start_time
    print(f"Response: {response.status} ({duration:.3f}s)")
    
    return response

async def auth_middleware(
    request: Request,
    handler: Callable[[Request], Awaitable[Response]]
) -> Response:
    """Add authentication header to all requests"""
    token = get_auth_token()  # Get current auth token
    request.add_header(b"Authorization", f"Bearer {token}".encode())
    
    response = await handler(request)
    
    # Handle auth errors
    if response.status == 401:
        # Refresh token and retry
        new_token = await refresh_auth_token()
        request.set_header(b"Authorization", f"Bearer {new_token}".encode())
        response = await handler(request)
    
    return response

# Add middleware to client
async def client_with_middleware():
    client = ClientSession(middlewares=[logging_middleware, auth_middleware])
    
    async with client:
        # All requests will go through middleware
        response = await client.get("https://api.example.com/protected")

Timeouts and Performance

Configure timeouts and optimize client performance.

Timeout Configuration

async def timeout_configuration():
    # Configure timeouts
    client = ClientSession(
        connection_timeout=10.0,  # 10 seconds to establish connection
        request_timeout=30.0      # 30 seconds for complete request
    )
    
    async with client:
        try:
            # Fast timeout for health check
            response = await client.get("https://api.example.com/health")
            
        except ConnectionTimeout:
            print("Could not connect within 10 seconds")
            
        except RequestTimeout:
            print("Request did not complete within 30 seconds")

# Per-request timeout override
async def per_request_timeout():
    async with ClientSession() as client:
        # Override default timeout for specific request
        headers = {"X-Timeout": "5"}  # Custom timeout hint
        response = await client.get(
            "https://api.example.com/slow-operation",
            headers=headers
        )

Complete Client Example

A comprehensive example showing various client features:

import asyncio
import ssl
from blacksheep.client import ClientSession, CookieJar
from blacksheep import JSONContent, FormContent, Headers
from typing import Optional, Dict, Any

class APIClient:
    """Example API client with authentication and error handling"""
    
    def __init__(self, base_url: str, api_key: Optional[str] = None):
        self.base_url = base_url
        self.api_key = api_key
        self._client: Optional[ClientSession] = None
    
    async def __aenter__(self) -> 'APIClient':
        # Configure client
        default_headers = Headers([
            (b"User-Agent", b"MyAPIClient/1.0"),
            (b"Accept", b"application/json")
        ])
        
        if self.api_key:
            default_headers.add(b"Authorization", f"Bearer {self.api_key}".encode())
        
        self._client = ClientSession(
            base_url=self.base_url,
            default_headers=default_headers,
            follow_redirects=True,
            connection_timeout=10.0,
            request_timeout=30.0,
            cookie_jar=CookieJar()
        )
        
        await self._client.__aenter__()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self._client:
            await self._client.__aexit__(exc_type, exc_val, exc_tb)
    
    async def get_user(self, user_id: int) -> Dict[str, Any]:
        """Get user by ID"""
        response = await self._client.get(f"/users/{user_id}")
        
        if response.status == 404:
            raise ValueError(f"User {user_id} not found")
        elif response.status != 200:
            raise RuntimeError(f"API error: {response.status}")
        
        return await response.json()
    
    async def create_user(self, user_data: Dict[str, Any]) -> Dict[str, Any]:
        """Create new user"""
        content = JSONContent(user_data)
        response = await self._client.post("/users", content=content)
        
        if response.status != 201:
            error = await response.json()
            raise RuntimeError(f"Failed to create user: {error}")
        
        return await response.json()
    
    async def upload_file(self, file_path: str, user_id: int) -> Dict[str, Any]:
        """Upload file for user"""
        with open(file_path, "rb") as f:
            file_data = f.read()
        
        # Create multipart form data
        from blacksheep import MultiPartFormData, FormPart
        
        file_part = FormPart(
            name=b"file",
            data=file_data,
            content_type=b"application/octet-stream",
            file_name=file_path.encode()
        )
        
        user_part = FormPart(b"user_id", str(user_id).encode())
        multipart = MultiPartFormData([file_part, user_part])
        
        response = await self._client.post(f"/users/{user_id}/files", content=multipart)
        return await response.json()

# Usage example
async def main():
    async with APIClient("https://api.example.com", "your-api-key") as client:
        # Get user
        user = await client.get_user(123)
        print(f"User: {user}")
        
        # Create user
        new_user = await client.create_user({
            "name": "Alice Johnson",
            "email": "alice@example.com"
        })
        print(f"Created: {new_user}")
        
        # Upload file
        result = await client.upload_file("document.pdf", new_user["id"])
        print(f"Upload: {result}")

if __name__ == "__main__":
    asyncio.run(main())

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

Install with Tessl CLI