Application¶

RestApplication ¶

RestApplication()

Main application class for the REST framework.

Functions¶

get ¶

get(path: str)

Decorator to register a GET route handler on the root router.

post ¶

post(path: str)

Decorator to register a POST route handler on the root router.

put ¶

put(path: str)

Decorator to register a PUT route handler on the root router.

patch ¶

patch(path: str)

Decorator to register a PATCH route handler on the root router.

delete ¶

delete(path: str)

Decorator to register a DELETE route handler on the root router.

options ¶

options(path: str)

Decorator to register an OPTIONS route handler on the root router.

dependency ¶

dependency(name: Optional[str] = None, scope: DependencyScope = 'request')

Decorator to register a dependency provider.

Parameters:

Name	Type	Description	Default
`name`	`Optional[str]`	Optional name for the dependency. If not provided, uses the function name.	`None`
`scope`	`DependencyScope`	Dependency scope - "request" (default) or "session". - "request": Cached per request, cleared between requests - "session": Cached across all requests, never cleared automatically	`'request'`

validates ¶

validates(func: Callable, scope: DependencyScope = 'request')

Decorator to mark a function as returning a validated Pydantic model.

Parameters:

Name	Type	Description	Default
`scope`	`DependencyScope`	Dependency scope - "request" (default) or "session"	`'request'`

on_startup ¶

on_startup(func: Optional[Callable] = None)

Register a startup handler to run when the application starts.

Can be used as a decorator::

@app.on_startup
async def database():
    print("Opening database connection...")
    return create_db_connection()

@app.get("/users")
def get_users(database):  # database from startup is injected here
    return database.query("SELECT * FROM users")

Or called directly::

app.on_startup(my_startup_function)

Startup handlers are automatically registered as session-scoped dependencies, so their return values can be injected into route handlers and other dependencies. Handlers can be sync or async functions.

on_shutdown ¶

on_shutdown(func: Optional[Callable] = None)

Register a shutdown handler to run when the application stops.

Can be used as a decorator::

@app.on_shutdown
async def shutdown():
    print("Application shutting down...")
    # Close connections, cleanup resources, etc.

Or called directly::

app.on_shutdown(my_shutdown_function)

Handlers can be sync or async functions.

mount ¶

mount(prefix: str, router: Router)

Mount a router with a given prefix.

Parameters:

Name	Type	Description	Default
`prefix`	`str`	The path prefix for all routes in the mounted router	required
`router`	`Router`	The router to mount	required

Example

app = RestApplication() users_router = Router() users_router.get("/")(lambda: {"users": []}) users_router.get("/{id}")(lambda id: {"user": id})

app.mount("/users", users_router)

This creates routes: GET /users/ and GET /users/{id}¶

execute ¶

execute(request: Request) -> Response

Execute a request through the state machine.

startup_sync ¶

startup_sync()

Synchronous wrapper for startup().

This is called automatically by the AWS Lambda adapter during cold start to execute startup handlers in a synchronous context (module-level initialization).

Uses anyio.run() to execute the async startup() method synchronously.

shutdown_sync ¶

shutdown_sync()

Synchronous wrapper for shutdown().

This can be called by AWS Lambda Extensions or other synchronous shutdown hooks to execute shutdown handlers in a synchronous context.

Uses anyio.run() to execute the async shutdown() method synchronously.

Overview¶

RestApplication is the main entry point for building REST APIs with RestMachine. It provides:

Route registration - Flask-style decorators for HTTP methods
Dependency injection - pytest-style parameter injection
Content negotiation - Automatic format selection based on Accept headers
Validation - Optional Pydantic integration for request/response validation
Lifecycle hooks - Startup and shutdown handlers
Router mounting - Compose applications from multiple routers

Quick Start¶

from restmachine import RestApplication

app = RestApplication()

@app.get('/hello/{name}')
def hello(path_params):
    name = path_params['name']
    return {"message": f"Hello, {name}!"}

Route Registration¶

HTTP Method Decorators¶

Register routes using HTTP method decorators:

@app.get('/users')
def list_users():
    return {"users": [...]}

@app.post('/users')
def create_user(json_body):
    # json_body automatically injected for JSON requests
    return {"created": True}, 201

@app.put('/users/{user_id}')
def update_user(path_params, json_body):
    user_id = path_params['user_id']
    return {"updated": user_id}

@app.delete('/users/{user_id}')
def delete_user(path_params):
    user_id = path_params['user_id']
    return {"deleted": user_id}, 204

Path Parameters¶

Use {param} syntax for path parameters, accessed via the path_params dependency:

@app.get('/users/{user_id}/posts/{post_id}')
def get_post(path_params):
    user_id = path_params['user_id']
    post_id = path_params['post_id']
    return {"user_id": user_id, "post_id": post_id}

Generic Route Registration¶

Use @app.route() for custom methods or multiple methods:

@app.route('PATCH', '/users/{user_id}')
def patch_user(request):
    return {"patched": True}

Dependency Injection¶

Share resources across handlers using pytest-style dependency injection:

@app.on_startup
def database():
    """Create database connection at startup."""
    return create_db_connection()

@app.resource_exists
def user(database, path_params):
    """Check if user exists and return it."""
    user_id = path_params.get('user_id')
    return database.get_user(user_id)  # Returns None if not found

@app.get('/users/{user_id}')
def get_user(user):
    """User is injected, 404 handled automatically if None."""
    return user

@app.get('/posts')
def list_posts(database):
    """Database instance is reused across all requests."""
    return {"posts": database.query("SELECT * FROM posts")}

Startup dependencies persist across all requests. See Dependency Injection Guide for details.

Request Validation¶

Validate request bodies using Pydantic with the @app.validates decorator:

from pydantic import BaseModel

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

@app.validates
def validate_user(json_body) -> UserCreate:
    """Validates JSON body, returns 422 on validation error."""
    return UserCreate.model_validate(json_body)

@app.post('/users')
def create_user(validate_user: UserCreate):
    """Validated model is injected, errors handled automatically."""
    return {"created": validate_user.model_dump()}, 201

Content Negotiation¶

Serve multiple formats using the @app.provides decorator:

@app.get('/data')
def get_data():
    return {"key": "value"}

@app.provides("text/html")
def render_html(get_data):
    data = get_data
    return f"<div>{data}</div>"

@app.provides("application/xml")
def render_xml(get_data):
    data = get_data
    return f"<result>{data}</result>"
# Returns JSON, HTML, or XML based on Accept header

Built-in renderers: JSON (default), HTML, Plain Text. See Content Negotiation Guide for details.

Error Handling¶

Register custom error handlers:

@app.error_handler(404)
def not_found_handler(error):
    """Handle 404 errors."""
    return {"error": "Not Found", "path": error.get("path")}, 404

@app.error_handler(500)
def server_error_handler(error):
    """Handle 500 errors."""
    return {"error": "Internal Server Error"}, 500

@app.error_handler()  # Default handler for all errors
def default_error_handler(error):
    """Fallback for unhandled errors."""
    return {"error": "Something went wrong"}, 500

Lifecycle Handlers¶

Manage application startup and shutdown:

@app.on_startup
def init_database():
    """Runs once at startup."""
    print("Initializing database...")
    return create_db_connection()

@app.on_shutdown
def close_database(init_database):
    """Runs at shutdown, can inject startup dependencies."""
    print("Closing database...")
    init_database.close()

@app.on_startup(scope=DependencyScope.SESSION)
def load_config():
    """SESSION scope - persists across requests."""
    return load_app_config()

See Lifecycle Handlers Guide for details.

Mounting Routers¶

Compose applications from multiple routers:

from restmachine import Router

api_v1 = Router()

@api_v1.get('/users')
def list_users():
    return {"users": [...]}

# Mount at /api/v1
app.mount('/api/v1', api_v1)

# Access: GET /api/v1/users

See Router API for details.

Executing Requests¶

Execute requests directly (useful for testing):

from restmachine import Request, HTTPMethod

request = Request(
    method=HTTPMethod.GET,
    path="/users/123",
    headers={},
    query_params={},
    body=""
)

response = app.execute(request)
print(response.status_code, response.body)

Deployment¶

ASGI Servers¶

from restmachine import ASGIAdapter

asgi_app = ASGIAdapter(app)

# Run with uvicorn
# uvicorn app:asgi_app --reload

AWS Lambda¶

from restmachine_aws import AwsApiGatewayAdapter

adapter = AwsApiGatewayAdapter(app)

def lambda_handler(event, context):
    return adapter.handle_event(event, context)

Configuration¶

Constructor Options¶

app = RestApplication(
    # Add custom configuration here if needed
)

Application¶

RestApplication ¶

Functions¶

get ¶

post ¶

put ¶

patch ¶

delete ¶

options ¶

dependency ¶

validates ¶

on_startup ¶

on_shutdown ¶

mount ¶

This creates routes: GET /users/ and GET /users/{id}¶

execute ¶

startup_sync ¶

shutdown_sync ¶

Overview¶

Quick Start¶

Route Registration¶

HTTP Method Decorators¶

Path Parameters¶

Generic Route Registration¶

Dependency Injection¶

Request Validation¶

Content Negotiation¶

Error Handling¶

Lifecycle Handlers¶

Mounting Routers¶

Executing Requests¶

Deployment¶

ASGI Servers¶

AWS Lambda¶

Configuration¶

Constructor Options¶

See Also¶