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(**kwds) 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.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