API Reference

Apps

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)

An application that implements the WSGI interface.

Parameters:

• 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.

Routing

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

A collection of routes.

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

Add a Route to this instance.

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

Add a set of routes to this instance.

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

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

Build a URI from a Route.

Raises:

• RouteNotFound – When the route doesn’t exist.
• RouteParamMissing – When a required parameter was not provided.

Parameters:

• 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)

An individual route.

Examples

>>> Route("/accounts", list_accounts)
>>> Route("/accounts", create_account, method="POST")
>>> Route("/accounts/{account_id}", get_account)

Parameters:

• 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)

Groups of routes prefixed by a common path.

Examples

>>> Include("/v1/accounts", [
...   Route("/", create_account, method="POST"),
...   Route("/", list_accounts),
...   Route("/{account_id}", get_account),
... ], namespace="accounts")

Parameters:

• 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.

molten.annotate(**options) → Callable[[...], Any]

Add arbitrary attributes to a callable.

Examples

>>> @annotate(openapi_tags=["a", "b"])
... def some_handler():
...   ...

>>> some_handler.openapi_tags
["a", "b"]

Middleware

class molten.ResponseRendererMiddleware

A middleware that renders responses.

Settings

class molten.Settings

A dictionary of settings.

deep_get(path: str, default: Optional[Any] = None) → Optional[Any]

Look up a deeply-nested setting by its path.

Examples

>>> settings = Settings({"a": {"b": [{"c": 42}]}})
>>> settings.deep_get("a.b.0.c")
42

Raises:

TypeError – When attempting to index into a primitive value or when indexing a list with a string value rather than an integer.

Parameters:

• path – A dot-separated string representing the path to the value.
• default – The value to return if the path cannot be traversed.

strict_get(path: str) → Any

Get a required setting.

Raises:RuntimeError – If the value for that setting cannot be found.

class molten.SettingsComponent(settings: molten.settings.Settings)

A component for settings that you build at app init time.

Examples

>>> settings = Settings({"database_engine_dsn": "sqlite://"})
>>> app = App(components=[SettingsComponent(settings)])

Request Objects

class molten.Request(*, method: str = 'GET', scheme: str = 'http', host: str = '127.0.0.1', 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)

Represents an individual HTTP request.

method

The HTTP method.

scheme

The URL scheme.

host

The hostname.

port

The port.

path

The path.

params

The query parameters.

headers

The request headers.

body_file

A file-like object representing the request body.

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

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)

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

Construct a QueryParams instance from a WSGI environ.

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

Construct a QueryParams instance from a query string.

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

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)

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

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

Construct a Headers instance from a WSGI environ.

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

Add values for a particular header.

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

Add a group of headers.

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

Get the last value for a given header.

get_all(header: str) → List[str]

Get all the values for a given header.

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

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

class molten.Cookies

A dictionary of request cookies.

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

Turn a cookie header into a Cookies instance.

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

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.

filename

The name the file had in the request.

headers

Headers sent with the file.

stream

The file-like object containing the data.

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

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)

Protocol for request parsers.

mime_type

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

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

parse() → Any

Attempt to parse the input data.

Raises:ParseError – if the data cannot be parsed.

class molten.JSONParser

A JSON request parser.

class molten.URLEncodingParser

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)

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.

Parameters:

• 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')

An HTTP response.

Parameters:

• 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]

Compute the content length of this response.

set_cookie(cookie: molten.http.cookies.Cookie) → None

Add a cookie to this response.

class molten.StreamingResponse(status: str, content: Generator[bytes, None, None], headers: Union[Dict[str, Union[str, List[str]]], molten.http.headers.Headers, None] = None, encoding: str = 'utf-8')

A chunked HTTP response, yielding content from a generator.

Parameters:

• status – The status line of the response.
• content – A response content generator.
• headers – Optional response headers.
• encoding – An optional encoding for the response.

get_content_length() → Optional[int]

Compute the content length of this response. Streaming responses can’t know their length up front so this always returns None.

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)

An individual response cookie.

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

Response Renderers

class molten.ResponseRenderer(*args, **kwargs)

Protocol for response renderers.

mime_type

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

can_render_response(accept: str) → bool

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

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

Attempt to render the response data.

class molten.JSONRenderer

A JSON response renderer.

HTTP Status Lines

# 1xx
HTTP_100 = "100 Continue"
HTTP_101 = "101 Switching Protocols"
HTTP_102 = "102 Processing"

# 2xx
HTTP_200 = "200 OK"
HTTP_201 = "201 Created"
HTTP_202 = "202 Accepted"
HTTP_203 = "203 Non-Authoritative Information"
HTTP_204 = "204 No Content"
HTTP_205 = "205 Reset Content"
HTTP_206 = "206 Partial Content"
HTTP_207 = "207 Multi-Status"
HTTP_208 = "208 Already Reported"

# 3xx
HTTP_300 = "300 Multiple Choices"
HTTP_301 = "301 Moved Permanently"
HTTP_302 = "302 Found"
HTTP_303 = "303 See Other"
HTTP_304 = "304 Not Modified"
HTTP_305 = "305 Use Proxy"
HTTP_307 = "307 Temporary Redirect"
HTTP_308 = "308 Permanent Redirect"

# 4xx
HTTP_400 = "400 Bad Request"
HTTP_401 = "401 Unauthorized"
HTTP_402 = "402 Payment Required"
HTTP_403 = "403 Forbidden"
HTTP_404 = "404 Not Found"
HTTP_405 = "405 Method Not Allowed"
HTTP_406 = "406 Not Acceptable"
HTTP_407 = "407 Proxy Authentication Required"
HTTP_408 = "408 Request Timeout"
HTTP_409 = "409 Conflict"
HTTP_410 = "410 Gone"
HTTP_411 = "411 Length Required"
HTTP_412 = "412 Precondition Failed"
HTTP_413 = "413 Payload Too Large"
HTTP_414 = "414 Request-URI Too Long"
HTTP_415 = "415 Unsupported Media Type"
HTTP_416 = "416 Requested Range Not Satisfiable"
HTTP_417 = "417 Expectation Failed"
HTTP_418 = "418 I'm a teapot"
HTTP_421 = "421 Misdirected Request"
HTTP_422 = "422 Unprocessable Entity"
HTTP_423 = "423 Locked"
HTTP_424 = "424 Failed Dependency"
HTTP_426 = "426 Upgrade Required"
HTTP_428 = "428 Precondition Required"
HTTP_429 = "429 Too Many Requests"
HTTP_431 = "431 Request Header Fields Too Large"
HTTP_444 = "444 Connection Closed Without Response"
HTTP_451 = "451 Unavailable For Legal Reasons"
HTTP_499 = "499 Client Closed Request"
HTTP_415 = "415 Unsupported Media Type"

# 5xx
HTTP_500 = "500 Internal Server Error"
HTTP_501 = "501 Not Implmeneted"
HTTP_502 = "502 Bad Gateway"
HTTP_503 = "503 Service Unavailable"
HTTP_504 = "504 Gateway Timeout"
HTTP_505 = "505 HTTP Version Not Supported"
HTTP_506 = "506 Variant Also Negotiates"
HTTP_507 = "507 Insufficient Storage"
HTTP_508 = "508 Loop Detected"
HTTP_510 = "510 Not Extended"
HTTP_511 = "511 Network Authentication Required"
HTTP_599 = "599 Network Connect Timeout Error"

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)

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

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])

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

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]

Resolve a function’s dependencies.

class molten.Component(*args, **kwargs)

The component protocol.

Examples

>>> 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"])

is_cacheable

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.

is_singleton

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

Returns True when parameter represents the desired component.

resolve() → _T

Returns an instance of the component.

Validation

Schemas

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

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.

Examples

>>> @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

Returns True if the given type is a schema.

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

Convert a schema instance into a dictionary.

Raises:

TypeError – If ob is not a schema instance.

Parameters:

• 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

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

Raises:

ValidationError – When the input data is not valid.

Parameters:

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

molten.forward_ref(name: str) → Any

Generate a proxy type for a schema that is going to be defined at some point after the current statement.

Examples

Schema “A” includes a reference to a yet-to-be-defined schema “B”:

@schema
class A:
    b: forward_ref("B")

@schema
class B:
    x: int

forward_ref can be used as a type parameter to other types as well:

@schema
class A:
    b: List[forward_ref("B")]

@schema
class B:
    x: int

Parameters:name – The name of the schema being referenced. This name must eventually be bound in the scope of the call site.

Fields and Validators

molten.field(*args, **kwargs) → Any

An alias for Field that tricks the type system into submission.

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)

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.

Examples

>>> @schema
... class Application:
...   name: str
...   rating: int = Field(minimum=1, maximum=5)

Parameters:

• 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

Find a suitable Validator for this field.

has_default

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

validate(value: Optional[Any]) → _T

Validate and possibly transform the given value.

Raises:FieldValidationError – When the value is not valid.

class molten.validation.field.Validator(*args, **kwargs)

Validators ensure that values conform to arbitrary specifications.

class molten.validation.field.NumberValidator

Validates numbers.

class molten.validation.field.StringValidator

Validates strings.

class molten.validation.field.ListValidator

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

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": {}})
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": {}})
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

Validates dictionaries against schema classes.

Helpers

The helpers module contains a collection of functions that are useful for general purpose applications.

class molten.helpers.RedirectType

An enumeration.

molten.helpers.redirect(target_location: str, *, redirect_type: molten.helpers.RedirectType = <RedirectType.TEMPORARY: 1>, use_modern_codes: bool = True) → molten.http.response.Response

Construct an HTTP Response to redirect the client elsewhere.

Parameters:

• target_location – Where the client should be redirected to.
• redirect_type – PERMANENT or TEMPORARY.
• use_modern_codes – Whether or not to use HTTP/1.1 response codes. The advantage to using HTTP/1.1 codes is the request method is preserved during redirect, but older clients (IE11 and older) might not support them.

Testing

class molten.TestClient(app: molten.app.BaseApp)

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, Union[Tuple[str, BinaryIO], Tuple[str, 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

Simulate a request against the application.

Raises:

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

Parameters:

• 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)

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

status

The response status line.

status_code

The response status code as an integer.

headers

The response headers.

stream

The response data as a binary file.

data

The response data as a string.

data

Rewinds the output stream and returns all its data.

status_code

Returns the HTTP status code as an integer.

json() → Any

Convert the response data to JSON.

Errors

class molten.MoltenError(message: str)

Base class for all Molten exceptions.

class molten.DIError(message: str)

Raised when a dependency cannot be resolved.

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

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

class molten.RouteNotFound(message: str)

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

class molten.RouteParamMissing(message: str)

Raised when a param is missing while reversing a route.

class molten.RequestParserNotAvailable(message: str)

Raised when no request parser can handle the incoming request.

class molten.ParseError(message: str)

Raised by parsers when the input data cannot be parsed.

class molten.FieldTooLarge(message: str)

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

class molten.FileTooLarge(message: str)

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

class molten.TooManyFields(message: str)

Raised by MultiPartParser when the input contains too many fields.

class molten.HeaderMissing(message: str)

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

class molten.ParamMissing(message: str)

Raised by QueryParams.__getitem__ when a param is missing.

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

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

class molten.FieldValidationError(message: str)

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

Contrib

Dramatiq

Dramatiq support for molten.

This functionality requires the dramatiq package.

molten.contrib.dramatiq.setup_dramatiq(app: molten.app.BaseApp) → None

Sets up the global state required to be able to inject components into Dramatiq actors.

Examples

>>> from molten.contrib.dramatiq import setup_dramatiq

>>> # All components that were registered with your app will be
>>> # available to your actors once you call this function.
>>> setup_dramatiq(app)

molten.contrib.dramatiq.actor(fn=None, **kwargs)

Use this in place of dramatiq.actor in order to create actors that can request components via dependency injection. This is just a wrapper around dramatiq.actor and it takes the same set of parameters.

Examples

>>> from molten.contrib.dramatiq import actor

>>> @actor(queue_name="example")
... def add(x, y, database: Database) -> None:
...   database.put(x + y)
...
>>> add.send(1, 2)

msgpack

A parser and a renderer for msgpack data.

This functionality requires the msgpack package to be installed.

class molten.contrib.msgpack.MsgpackParser

A msgpack request parser.

class molten.contrib.msgpack.MsgpackRenderer

A msgpack response renderer.

Prometheus

Prometheus metrics support.

This functionality requires the prometheus-client package to be installed.

molten.contrib.prometheus.expose_metrics() → molten.http.response.Response

Expose prometheus metrics from the current process.

molten.contrib.prometheus.expose_metrics_multiprocess() → molten.http.response.Response

Expose prometheus metrics from the current set of processes. Use this instead of expose_metrics if you’re using a multi-process server.

molten.contrib.prometheus.prometheus_middleware(handler: Callable[[...], Any]) → Callable[[...], Any]

Collect prometheus metrics from your handlers.

Request Id

Request Id support for molten.

molten.contrib.request_id.get_request_id() → Optional[str]

Retrieves the request id for the current thread.

molten.contrib.request_id.set_request_id(request_id: Optional[str]) → None

Set a request id for the current thread. If request_id is None, then a random id will be generated.

class molten.contrib.request_id.RequestIdFilter(name='')

Adds the current request id to log records, making it possible to log request ids via the standard logging module.

Example logging configuration:

import logging.config
logging.config.dictConfig({
    "version": 1,
    "filters": {
        "request_id": {
            "()": "molten.contrib.request_id.RequestIdFilter"
        },
    },
    "formatters": {
        "standard": {
            "format": "%(levelname)-8s [%(asctime)s] [%(request_id)s] %(name)s: %(message)s"
        },
    },
    "handlers": {
        "console": {
            "level": "DEBUG",
            "class": "logging.StreamHandler",
            "filters": ["request_id"],
            "formatter": "standard",
        },
    },
    "loggers": {
        "myapp": {
            "handlers": ["console"],
            "level": "DEBUG",
            "propagate": False,
        },
    }
})

class molten.contrib.request_id.RequestIdMiddleware

Adds an x-request-id to responses containing a unique request id value. If the incoming request has an x-request-id header then that value is reused for the response. This makes it easy to trace requests within a microservice architecture.

Sessions

Session support for molten. Good APIs are stateless, but sometimes you may need something like this for a one-off part of your app.

class molten.contrib.sessions.Session(id: str, **data)

Session objects are ordinary dictionaries that are guaranteed to be constructed with an “id” key.

class molten.contrib.sessions.SessionComponent(store: molten.contrib.sessions.SessionStore)

A component that loads Session objects from the request.

Parameters:store – A session store.

class molten.contrib.sessions.SessionMiddleware(store: molten.contrib.sessions.SessionStore)

A middleware that dumps Session data into the response.

Parameters:store – A session store.

Session Stores

Session stores determine where and how session data is stored. molten comes with a stateless session store that’s based on cookies by default, but you can implement your own by implementing the SessionStore protocol.

class molten.contrib.sessions.SessionStore(*args, **kwargs)

Protocol for session stores.

dump() → molten.http.cookies.Cookie

Convert the session to a Cookie and possibly store it somewhere (eg. memcached, Redis).

This method may request components via DI.

load() → molten.contrib.sessions.Session

Load a session from the request.

This method may request components via DI.

class molten.contrib.sessions.CookieStore(signing_key: Union[bytes, str], *, signing_method: str = 'sha256', cookie_ttl: int = 604800, cookie_name: str = '__sess__', cookie_domain: Optional[str] = None, cookie_path: Optional[str] = None, cookie_secure: bool = False)

A stateless session store based on cookies. Sessions are converted to JSON and then base64-encoded. The values are signed with a signing key and validated when the sessions are subsequently loaded.

An expiration time is inserted into the session before it’s dumped so as to provide minimal protection against session replay attacks.

Warning

Don’t store sensitive information in sessions using this store. They are tamper-proof, but users can decode them.

SQLAlchemy

The SQLAlchemyComponent automatically creates database session objects whenever a handler requests a parameter whose type is sqlalchemy.Session. You need to install the sqlalchemy package before you can use this package.

This component requires a Settings component.

class molten.contrib.sqlalchemy.SQLAlchemyEngineComponent

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.

Examples

>>> 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

A component that creates and injects SQLAlchemy sessions.

Examples

>>> def find_todos(session: Session) -> List[Todo]:
...   todos = session.query(TodoModel).all()
...   ...

class molten.contrib.sqlalchemy.SQLAlchemyMiddleware

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.

TOML Settings

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

class molten.contrib.toml_settings.TOMLSettings

A dictionary of settings parsed from a TOML file.

classmethod from_path(path: str, environment: str) → molten.settings.Settings

Load a TOML file into a dictionary.

Raises:

FileNotFoundError – When the settings file does not exist.

Parameters:

• path – The path to the TOML file containing your settings.
• environment – The config environment to use.

class molten.contrib.toml_settings.TOMLSettingsComponent(path: str = './settings.toml', environment: Optional[str] = None)

A component that loads settings from a TOML file.

The settings file should have a “common” section and one section for each environment the application is expected to run in. The environment-specific settings are merged on top of the “common” settings on load.

The current environment is determined by the ENVIRONMENT config variable.

Example settings file:

[common]
coon_uri = "sqlite:///"
conn_pooling = true
conn_pool_size = 1

[dev]

[prod]
# conn_uri is loaded from the DATABASE_URL environment variable.
# This is not a standard TOML feature, but is provided for
# convenience.  This uses the built-in `string.Template` parser.
conn_uri = "$DATABASE_URL"
con_pool_size = 32

Examples:

from molten import Settings

def handle(settings: Settings):
  settings.get("conn_pooling")

Parameters:

• path – The path to the TOML file containing your settings.
• environment – The config environment to use. If not provided, this defaults to the value of the “ENVIRONMENT” environment variable. If that’s not set either, then this defaults to “dev”.

Templates

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)

Renders jinja2 templates.

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

Find a template and render it.

Parameters:

• template_name – The name of the template to render.
• **context – Bindings passed to the template.

class molten.contrib.templates.TemplatesComponent(path: str)

A component that builds a jinja2 template renderer.

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

Websockets

molten has builtin support for websockets. However, there are a couple limitations:

• you must use gunicorn as your web server and
• you need to use the gevent worker class.

Check out the websockets example in the molten repo.

class molten.contrib.websockets.WebsocketsMiddleware(origin_re: Optional[Pattern[str]] = None)

A middleware that handles websocket upgrades.

Warning

Please note that this functionality is currently gunicorn-specific and it requires the use of async workers in order to function correctly.

Parameters:origin_re – An optional regular expression that can be used to validate the origin of incoming browser requests.

handle_exception(exception: BaseException, websocket: molten.contrib.websockets.Websocket) → None

Called whenever an unhandled exception occurs in middleware or a handler. Overwrite this in a subclass to implement custom error handling for websocket handlers.

If you do overwrite this, don’t forget to close the websocket connection when necessary.

class molten.contrib.websockets.Websocket(stream: molten.contrib.websockets._BufferedStream)

Represents a single websocket connection. These are used for bi-directional communication with a websocket client.

Websockets are not thread-safe.

Example

>>> from molten import annotate
>>> from molten.contrib.websockets import CloseMessage, Websocket

>>> @annotate(supports_ws=True)
... def echo(sock: Websocket):
...     while not sock.closed:
...         message = sock.receive()
...         if isinstance(message, CloseMessage):
...             break
...
...         sock.send(message)

closed

bool – Whether or not this socket has been closed.

receive(*, timeout: Optional[float] = None) → Optional[molten.contrib.websockets.Message]

Waits for a message from the client for up to timeout seconds.

send(message: molten.contrib.websockets.Message) → None

Send a message to the client.

close(message: Optional[molten.contrib.websockets.Message] = None) → None

Close this websocket and send a close message to the client.

Note

This does not close the underlying websocket as it’s better to let gunicorn handle that by itself.

Websocket Messages

class molten.contrib.websockets.Message(message: bytes = b'')

A websocket message, composed of one or more data frames.

to_stream(stream: molten.contrib.websockets._BufferedStream) → None

Write this message to the output stream.

get_data() → bytes

Get this message’s data as a bytestring.

get_text() → str

Get this message’s contents as text.

get_output() → Union[bytes, bytearray, memoryview]

Get this message’s output payload. CloseMessage hooks into this to prepend the status code to the payload.

class molten.contrib.websockets.BinaryMessage(message: bytes = b'')

A message containing binary data.

class molten.contrib.websockets.TextMessage(message: str = '')

A message containing text data.

class molten.contrib.websockets.CloseMessage(code: int = 1000, reason: str = '')

Received (or sent) when the connection should be closed. Close messages sent by the client are automatically handled by receive().

code

int – The close status code.

class molten.contrib.websockets.PingMessage(message: bytes = b'')

A PING message. These are automatically handled by receive().

class molten.contrib.websockets.PongMessage(message: bytes = b'')

A PONG message. These are automatically handled by receive().

Websocket Errors

class molten.contrib.websockets.WebsocketError(message: str)

Base class for errors related to websockets.

class molten.contrib.websockets.WebsocketProtocolError(message: str)

Raised whenever the protocol is violated.

class molten.contrib.websockets.WebsocketMessageTooLargeError(message: str)

Raised when an incoming message contains too much data.

class molten.contrib.websockets.WebsocketFrameTooLargeError(message: str)

Raised when a frame’s payload is too large to be sent in a single frame.

class molten.contrib.websockets.WebsocketClosedError(message: str)

Raised when a message is sent to a closed socket.

Websocket Testing Support

class molten.contrib.websockets.WebsocketsTestClient(app: molten.app.BaseApp)

This is a subclass of the standard test client that adds an additional method called connect() that may be used to connect to websocket endpoints.

Example

>>> client = WebsocketsTestClient(app)
>>> with client.connect("/echo") as sock:
...   sock.send(TextMessage("hi!"))
...   assert sock.receive(timeout=1).get_text() == "hi!"

Note

In order for receive's “timeout” parameter to work, you need use gevent to monkeypatch sockets before running your tests.

connect(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, auth: Optional[Callable[[molten.http.request.Request], molten.http.request.Request]] = None) → molten.contrib.websockets._WebsocketsTestConnection

Initiate a websocket connection against the application.

Parameters:

• path – The request path.
• headers – Optional request headers.
• params – Optional query params.
• auth – An optional function that can be used to add auth headers to the request.

OpenAPI

molten has builtin support for generating OpenAPI documents.

class molten.openapi.OpenAPIHandler(metadata: molten.openapi.documents.Metadata, security_schemes: Optional[List[Union[molten.openapi.documents.APIKeySecurityScheme, molten.openapi.documents.HTTPSecurityScheme]]] = None, default_security_scheme: Optional[str] = None)

Dynamically generates and serves OpenAPI v3 documents based on the current application object. Once generated, the document is subsequently served from cache.

Examples

>>> get_schema = OpenAPIHandler(
...   metadata=Metadata(title="Pet Store", version="0.1.0"),
...   security_schemes=[HTTPSecurityScheme("Bearer", "bearer")],
...   default_security_scheme="Bearer",
... )

Parameters:

• metadata – Various meta-information about the current API.
• security_schemes – A list of security schemes used throughout the API.

class molten.openapi.OpenAPIUIHandler(schema_route_name: str = 'OpenAPIHandler')

Renders the Swagger UI.

Parameters:schema_route_name – The name of the route that exposes the schema. The actual path to the schema is looked up whenever the handler is called.

class molten.openapi.Metadata(title, description, version, terms_of_service=Missing, contact=Missing, license=Missing)

Metadata about the exposed API.

https://github.com/OAI/OpenAPI-Specification/blob/master/versions/3.0.1.md#infoObject

class molten.openapi.Contact(name=Missing, url=Missing, email=Missing)

Contact information for the exposed API.

https://github.com/OAI/OpenAPI-Specification/blob/master/versions/3.0.1.md#contact-object

class molten.openapi.License(name, url=Missing)

License information for the exposed API.

https://github.com/OAI/OpenAPI-Specification/blob/master/versions/3.0.1.md#license-object

class molten.openapi.APIKeySecurityScheme(name, param_name, in_, type=Missing)

Describes an API key-based security scheme.

https://github.com/OAI/OpenAPI-Specification/blob/master/versions/3.0.1.md#securitySchemeObject

class molten.openapi.HTTPSecurityScheme(name, scheme, type=Missing)

Describes an HTTP-based security scheme.

https://github.com/OAI/OpenAPI-Specification/blob/master/versions/3.0.1.md#securitySchemeObject

molten.openapi.generate_openapi_document(app: molten.app.BaseApp, metadata: molten.openapi.documents.Metadata, security_schemes: List[Union[molten.openapi.documents.APIKeySecurityScheme, molten.openapi.documents.HTTPSecurityScheme]], default_security_scheme: Optional[str] = None) → Dict[str, Any]

Generate an OpenAPI v3 document from an application object and some metadata.