Request & Response Models¶

Request¶

Request `dataclass` ¶

Request(method: HTTPMethod, path: str, headers: Union[Dict[str, str], MultiValueHeaders], body: Optional[BinaryIO] = None, query_params: Optional[Dict[str, str]] = None, path_params: Optional[Dict[str, str]] = None, tls: bool = False, client_cert: Optional[Dict[str, Any]] = None)

Represents an HTTP request.

Supports the ASGI TLS extension for TLS/SSL connection information.

The body is a file-like stream of bytes that can be read by content parsers. This allows efficient handling of large request bodies without loading everything into memory.

Functions¶

__post_init__ ¶

__post_init__()

Ensure headers is a MultiValueHeaders for case-insensitive header lookups.

get_accept_header ¶

get_accept_header() -> str

Get the Accept header, defaulting to / if not present.

get_content_type ¶

get_content_type() -> Optional[str]

Get the Content-Type header.

get_authorization_header ¶

get_authorization_header() -> Optional[str]

Get the Authorization header.

get_if_match ¶

get_if_match() -> Optional[List[str]]

Get the If-Match header values as a list of ETags.

get_if_none_match ¶

get_if_none_match() -> Optional[List[str]]

Get the If-None-Match header values as a list of ETags.

get_if_modified_since ¶

get_if_modified_since() -> Optional[datetime]

Get the If-Modified-Since header as a datetime object.

get_if_unmodified_since ¶

get_if_unmodified_since() -> Optional[datetime]

Get the If-Unmodified-Since header as a datetime object.

Response¶

Response `dataclass` ¶

Response(status_code: int, body: Optional[Union[str, bytes, BinaryIO, Path, dict, list]] = None, headers: Optional[Union[Dict[str, str], MultiValueHeaders]] = None, content_type: Optional[str] = None, request: Optional[Request] = None, available_content_types: Optional[list] = None, pre_calculated_headers: Optional[Union[Dict[str, str], MultiValueHeaders]] = None, etag: Optional[str] = None, last_modified: Optional[datetime] = None, range_start: Optional[int] = None, range_end: Optional[int] = None)

Represents an HTTP response.

The body can be: - str: Will be encoded to UTF-8 bytes - bytes: Used directly - BinaryIO: File-like object that will be streamed (useful for large files, S3 objects, etc.) - Path: Local filesystem path to a file (will be served efficiently) - dict/list: Will be JSON-encoded - None: Empty response body

For Path objects: - ASGI servers will use the path send extension for efficient file serving - Lambda will read the file and send as body - Content-Type is automatically detected from file extension - Content-Length is automatically set from file size

Range Request Support: - range_start/range_end: Set by framework when processing Range header - Adapters use these fields to send only requested byte range - ASGI adapter uses zero-copy extension when possible

Functions¶

__post_init__ ¶

__post_init__()

Initialize Response object after dataclass creation.

set_etag ¶

set_etag(etag: str, weak: bool = False)

Set the ETag header.

Parameters:

Name	Type	Description	Default
`etag`	`str`	The ETag value (without quotes)	required
`weak`	`bool`	Whether this is a weak ETag (prefixed with W/)	`False`

set_last_modified ¶

set_last_modified(last_modified: datetime)

Set the Last-Modified header.

Parameters:

Name	Type	Description	Default
`last_modified`	`datetime`	The last modified datetime	required

is_range_response ¶

is_range_response() -> bool

Check if this is a partial content (range) response.

Returns:

Type	Description
`bool`	True if range_start and range_end are set, indicating this is a 206 Partial Content response

generate_etag_from_content ¶

generate_etag_from_content(weak: bool = False)

Generate and set ETag based on response body content.

Parameters:

Name	Type	Description	Default
`weak`	`bool`	Whether to generate a weak ETag	`False`

Note

For streaming bodies (BinaryIO), this will read the entire stream to calculate the hash. The stream will be reset to the beginning after hashing.

HTTPMethod¶

HTTPMethod ¶

Bases: Enum

Enumeration of supported HTTP methods.

Overview¶

These are the core models for handling HTTP requests and responses in RestMachine.

Request Model¶

The Request object encapsulates an incoming HTTP request with:

method - HTTP method (GET, POST, PUT, DELETE, etc.)
path - Request path
headers - HTTP headers (case-insensitive)
query_params - Query string parameters
path_params - Path parameters from route matching
body - Request body (string or bytes)

Response Model¶

The Response object represents an HTTP response with:

status_code - HTTP status code (200, 404, 500, etc.)
body - Response body (string)
headers - HTTP response headers

HTTPMethod Enum¶

Enum of supported HTTP methods: - GET - POST - PUT - DELETE - PATCH - HEAD - OPTIONS

Request Usage¶

Accessing Request Data¶

@app.get('/users/{user_id}')
def get_user(request: Request):
    # Path parameters
    user_id = request.path_params['user_id']

    # Query parameters
    page = request.query_params.get('page', '1')

    # Headers
    auth = request.headers.get('authorization')

    # Body (for POST/PUT)
    # body = request.body

    return {"user_id": user_id, "page": page}

Creating Requests Manually¶

Useful for testing:

from restmachine import Request, HTTPMethod

request = Request(
    method=HTTPMethod.GET,
    path="/users/123",
    headers={"authorization": "Bearer token"},
    query_params={"page": "1"},
    path_params={"user_id": "123"},
    body=""
)

response = app.execute(request)

Response Usage¶

Returning Responses¶

Handlers can return responses in multiple ways:

1. Dictionary (auto-converted to JSON):

@app.get('/users')
def list_users():
    return {"users": [...]}  # 200 OK with JSON

2. Tuple (body, status_code):

@app.post('/users')
def create_user(request):
    return {"created": True}, 201  # 201 Created

3. Tuple (body, status_code, headers):

@app.get('/file')
def download_file():
    return "file contents", 200, {
        "Content-Type": "text/plain",
        "Content-Disposition": "attachment; filename=file.txt"
    }

4. Explicit Response object:

from restmachine import Response

@app.get('/custom')
def custom_response():
    return Response(
        status_code=200,
        body='{"custom": true}',
        headers={"X-Custom-Header": "value"}
    )

Response Status Codes¶

Use http.HTTPStatus for readable status codes:

from http import HTTPStatus

@app.post('/users')
def create_user(request):
    return {"created": True}, HTTPStatus.CREATED

@app.get('/users/{user_id}')
def get_user(request):
    user_id = request.path_params['user_id']
    user = db.get(user_id)

    if not user:
        return {"error": "Not found"}, HTTPStatus.NOT_FOUND

    return user, HTTPStatus.OK

Common status codes: - 200 OK - Successful GET/PUT/PATCH - 201 Created - Successful POST - 204 No Content - Successful DELETE - 400 Bad Request - Invalid request - 401 Unauthorized - Authentication required - 403 Forbidden - Insufficient permissions - 404 Not Found - Resource not found - 500 Internal Server Error - Server error

HTTPMethod Usage¶

The HTTPMethod enum provides type-safe HTTP method constants:

from restmachine import HTTPMethod

# Used internally for route matching
@app.route(HTTPMethod.POST, '/users')
def create_user(request):
    return {"created": True}

# Or use string (converted automatically)
@app.route('PATCH', '/users/{user_id}')
def patch_user(request):
    return {"patched": True}

Headers¶

Request Headers¶

Headers are case-insensitive:

@app.get('/data')
def get_data(request: Request):
    # All of these work:
    auth = request.headers.get('Authorization')
    auth = request.headers.get('authorization')
    auth = request.headers.get('AUTHORIZATION')

    content_type = request.headers.get('content-type')

    return {"auth": auth, "type": content_type}

Response Headers¶

Set response headers in tuple return or Response object:

# Via tuple
@app.get('/data')
def get_data():
    return {"data": ...}, 200, {
        "Cache-Control": "max-age=3600",
        "X-API-Version": "1.0"
    }

# Via Response
from restmachine import Response

@app.get('/data')
def get_data():
    return Response(
        status_code=200,
        body='{"data": "value"}',
        headers={
            "Cache-Control": "max-age=3600",
            "X-API-Version": "1.0"
        }
    )

Query Parameters¶

Access query parameters from the request:

@app.get('/search')
def search(request: Request):
    # GET /search?q=python&page=2
    query = request.query_params.get('q')
    page = int(request.query_params.get('page', '1'))

    results = search_db(query, page=page)
    return {"results": results, "page": page}

Request Body¶

Access and parse request bodies:

@app.post('/users')
def create_user(request: Request):
    import json

    # Parse JSON body
    data = json.loads(request.body)
    name = data['name']
    email = data['email']

    user = db.create_user(name=name, email=email)
    return {"created": user}, 201

For automatic validation, use @app.validates:

from pydantic import BaseModel

class UserCreate(BaseModel):
    name: str
    email: str

@app.validates
def validate_user(request: Request) -> UserCreate:
    import json
    return UserCreate.model_validate(json.loads(request.body))

@app.post('/users')
def create_user(validate_user: UserCreate):
    # validate_user is already parsed and validated
    return {"created": validate_user.model_dump()}, 201

Request & Response Models¶

Request¶

Request dataclass ¶

Functions¶

__post_init__ ¶

get_accept_header ¶

get_content_type ¶

get_authorization_header ¶

get_if_match ¶

get_if_none_match ¶

get_if_modified_since ¶

get_if_unmodified_since ¶

Response¶

Response dataclass ¶

Functions¶

__post_init__ ¶

set_etag ¶

set_last_modified ¶

is_range_response ¶

generate_etag_from_content ¶

HTTPMethod¶

HTTPMethod ¶

Overview¶

Request Model¶

Response Model¶

HTTPMethod Enum¶

Request Usage¶

Accessing Request Data¶

Creating Requests Manually¶

Response Usage¶

Returning Responses¶

Response Status Codes¶

HTTPMethod Usage¶

Headers¶

Request Headers¶

Response Headers¶

Query Parameters¶

Request Body¶

See Also¶

Request `dataclass` ¶

Response `dataclass` ¶