Warning: This document is for an old version of molten. The latest version is v0.7.3.

API Reference


class molten.App(routes: Optional[List[Union[Route, Include]]] = None, middleware: Optional[List[Callable[[Callable[[...], Any]], Callable[[...], Any]]]] = None, components: Optional[List[molten.dependency_injection.Component[typing.Any][Any]]] = None, parsers: Optional[List[molten.parsers.RequestParser]] = None, renderers: Optional[List[molten.renderers.ResponseRenderer]] = None)[source]

An application that implements the WSGI interface.

  • routes – An optional list of routes to register with the router.
  • middleware – An optional list of middleware. If provided, this replaces the default set of middleware, including the response renderer so make sure to include that in your middleware list.
  • parsers – An optional list of request parsers to use. If provided, this replaces the default list of request parsers.
  • renderers – An optional list of response renderers. If provided, this replaces the default list of response renderers.
handle_404() → molten.http.response.Response

Called whenever a route cannot be found. Dependencies are injected into this just like a normal handler.

handle_415() → molten.http.response.Response

Called whenever a request comes in with an unsupported content type. Dependencies are injected into this just like a normal handler.

handle_exception(exception: BaseException) → molten.http.response.Response

Called whenever an unhandled exception occurs in middleware or a handler. Dependencies are injected into this just like a normal handler.

Parameters:exception – The exception that occurred.
handle_parse_error(exception: molten.errors.ParseError) → molten.http.response.Response

Called whenever a request comes in with a payload that fails to parse. Dependencies are injected into this just like a normal handler.

Parameters:exception – The ParseError that was raised by the request parser on failure.


class molten.Router(routes: Optional[List[Union[Route, Include]]] = None)[source]

A collection of routes.

add_route(route_like: Union[Route, Include], prefix: str = '', namespace: Optional[str] = None) → None[source]

Add a Route to this instance.

add_routes(route_likes: List[Union[Route, Include]], prefix: str = '', namespace: Optional[str] = None) → None[source]

Add a set of routes to this instance.

match(method: str, path: str) → Union[None, Tuple[molten.router.Route, Dict[str, str]]][source]

Look up the route matching the given method and path. Returns the route and any path params that were extracted from the path.

reverse_uri(route_name: str, **params) → str[source]

Build a URI from a Route.

  • route_name – The name of the route to reverse.
  • **params – Route params used to build up the path.
class molten.Route(template: str, handler: Callable[[...], Any], method: str = 'GET', name: Optional[str] = None)[source]

An individual route.


>>> Route("/accounts", list_accounts)
>>> Route("/accounts", create_account, method="POST")
>>> Route("/accounts/{account_id}", get_account)
  • template – A route template.
  • handler – The request handler for this route.
  • method – The request method.
  • name – An optional name for the route. Used in calls to reverse_uri. Defaults to the name of the handler.
class molten.Include(prefix: str, routes: List[Union[Route, Include]], *, namespace: Optional[str] = None)[source]

Groups of routes prefixed by a common path.


>>> Include("/v1/accounts", [
...   Route("/", create_account, method="POST"),
...   Route("/", list_accounts),
...   Route("/{account_id}", get_account),
... ], namespace="accounts")
  • prefix – The path that each route will be prefixed with.
  • routes – The list of routes to include.
  • namespace – An optional prefix that will be prepended to each route’s name. This is useful to avoid conflicts if your handlers have similar names.


class molten.ResponseRendererMiddleware[source]

A middleware that renders responses.

Request Objects

All of the following types, except for UploadedFile can be requested by handlers with DI:

def index(request: Request):
class molten.Request(*, method: str = 'GET', scheme: str = 'http', host: str = '', port: int = 8000, path: str = '/', params: Union[Dict[str, Union[str, List[str]]], molten.http.query_params.QueryParams, None] = None, headers: Union[Dict[str, Union[str, List[str]]], molten.http.headers.Headers, None] = None, body_file: Optional[BinaryIO] = None)[source]

Represents an individual HTTP request.


The HTTP method.


The URL scheme.


The hostname.


The port.


The path.


The query parameters.


The request headers.


A file-like object representing the request body.

classmethod from_environ(environ: Dict[str, Any]) → molten.http.request.Request[source]

Construct a Request object from a WSGI environ.

class molten.QueryParams(mapping: Union[Dict[KT, Union[VT, List[VT]]], Iterable[Tuple[KT, Union[VT, List[VT]]]], None] = None)[source]

A mapping from param names to lists of values. Once constructed, these instances cannot be modified.

classmethod from_environ(environ: Dict[str, Any]) → molten.http.query_params.QueryParams[source]

Construct a QueryParams instance from a WSGI environ.

classmethod parse(query_string: str) → molten.http.query_params.QueryParams[source]

Construct a QueryParams instance from a query string.

get(name: str, default: Optional[str] = None) → Optional[str][source]

Get the last value for a given key.

get_all(name: KT) → List[VT]

Get all the values for a given key.

class molten.Headers(mapping: Optional[Dict[str, Union[str, List[str]]]] = None)[source]

A mapping from case-insensitive header names to lists of values.

classmethod from_environ(environ: Dict[str, Any]) → molten.http.headers.Headers[source]

Construct a Headers instance from a WSGI environ.

add(header: str, value: Union[str, List[str]]) → None[source]

Add values for a particular header.

add_all(mapping: Dict[str, Union[str, List[str]]]) → None[source]

Add a group of headers.

get(header: str, default: Optional[str] = None) → Optional[str][source]

Get the last value for a given header.

get_all(header: str) → List[str][source]

Get all the values for a given header.

get_int(header: str, default: Optional[int] = None) → Optional[int][source]

Get the last value for a given header as an integer.

class molten.Cookies[source]

A dictionary of request cookies.

classmethod parse(cookie_header: str) → molten.http.cookies.Cookies[source]

Turn a cookie header into a Cookies instance.

class molten.UploadedFile(filename: str, headers: molten.http.headers.Headers, stream: BinaryIO)[source]

Represents a file that was uploaded as part of an HTTP request. May be backed by an in-memory file-like object or a real temporary file on disk.


The name the file had in the request.


Headers sent with the file.


The file-like object containing the data.

save(destination: Union[str, BinaryIO]) → None[source]

Save the file’s contents either to another file object or to a path on disk.

Alias Components

All of the following types are convenience aliases for parts of the request:

def index(content_type: Header, x: QueryParam):
molten.Method = <function NewType.<locals>.new_type>
molten.Scheme = <function NewType.<locals>.new_type>
molten.Host = <function NewType.<locals>.new_type>
molten.Port = <function NewType.<locals>.new_type>
molten.QueryString = <function NewType.<locals>.new_type>
molten.QueryParam = <function NewType.<locals>.new_type>
molten.Header = <function NewType.<locals>.new_type>
molten.RequestInput = <function NewType.<locals>.new_type>
molten.RequestBody = <function NewType.<locals>.new_type>
molten.RequestData = <function NewType.<locals>.new_type>

Request Parsers

class molten.RequestParser(*args, **kwargs)[source]

Protocol for request parsers.


Returns a string representing the mime type of the rendered content. This is used to generate OpenAPI documents.

can_parse_content(content_type: str) → bool[source]

Returns True if this parser can parse the given content type.

parse() → Any[source]

Attempt to parse the input data.

Raises:ParseError – if the data cannot be parsed.
class molten.JSONParser[source]

A JSON request parser.

class molten.URLEncodingParser[source]

A parser for urlencoded requests.

class molten.MultiPartParser(*, bufsize: int = 65536, encoding: str = 'utf-8', encoding_errors: str = 'replace', max_field_size: int = 512000, max_file_size: int = 10485760, max_num_fields: int = 100, max_spooled_size: int = 1048576)[source]

A parser for multipart requests. Returns a MultiDict mapping field names to lists of field string values or UploadedFiles.

This is a reasonably simple streaming parser implementation for the multipart/form-data media type. As such, it does not support deprecated parts of RFC7578 like multipart/mixed content and content-transfer-encoding headers.

  • bufsize – The max size of the streaming data buffer. This should be a 32 bit integer that’s a multiple of 4. In some cases, the streaming data buffer may contain double this amount so take that into account when choosing a value. Additionally, the value should be greater than the longest individual header value you want to accept.
  • encoding – The codec to use when decoding form field values.
  • encoding_errors – What to do when an decoding error is encountered.
  • max_field_size – The max number of bytes a field can contain.
  • max_file_size – The max number of bytes a file can contain.
  • max_num_fields – The max number of fields accepted per request.
  • max_spooled_size – The max number of bytes a file in the request can have before it’s written to a temporary file on disk.

Response Objects

class molten.Response(status: str, headers: Union[Dict[str, Union[str, List[str]]], molten.http.headers.Headers, None] = None, content: Optional[str] = None, stream: Optional[BinaryIO] = None, encoding: str = 'utf-8')[source]

An HTTP response.

  • status – The status line of the response.
  • headers – Optional response headers.
  • content – Optional response content as a string.
  • stream – Optional response content as a file-like object.
  • encoding – An optional encoding for the response.
get_content_length() → Optional[int][source]

Compute the content length of this response.

Add a cookie to this response.

class molten.Cookie(name: str, value: str, max_age: Union[int, float, datetime.timedelta, None] = None, expires: Union[int, float, datetime.datetime, None] = None, domain: Optional[str] = None, path: Optional[str] = None, secure: bool = False, http_only: bool = False, same_site: Optional[str] = None)[source]

An individual response cookie.

Raises:ValueError – If the value of same_site is not ‘strict’ or ‘lax’.

Response Renderers

class molten.ResponseRenderer(*args, **kwargs)[source]

Protocol for response renderers.


Returns a string representing the mime type of the rendered content. This is used to generate OpenAPI documents.

can_render_response(accept: str) → bool[source]

Returns True if this renderer can render data for the given mime type.

render(status: str, response_data: Any) → molten.http.response.Response[source]

Attempt to render the response data.

class molten.JSONRenderer[source]

A JSON response renderer.

Dependency Injection

class molten.DependencyInjector(components: List[molten.dependency_injection.Component[typing.Any][Any]], singletons: Optional[Dict[molten.dependency_injection.Component[typing.Any][Any], Any]] = None)[source]

The dependency injector maintains component state and instantiates the resolver.

Parameters:components – The list of components that are used to resolve functions’ dependencies.
get_resolver(instances: Optional[Dict[Any, Any]] = None) → molten.dependency_injection.DependencyResolver[source]

Get the resolver for this Injector.

class molten.DependencyResolver(components: List[molten.dependency_injection.Component[typing.Any][Any]], instances: Dict[molten.dependency_injection.Component[typing.Any][Any], Any])[source]

The resolver does the work of actually filling in all of a function’s dependencies.

add_component(component: molten.dependency_injection.Component[typing.Any][Any]) → None[source]

Add a component to this resolver without adding it to the base dependency injector. This is useful for runtime-built components like RouteParamsComponent.

resolve(fn: Callable[[...], Any], resolving_parameter: Optional[inspect.Parameter] = None) → Callable[[...], Any][source]

Resolve a function’s dependencies.

class molten.Component(*args, **kwargs)[source]

The component protocol.


>>> class DBComponent:
...   is_cacheable = True
...   is_singleton = True
...   def can_handle_parameter(self, parameter: Parameter) -> bool:
...     return parameter.annotation is DB
...   def resolve(self, settings: Settings) -> DB:
...     return DB(settings["database_dsn"])

If True, then the component will be cached within a resolver meaning that instances of the resolved component will be reused within a single request-response cycle. This should be True for most components. Defaults to True.


If True, then the component will be treated as a singleton and cached forever after its first use. Defaults to False.

can_handle_parameter(parameter: inspect.Parameter) → bool[source]

Returns True when parameter represents the desired component.

resolve() → _T[source]

Returns an instance of the component.



molten.schema(cls: Type[_T]) → Type[_T][source]

Construct a schema from a class.

Schemas are plain Python classes with automatically-generated __init__, __eq__ and __repr__ methods. They may be used to validate requests and serialize responses.


>>> @schema
... class Account:
...   username: str
...   password: str = Field(request_only=True)
...   is_admin: bool = Field(response_only=True, default=False)
>>> load_schema(Account, {})
Traceback (most recent call last):
ValidationError: {'username': 'this field is required', 'password': 'this field is required'}
>>> load_schema(Account, {"username": "example", "password": "secret"})
Account(username='example', password='secret', is_admin=False)
>>> dump_schema(load_schema(Account, {"username": "example", "password": "secret"}))
{'username': 'example', 'is_admin': False}
Raises:RuntimeError – When the attributes are invalid.
molten.is_schema(ob: Type[Any]) → bool[source]

Returns True if the given type is a schema.

molten.dump_schema(ob: Any, *, sparse: bool = False) → Dict[str, Any][source]

Convert a schema instance into a dictionary.


TypeError – If ob is not a schema instance.

  • ob – An instance of a schema.
  • sparse – If true, fields whose values are None are going to be dropped from the output.
molten.load_schema(schema: Type[_T], data: Dict[str, Any]) → _T[source]

Validate the given data dictionary against a schema and instantiate the schema.


ValidationError – When the input data is not valid.

  • schema – The schema class to validate the data against.
  • data – Data to validate against and populate the schema with.

Fields and Validators

class molten.Field(name: Optional[str] = None, annotation: Optional[Type[_T]] = None, description: Optional[str] = None, default: Union[_T, molten.validation.common._Missing] = Missing, default_factory: Optional[Callable[[], _T]] = None, request_name: Optional[str] = None, response_name: Optional[str] = None, request_only: bool = False, response_only: bool = False, allow_coerce: bool = False, validator: Optional[molten.validation.field.Validator[~_T][_T]] = None, **validator_options)[source]

An individual field on a schema. The @schema decorator automatically turns annotated attributes into fields, but the field class can also be used to enrich annotated attributes with metadata.


>>> @schema
... class Application:
...   name: str
...   rating: int = Field(minimum=1, maximum=5)
  • name – The name of the field. Automatically populated by the schema decorator.
  • annotation – The field’s annotation. Like name, this is automatically populated by @schema.
  • description – An optional description for the field.
  • default – An optional default value for the field.
  • default_factory – An optional default function for the field.
  • request_name – The field’s name within a request. This is the same as the field’s name by default.
  • response_name – The field’s name within a response. This is the same as the field’s name by default.
  • request_only – Whether or not to exclude this field from responses. Defaults to False.
  • response_only – Whether or not to ignore this field when loading requests. Defaults to False.
  • allow_coerce – Whether or not values passed to this field may be coerced to the correct type. Defaults to False.
  • validator – The validator to use when loading data. The schema decorator will automatically pick a validator for builtin types.
  • **validator_options – Arbitrary options passed to the field’s validator.
select_validator() → None[source]

Find a suitable Validator for this field.


Returns True if the field has either a default value or a default factory.

validate(value: Optional[Any]) → _T[source]

Validate and possibly transform the given value.

Raises:FieldValidationError – When the value is not valid.
class molten.validation.field.Validator(*args, **kwargs)[source]

Validators ensure that values conform to arbitrary specifications.

class molten.validation.field.NumberValidator[source]

Validates numbers.

class molten.validation.field.StringValidator[source]

Validates strings.

class molten.validation.field.ListValidator[source]

Validates lists.

When a generic parameter is provided, then the values will be validated against that annotation:

>>> @schema
... class Setting:
...   name: str
...   value: str

>>> @schema
... class Account:
...   settings: List[Setting]

>>> load_schema(Account, {"settings": [{"name": "a", "value": "b"}]})
Account(settings=[Setting(name="a", value="b")])

>>> load_schema(Account, {"settings": [{"name": "a"}]})
Traceback (most recent call last):
ValidationError: {"settings": {0: {"value": "this field is required"}}}

When a generic parameter isn’t provided, then any list is accepted.

class molten.validation.field.DictValidator[source]

Validates dictionaries.

When the fields option is provided, only the declared fields are going to be extracted from the input and will be validated.

>>> @schema
... class Account:
...   settings: Dict[str, str] = Field(fields={
...     "a": Field(annotation=str),
...   })
>>> load_schema(Account, {"settings": {}})
>>> load_schema(Account, {"settings": {"a": "b", "c": "d"}})
Account(settings={"settings" {"a": "b"}})
>>> load_schema(Account, {"settings": {"a": 42}})
Traceback (most recent call last):
ValidationError: {"settings": {"a": "unexpected type int"}}

When the fields option is not provided and the annotation has generic parameters, then the items from the input will be validated against the generic parameter annotations:

>>> @schema
... class Account:
...   settings: Dict[str, str]

>>> load_schema(Account, {"settings": {}})

>>> load_schema(Account, {"settings": {"a": "b"}})
Account(settings={"a": "b"})

>>> load_schema(Account, {"settings": {"a": 42})  # invalid
Traceback (most recent call last):
ValidationError: {"settings": {"a": "unexpected type int"}}

When neither fields or generic parameters are provided, then any dictionary will be accepted.

class molten.validation.field.SchemaValidator[source]

Validates dictionaries against schema classes.


class molten.TestClient(app: molten.app.BaseApp)[source]

Test clients are used to simulate requests against an application instance.

request(method: str, path: str, headers: Union[Dict[str, Union[str, List[str]]], molten.http.headers.Headers, None] = None, params: Union[Dict[str, Union[str, List[str]]], molten.http.query_params.QueryParams, None] = None, body: Optional[bytes] = None, data: Optional[Dict[str, str]] = None, files: Optional[Dict[str, Tuple[str, BinaryIO]]] = None, json: Optional[Any] = None, auth: Optional[Callable[[molten.http.request.Request], molten.http.request.Request]] = None, prepare_environ: Optional[Callable[[Dict[str, Any]], Dict[str, Any]]] = None) → molten.testing.client.TestResponse[source]

Simulate a request against the application.


RuntimeError – If both ‘data’ and ‘json’ are provided.

  • method – The request method.
  • path – The request path.
  • headers – Optional request headers.
  • params – Optional query params.
  • body – An optional bytestring for the request body.
  • data – An optional dictionary for the request body that gets url-encoded.
  • files – An optional dictionary of files to upload as part of a multipart request.
  • json – An optional value for the request body that gets json-encoded.
  • auth – An optional function that can be used to add auth headers to the request.
class molten.TestResponse(response: molten.http.response.Response)[source]

A wrapper around Response objects that adds a few additional helper methods for testing.


The response status line.


The response status code as an integer.


The response headers.


The response data as a binary file.


The response data as a string.


Rewinds the output stream and returns all its data.


Returns the HTTP status code as an integer.

json() → Any[source]

Convert the response data to JSON.


class molten.MoltenError(message: str)[source]

Base class for all Molten exceptions.

class molten.DIError(message: str)[source]

Raised when a dependency cannot be resolved.

class molten.HTTPError(status: str, response: Any, headers: Any = None)[source]

Base class for HTTP errors. Handlers and middleware can raise these to short-circuit execution.

class molten.RouteNotFound(message: str)[source]

Raised when trying to reverse route to a route that doesn’t exist.

class molten.RouteParamMissing(message: str)[source]

Raised when a param is missing while reversing a route.

class molten.RequestParserNotAvailable(message: str)[source]

Raised when no request parser can handle the incoming request.

class molten.ParseError(message: str)[source]

Raised by parsers when the input data cannot be parsed.

class molten.FieldTooLarge(message: str)[source]

Raised by MultiPartParser when a field exceeds the maximum field size limit.

class molten.FileTooLarge(message: str)[source]

Raised by MultiPartParser when a file exceeds the maximum file size limit.

class molten.TooManyFields(message: str)[source]

Raised by MultiPartParser when the input contains too many fields.

class molten.HeaderMissing(message: str)[source]

Raised by Headers.__getitem__ when a header does not exist.

class molten.ParamMissing(message: str)[source]

Raised by QueryParams.__getitem__ when a param is missing.

class molten.ValidationError(reasons: Dict[str, Any])[source]

Raised by validator.load when the input data is invalid.

class molten.FieldValidationError(message: str)[source]

Raised by Field.validate when a given value is invalid.



The SQLAlchemyComponent automatically creates database session objects whenever a handler requests a parameter whose type is sqlalchemy.Session.

This component depends on the SettingsComponent.

class molten.contrib.sqlalchemy.SQLAlchemyEngineComponent[source]

A component that sets up an SQLAlchemy Engine. This component depends on the availability of a molten.Settings component.

Your settings dictionary must contain a database_engine_dsn setting pointing at the database to use. Additionally, you may provide a database_engine_params setting representing dictionary data that will be passed directly to sqlalchemy.create_engine.


>>> from molten import App
>>> from molten.contrib.sqlalchemy import SQLAlchemyEngineComponent, SQLAlchemySessionComponent, SQLAlchemyMiddleware
>>> from molten.contrib.toml_settings import TOMLSettingsComponent
>>> app = App(
...   components=[
...     TOMLSettingsComponent(),
...     SQLAlchemyEngineComponent(),
...     SQLAlchemySessionComponent(),
...   ],
...   middleware=[SQLAlchemyMiddleware()],
... )
class molten.contrib.sqlalchemy.SQLAlchemySessionComponent[source]

A component that creates and injects SQLAlchemy sessions.


>>> def find_todos(session: Session) -> List[Todo]:
...   todos = session.query(TodoModel).all()
...   ...
class molten.contrib.sqlalchemy.SQLAlchemyMiddleware[source]

A middleware that automatically commits SQLAlchemy sessions on handler success and automatically rolls back sessions on handler failure.

Sessions are only instantiated and operated upon if the handler or any other middleware has requested an SQLAlchemy session object via DI. This means that handlers that don’t request a Session object don’t automatically connect to the Database.

If you need access to the SQLAlchemy engine instance, you may request EngineData in your function.

molten.contrib.sqlalchemy.EngineData = <class 'molten.contrib.sqlalchemy.EngineData'>

A named tuple containing an instantiated SQLAlchemy engine object and the session_factory.


The SettingsComponent loads environment-specific settings from a TOML config file. You’ll have to install the toml package yourself before using this module.


The TemplatesComponent renders templates using jinja2. You’ll have to install the jinja2 package yourself before using this module.

class molten.contrib.templates.Templates(path: str)[source]

Renders jinja2 templates.

render(template_name: str, **context) → molten.http.response.Response[source]

Find a template and render it.

  • template_name – The name of the template to render.
  • **context – Bindings passed to the template.
class molten.contrib.templates.TemplatesComponent(path: str)[source]

A component that builds a jinja2 template renderer.

Parameters:path – The path to a folder containing your templates.