API docs: fresco.util

fresco.util.common

Common utilities for writing applications in fresco

fresco.util.common.object_or_404(ob, exception=<class 'fresco.exceptions.NotFound'>)

Return the value of ob if it is not None. Otherwise raise a NotFound exception.

fresco.util.security

Security-related utilities

fresco.util.security.check_equal_constant_time(a, b)

Return True if string a is equal to string b.

If a and b are of the same length, this function will take the same amount of time to execute, regardless of whether or not a and b are equal.

fresco.util.http

Utilities for working with data on the HTTP level

fresco.util.http.CHUNK_SIZE = 1024

Data chunk size to read from the input stream (wsgi.input)

class fresco.util.http.FileUpload(filename, headers, fileob)

Represent a file uploaded in an HTTP form submission

save(fileob)

Save the upload to the file object or path fileob

Parameters:

fileob – a file-like object open for writing, or the path to the file to be written

class fresco.util.http.HTTPMessage(policy=Compat32())

Represent HTTP request message headers

exception fresco.util.http.MissingContentLength(*args, **kwargs)

No Content-Length header given

class fresco.util.http.ParsedContentType(content_type, encoding, params)

content_type

Alias for field number 0

encoding

Alias for field number 1

params

Alias for field number 2

exception fresco.util.http.TooBig(*args, **kwargs)

Request body is too big

fresco.util.http.encode_multipart(data=None, files=None, charset='UTF-8', **kwargs)

Encode data using multipart/form-data encoding, returning a tuple of (<encoded data>, <environ items>).

Parameters:

• data – POST data to be encoded, either a dict or list of (name, value) tuples.

• charset – Encoding used for any string values encountered in data

• files – list of (name, filename, content_type, data) tuples. data may be either a byte string, iterator or file-like object.

• kwargs – other data items as keyword arguments

Returns:

a tuple of (<encoded_data>, <environ_items>), where encoded_data is a BytesIO object and environ is a dict containing the Content-Type and Content-Length headers encoded for inclusion in a WSGI environ dict.

fresco.util.http.get_body_bytes(environ, max_size=16384) → bytes

Read a single message body from environ[‘wsgi.input’], returning a bytes object.

fresco.util.http.get_content_type_info(environ, default_type='application/octet-stream', default_encoding='iso-8859-1') → ParsedContentType

Read and parse the Content-Type header and return a ParsedContentType object.

fresco.util.http.parse_header(header: str | ~email.header.Header, ie_workaround: bool = False, _broken_encoding_sniffer=<built-in method search of re.Pattern object>) → Tuple[str, Dict[str, str]]

Given a header, return a tuple of (value, {parameter_name: parameter_value}]).

Example usage:

>>> parse_header("text/html; charset=UTF-8")
('text/html', {'charset': 'UTF-8'})
>>> parse_header("multipart/form-data; boundary=-------7d91772e200be")
('multipart/form-data', {'boundary': '-------7d91772e200be'})

fresco.util.http.parse_multipart(stream: Iterator[bytes], boundary, default_charset, max_size, ie_workaround=True) → Tuple[Iterable[Tuple[str, FileUpload | str]], Callable | None]

Parse data encoded as multipart/form-data. Return an iterator over tuples of ` (<field-name>, <data>)``, and an optional close function.

data will be a string in the case of a regular input field, or a FileUpload instance if a file was uploaded.

If a close function is returned, the caller must call it in order to close any temporary files created at the end of the request lifecycle.

Parameters:

• stream – input stream from which to read data

• boundary – multipart boundary string, as specified by the Content-Disposition header

• default_charset – character set to use for encoding, if not specified by a content-type header. In practice web browsers don’t supply a content-type header so this needs to contain a sensible value.

• max_size – Maximum size in bytes for any non file upload part

• ie_workaround – If True (the default), enable a work around for IE’s broken content-disposition header encoding.

fresco.util.http.parse_parameters(s, preserve_backslashes=False) → Dict[str, str]

Return s parsed as a sequence of semi-colon delimited name=value pairs.

Example usage:

>>> from fresco.util.http import parse_parameters
>>> parse_parameters('foo=bar')
{'foo': 'bar'}
>>> parse_parameters('foo="bar\""')
{'foo': 'bar"'}

The preserve_backslashes flag is used to preserve IE compatibility for file upload paths, which it incorrectly encodes without escaping backslashes, eg:

Content-Disposition: form-data; name="file"; filename="C:\tmp\Ext.js"

(To be RFC compliant, the backslashes should be doubled up).

fresco.util.http.parse_post(environ, _io: IO[bytes], default_charset: str | None = None, max_size=16384, max_multipart_size=2097152, ie_workaround=True) → Tuple[Iterable[Tuple[str, FileUpload | str]], Callable | None]

Parse the contents of an HTTP POST request, which may be either application/x-www-form-urlencoded or multipart/form-data encoded.

Returned items are either tuples of (name, value) for simple string values or (name, FileUpload) for uploaded files.

Parameters:

• max_multipart_size – Maximum size of total data for a multipart form submission

• max_size – The maximum size of data allowed to be read into memory. For a application/x-www-form-urlencoded submission, this is the maximum size of the entire data. For a multipart/form-data submission, this is the maximum size of any individual field (except file uploads).

fresco.util.http.parse_querystring(data: str, charset: str | None = None, strict: bool = False, keep_blank_values: bool = True) → List[Tuple[str, str]]

Return (key, value) pairs from the given querystring:

>>> list(parse_querystring('green%20eggs=ham;me=sam+i+am'))
[('green eggs', 'ham'), ('me', 'sam i am')]

Parameters:

• data – The query string to parse.

• charset – Character encoding used to decode values. If not specified, fresco.DEFAULT_CHARSET will be used.

• keep_blank_values – if True, keys without associated values will be returned as empty strings. if False, no key, value pair will be returned.

• strict – if True, a RequestParseError will be raised on parsing errors.

fresco.util.http.read_until(stream: Iterator[bytes], delimiter: bytes) → Tuple[Iterator[bytes], Iterator[bytes], Callable[[], bool]]

Return two iterators over byte stream stream` and a callable which indicates whether the delimiter was found.

The first iterator yields all data up to delimiter. The second iterator generates all remaining data.

The first iterator must be exhausted before the second is iterated. The callable must only be called after the first iterator has been exhausted.

fresco.util.io

Utilities for wrapping IO streams. These are used internally by fresco when parsing wsgi request input streams.

fresco.util.wsgi

Utilities for interfacing with WSGI

class fresco.util.wsgi.ClosingIterator(iterable, *close_funcs)

Wrap a WSGI iterator to allow additional close functions to be called on application exit.

Synopsis:

>>> class filelikeobject(object):
...
...     def read(self):
...         print("file read!")
...         return ''
...
...     def close(self):
...         print("file closed!")
...
>>> def app(environ, start_response):
...     f = filelikeobject()
...     start_response('200 OK', [('Content-Type', 'text/plain')])
...     return ClosingIterator(iter(f.read, ''), f.close)
...
>>> from flea import Agent
>>> m = Agent(app).get('/')
file read!
file closed!

close()

Call all close functions listed in *close_funcs.

class fresco.util.wsgi.StartResponseWrapper(start_response)

Wrapper class for the start_response callable, allowing middleware applications to intercept and interrogate the proxied start_response arguments.

Synopsis:

>>> def my_wsgi_app(environ, start_response):
...     start_response('200 OK', [('Content-Type', 'text/plain')])
...     return ['Whoa nelly!']
...
>>> def my_other_wsgi_app(environ, start_response):
...     responder = StartResponseWrapper(start_response)
...     result = my_wsgi_app(environ, responder)
...     print("Got status", responder.status)
...     print("Got headers", responder.headers)
...     responder.call_start_response()
...     return result
...
>>> from flea import Agent
>>> result = Agent(my_other_wsgi_app).get('/')
Got status 200 OK
Got headers [('Content-Type', 'text/plain')]

See also Response.from_wsgi, which takes a wsgi callable, environ and start_response, and returns a Response object, allowing the client to further interrogate and customize the WSGI response.

Note that it is usually not advised to use this directly in middleware as start_response may not be called directly from the WSGI application, but rather from the iterator it returns. In this case the middleware may need logic to accommodate this. It is usually safer to use Response.from_wsgi, which takes this into account.

call_start_response()

Invoke the wrapped WSGI start_response function.

fresco.util.wsgi.bytes_to_environ(s)

Encode a byte string to a WSGI environ value

This returns a ‘bytes-as-unicode’ string.

fresco.util.wsgi.environ_to_bytes(s)

Decode a WSGI environ value to a bytes object

fresco.util.wsgi.environ_to_str(s, enc='UTF-8')

Decode a WSGI environ value to a string

fresco.util.wsgi.make_environ(url='/', environ=None, wsgi_input=b'', **kwargs)

Return a WSGI environ dict populated with values modelling the specified request url and data.

Parameters:

• url – The URL for the request, eg /index.html or /search?q=foo. Note that url must be properly URL encoded.

• environ – values to pass into the WSGI environ dict

• wsgi_input – The input stream to use in the wsgi.input key of the environ dict

• kwargs – additional keyword arguments will be passed into the WSGI request environment

fresco.util.wsgi.str_to_environ(s, enc='UTF-8')

Return a string encoded for a WSGI environ value

This returns a ‘bytes-as-unicode’ string:

• encode s using the specified encoding (eg utf8)

• decode the resulting byte string as latin-1