webtools package

Submodules

webtools.html_table module

webtools.html_table.generate(data: dict, css='')[source]
webtools.html_table.main(filename: str, stylesheet: str = None, html=True, md=False)[source]
webtools.html_table.parse(data: dict)[source]
webtools.html_table.parse_cell(text: str, rowspan=1)[source]
webtools.html_table.parse_column(column: dict, height: int)[source]
webtools.html_table.parse_row(row: dict, headings: list, height=1)[source]
webtools.html_table.render(css, max_height, headings, with_width)[source]
webtools.html_table.render_table_body(max_height, headings, with_width)[source]
webtools.html_table.render_table_head(doc, css)[source]

webtools.html_table_data module

A datastructures that represent an HTML table with cells of variable height.

class webtools.html_table_data.HTMLTableData(parameters: webtools.html_table_data.Parameters, rows: List[webtools.html_table_data.Row])[source]

Bases: object

property max_row_height
parameters: Parameters
rows: List[Row]
class webtools.html_table_data.Markdown[source]

Bases: str

static parse_value(value)[source]
class webtools.html_table_data.Parameters(headings: Dict[str, webtools.html_table_data.Markdown])[source]

Bases: object

headings: Dict[str, Markdown]
class webtools.html_table_data.Row(row: Dict[str, List[webtools.html_table_data.Markdown]])[source]

Bases: object

property height

The max. number of “stacked” cells. This can be used to infer the html rowspan property.

row: Dict[str, List[Markdown]]
webtools.html_table_data.parse_json(json: dict)[source]

webtools.parallel module

Generic parallelization functions using asyncio.

webtools.parallel.asynchronous(func, inputs, concurrency=4, **kwds)[source]

Executes func(task) for every task in tasks.

Parameters

func : async funcion(client: aiohttp.ClientSession, *) -> Result tasks : iterable of (unique) input for each function invocation * : constants arguments and keywords to be passed to each function

webtools.parallel.main()[source]
webtools.parallel.new_event_loop()[source]
webtools.parallel.run(func, items, batch_size, duration, n_threads=2, **kwds)[source]

Executes func(i) N x M times. It is assumed that all function invocations are independent.

Parameters

func : async funcion(client: aiohttp.ClientSession, *) -> Result items : inputs per function call batch_size : number of function call results that are yielded duration : timeout of the process. This evaluated between batches and not during batches. n_threads : int concurrency : max. number of async connections per thread **kwds : arguments for func. func() must be threadsafe batches : iterable of iterables

webtools.parallel.show_status(status, times, dt, exceptions=[], new_line=False, **kwds)[source]
async webtools.parallel.simple_custom_func(session: ClientSession, i: int, url='')[source]
async webtools.parallel.some_custom_func(session: ClientSession, i: int, url='', timeout=10)[source]
async webtools.parallel.try_task(func, inputs, session, results, errors, **kwds)[source]
async webtools.parallel.worker(func, queue: Queue, results=[], errors=[], **kwds)[source]

webtools.parallel_requests module

webtools.parallel_requests.compute(N=1, strategy=Strategy.push)[source]
webtools.parallel_requests.echo(json, n_calls=1)[source]
webtools.parallel_requests.echo_offline(json)[source]
webtools.parallel_requests.generate_(*args) → int[source]
async webtools.parallel_requests.get(session: ClientSession, i: int, url='', timeout=10, **kwds)[source]
webtools.parallel_requests.parse_response_(response, default=None) → int[source]
async webtools.parallel_requests.post(session: ClientSession, i: int, url='', timeout=10, **kwds)[source]
webtools.parallel_requests.randomize_(*args) → int[source]
webtools.parallel_requests.run_compute_pipeline()[source]
webtools.parallel_requests.sleep(seconds, n_calls=1)[source]
webtools.parallel_requests.test_compute_pipeline()[source]
webtools.parallel_requests.to_float_(s: str) → int[source]
webtools.parallel_requests.to_int_(s: str) → int[source]
webtools.parallel_requests.to_str_(s: str) → int[source]

webtools.pipeline module

class webtools.pipeline.Buffer(*args, n=2, **kwds)[source]

Bases: Processor

A Processor with an explicit buffer that can be used to adjust buffer sizes. E.g. group or split up items.

full() → bool[source]
class webtools.pipeline.Combiner(*args, n=2, **kwds)[source]

Bases: Buffer

Many-to-one e.g. Combiner(n=2) transforms items into pairs of items

process_item(item)[source]

Override this method to change the default identiy method.

ready_to_process() → bool[source]
class webtools.pipeline.Distributer(*args, n=2, **kwds)[source]

Bases: Buffer

One-to-many e.g. Distributer(n=2) transforms pairs into items

append(group_of_items)[source]
extend(groups_of_items)[source]
class webtools.pipeline.Pipeline(items=[], *, processors: List[Processor] = [])[source]

Bases: Processor

A Processor that combines of multiple processing-stages.

Note that len(Pipeline) returns the length of the buffer of the whole Pipeline and not the total length of all components.

init_processors(processors: List[Processor])[source]
class webtools.pipeline.Processor(items=[])[source]

Bases: object

A queue-like object that can “process” items.

The superclass list allows the “queue” to be sorted and grouped. * Use .append() or .extend() to add items to the queue/buffer. * Use .process() to handle or process an item.

append(*args)[source]
clear()[source]
extend(*args)[source]
static from_function(pure_func)[source]

Decorator that defines a “stateless” processor.

Usage

new_func = Processor.from_function(func)

Note that new_func and func must have the same name in order to be compatible with pickle.

Parameters

pure_func : (object) -> object

process(item=None)[source]

Add an item to a queue, process the next item in the queue and return the result. Override this method to change queue behaviour into e.g. FIFO.

process_item(item)[source]

Override this method to change the default identiy method.

ready_to_process() → bool[source]
class webtools.pipeline.PushPull(*args, strategy=Strategy.constant, **kwds)[source]

Bases: Pipeline

append(item)[source]
extend(items)[source]
property in_queue
init_resources(strategy)[source]
n_processors = 1
property out_queue
process(item=None)[source]

Process an item and then yield the result

process_buffer()[source]
process_item(item)[source]

Override this method to change the default identiy method.

start_resources()[source]
class webtools.pipeline.Resource(processor: webtools.pipeline.Processor, delivery_queues: Tuple[queue.Queue, queue.Queue], demand_queues: Tuple[queue.Queue, queue.Queue] = None, strategy: webtools.pipeline.Strategy = <Strategy.push: 1>)[source]

Bases: object

delivery_queues: Tuple[Queue, Queue]
demand_input()[source]
demand_queues: Tuple[Queue, Queue] = None
property in_queue: Queue
property out_queue: Queue
process()[source]
processor: Processor
put(item)[source]
start(max_items=None)[source]
strategy: Strategy = 1
wait_for_input_demand()[source]
class webtools.pipeline.Strategy(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

constant = 3
pull = 2
push = 1
webtools.pipeline.constant_(*args)[source]
webtools.pipeline.duplicate_(x)[source]
webtools.pipeline.identity_(x)[source]

webtools.verify_server module

exception webtools.verify_server.VerificationException[source]

Bases: Exception

webtools.verify_server.connect(hostname, port=443, timeout=3)[source]
webtools.verify_server.infer_protocol_and_port_from_hostname(hostname, default_port)[source]
webtools.verify_server.main(url: str)[source]
webtools.verify_server.parse_hostname(hostname, default_protocol='http', default_port=80)[source]
webtools.verify_server.parse_port(port: str)[source]
webtools.verify_server.parse_url(url: str, default_scheme='http')[source]

Return the scheme, domain, port and path or a url

Parameters

The following inputs are supported: - https://example.com - www.example.com:80 - 1.0.0.1:443

Returns

ParseResult(scheme=’http’, netloc=’example.com:443’, path=’/path;parameters’, params=’’,

query=’query’, fragment=’fragment’)

webtools.verify_server.resolve(hostname: str)[source]
webtools.verify_server.resolve_then_connect(hostname, *args)[source]

Module contents

A collection of web-related tools.

- html_table: Converts Python objects to HTML 
- parallel_requests: Load testing
- verify_server: Verify connectivity to a URL