toolbox.py Documentation v1.5.1.1

toolbox.py is a general-purpose utility module containing:

• Lazy module loading
• Async and sync HTTP helpers
• File and system utilities
• Statistical analysis helpers
• Benchmarking tools
• ASCII and HTML table generators
• CSV readers, indexed collections, cached descriptors, and parsers
• IP/request inspection, address and phone normalization, calendar/vCard output
• Mailgun, Twilio, and database-backed user-management integrations

It is designed as a broad utility library for rapid Python development, especially in backend, data, and tooling workflows.

Import: the module is a single file, so consumers normally use import toolbox or import selected names directly. Several integrations require optional third-party packages or local data files; see the relevant section before deploying a feature.

The module combines unrelated but commonly useful utilities into a single file. Key themes include:

• Deferred imports for performance
• Caching of computed values
• Human-readable formatting helpers
• Lightweight internal reporting and inspection tools

The standard library covers only part of the module. Dependencies are loaded at different times, so a successful import toolbox does not prove that every feature is ready to use.

Important: importing this module currently enables AUTO_UPDATER and calls pmblue_update.self_update(). If pmblue_update is missing, the import path attempts to download it. The missing-updater bootstrap still uses its legacy HTTP URL. Provision pmblue_update before import to avoid that bootstrap. Account for this network and file-write side effect in offline builds, tests, and restricted production environments.

Purpose

Defers module import until the module is actually used.

Constructor

DelayModuleLoad(module_name, name_as=None, module_to_globals=True)

Behavior

• Stores module name without importing immediately
• Installs itself into globals or caller module namespace
• Performs real import on first attribute access

Important Method

__getattr__(self, name)

Loads the real module, optionally replaces the placeholder, then resolves the requested attribute.

Example

asyncio = DelayModuleLoad("asyncio")
await asyncio.sleep(1)

AiohttpResponse

Defined in the aiohttp feature block. Because aiohttp is represented by a deferred loader, a missing installation can surface on first request rather than at module import.

Purpose

Wraps aiohttp responses in a more requests-like interface.

Constructor

AiohttpResponse(response, elapsed, data)

Key Properties

Async Request Methods

await AiohttpResponse.get(...)
await AiohttpResponse.post(...)
await AiohttpResponse.delete(...)
await AiohttpResponse.ping(...)

download_file_async

async def download_file_async(url, dest_path, chunk_size=1024*1024, verbose=False, as_gzip=False): ...

• Streams downloads asynchronously
• Supports optional gzip output
• Uses chunked iteration
• Can emit progress information

When as_gzip=True, pass a destination without a .gz suffix; the current implementation appends it.

download_file

def download_file(url, dest_path, chunk_size=1024*1024): ...

• Synchronous downloader using requests
• Falls back to a legacy SSL context for problematic endpoints
• Writes in chunks to avoid loading the full file into memory

TestManager

Measures performance across a fixed number of iterations or a fixed time window.

Examples

tm = TestManager(trial_amt=1000)

@tm.test_func
def test():
    pass

tm.run()
tm.print_results()

Key Options

Important Methods

DataInfo

Analyzes numeric datasets and exposes cached summary statistics.

Constructor

DataInfo(data)

Main Properties

Main Methods

Nested Type: DataInfo.Item

Represents a distinct numeric value in the dataset.

DataInfoMisc

Similar in structure to DataInfo, but designed for more general sortable values and supports reverse sorting.

dir_size(dir_path)

Recursively computes total directory size.

FileInfo

FileInfo(x)

x may be a path-like value, an os.DirEntry, or a directory entry with callable is_dir()/stat(). Pass an existing FileInfo's path to construct fresh metadata. Paths are normalized to forward slashes, but relative inputs remain relative.

FileInfo Methods

scan_for_file(file)

Checks whether the given path exists and returns its FileInfo, or None.

Table

Simple ASCII table builder.

Support Classes

• Column: metadata, length calculation, totals
• Row: row storage
• Cell: typed cell storage and formatting

TxtTable

Enhanced table builder supporting text output, HTML export, searching, sorting, and Flask-friendly pagination.

Important Methods

HTML Export

table.to_html(max_len=100, condensed=True)

Flask Pagination Notes

flask_pagination() reads request args like page, rows_per_page, search, sort_by, case, and adv, then filters and formats the selected rows.

to_numb(s)

Attempts string-to-number conversion:

• "123" → 123
• "12.3" → 12.3
• Non-numeric input is returned unchanged

CSVReader

Static helpers for reading, writing, and condensing CSV data. The default result is a ListTable with header-indexed Row objects; as_list=True returns raw row lists, and row.dict() converts a row.

Avoid reusing and mutating the default list passed to remove_columns; pass an explicit list in new code.

Indexed and constrained collections

Related helpers are randomize_list, count_dict, count_print, make_json_parsable, and dict_print.

class Service:
    @CachedProperty(expire=60)
    async def status(self):
        return await load_status()

service = Service()
status = await service.status

These are custom descriptors. Async values use await obj.property, not a method call.

Additional small helpers include get_between, get_all_between, extract_int, place_suffix, bytes_to_str, word_to_digit, name_to_normal, convert_to_normal, convert_to_single_letters, and the intentionally misspelled compatibility function equivilency_convert.

RandomString

get* methods use the ordinary random source. Prefer secure_get, secure_get_alpha, secure_get_num, or secure_get_alphanum for tokens and credentials.

These helpers are available in the current 1.5.1.1 source. The ID format combines a base62 Unix-millisecond timestamp prefix with a random alphanumeric suffix.

import datetime
import toolbox

created_at = datetime.datetime.now(datetime.timezone.utc)
record_id = toolbox.peyton_id(dt=created_at, length=24)
decoded_local_time = toolbox.peyton_id_prefix_to_timestamp(record_id[:8])

print(toolbox.base10_to_base36(35))  # Z
print(toolbox.base10_to_base62(61))  # z
secure_record_id = toolbox.peyton_id_secure(length=32)

The timestamp prefix is predictable and reveals approximate creation time; the secure variant only makes the suffix cryptographically random. Neither function checks for collisions or validates requested length. Supply an integer length greater than eight when randomness is required. The conversion functions require integers; negative values currently produce an empty string. Use timezone-aware datetime inputs when the intended epoch must be unambiguous.

For an opaque random token without a timestamp, use the existing RandomString.secure_get_alphanum(size) helper.

Matching is a configurable text-comparison namespace. It can normalize Unicode, ignore spaces, collapse repeated characters, apply phonetic substitutions, treat an asterisk as a positional wildcard, and compare either complete strings or a substring window.

Version 1.5 adds Matching.Methods.levenshtein_local(a, b), returning an integer edit distance. The main fuzzy methods continue to use positional matching; they do not select this algorithm automatically.

Matching.Methods.levenshtein_local("kitten", "sitting")
# 3

Public API

Settings

Matching.Settings(
    normalize=True,
    extraneous_characters=False,
    ignore_spaces=False,
    match_asterisk=False,
    fuzzy_threshold=1,
    phonetic=False,
)

Transformation pipeline

match() and within() transform both inputs in this order:

1. If enabled, normalize Unicode and visually related characters with convert_to_normal().
2. If enabled, remove ordinary spaces.
3. Apply interchange substitutions: 0 → o, @ → a, $ → s, 1 → i, z → 2, and + → t.
4. If enabled, collapse consecutive duplicate characters.
5. If enabled, apply ordered phonetic substitutions.
6. Apply wildcard handling, then perform exact or fuzzy full-string/substring comparison.

Case behavior: normalization lowercases text. With normalize=False, match() can be case-sensitive on the exact path, while within() and low-level fuzzy comparisons still lowercase. Interchange substitutions are always applied by the high-level methods. This differs from the older standalone normalize module.

Choosing a method

Examples

# Default comparison: normalized, case-insensitive, exact.
Matching.match("Héllô", "hello")
# True

# Interchange substitutions are always applied.
Matching.match("0scar", "Oscar")
# True

# Ignore formatting spaces.
compact = Matching.Settings(ignore_spaces=True)
Matching.match("AB 123", "AB123", compact)
# True

# Collapse repeated consecutive letters.
dedupe = Matching.Settings(extraneous_characters=True)
Matching.match("hello", "helo", dedupe)
# True

# Asterisk matches one position, not an arbitrary substring.
masked = Matching.Settings(match_asterisk=True)
Matching.match("h*llo", "hello", masked)
# True

# Require at least 80% equal character positions.
fuzzy = Matching.Settings(fuzzy_threshold=.8)
Matching.match("12345", "12344", fuzzy)
# True

# Apply the module's phonetic substitutions.
phonetic = Matching.Settings(phonetic=True)
Matching.match("phone", "fone", phonetic)
# True

# Search inside a longer transformed string.
Matching.within("Main St", "10 Main Street")
# True

Behavioral caveats

• Inputs should be strings. The high-level methods do not validate or coerce arbitrary objects before normalization.
• Fuzzy matching is positional, not edit distance. It does not account for insertions or deletions the way Levenshtein similarity would.
• Fuzzy comparisons do not check equal length. fuzzy_match() compares only pairs produced by zip(), so an unmatched trailing suffix on the longer input is ignored. High-level methods inherit this behavior whenever fuzzy_threshold is not 1.
• Empty low-level comparisons can fail. fuzzy_match() divides by the number of compared positions and therefore raises ZeroDivisionError when no character pair is available.
• Thresholds are not validated. Callers should normally provide a value from 0 through 1.
• Wildcard matching is positional. It also compares paired positions and does not enforce equal lengths.
• Phonetic conversion is intentionally aggressive. It can increase false positives and should be enabled only for domains where approximate pronunciation matters.
• Default settings are shared. Avoid mutating Matching.default_settings globally; construct a dedicated Matching.Settings object for non-default behavior.

Recommended reusable configuration

name_matching = Matching.Settings(
    normalize=True,
    ignore_spaces=True,
    extraneous_characters=False,
    phonetic=True,
    fuzzy_threshold=.9,
)

if Matching.match(submitted_name, stored_name, name_matching):
    ...

For identity, authentication, compliance, or deduplication decisions, treat this class as one signal rather than definitive proof of a match.

Request-aware user-agent parsing

request = StandardRequest(framework_request)
ua = UserAgent.from_request(request)
if ua and ua.is_mobile:
    print(ua.browser, ua.browser_version)

UserAgent.from_request considers Client Hints as well as the User-Agent header and accepts both a native request object and StandardRequest. Current parsing includes modern iPhone Safari formats, tolerates malformed version strings, and adds the UserAgent.Formats.bot6 pattern for ClaudeBot-style crawler headers.

ZF_REQ and run_code(code, **kwargs) are specialized internal-facing helpers; inspect their implementation before exposing them to untrusted input. Never pass untrusted code to run_code.

calendar = iCalendar("Support rota", "[email protected]")
calendar.add_event("On call", "event-1", starts_at, ends_at)
ics_text = calendar.generate()

Address

Address(d, correct_address=True, parse_address=True) accepts a string, mapping, or supported address-like value. It normalizes street suffixes, state abbreviations, ZIP/postal codes, and common misplaced fields. Use Address.parse_address(text) for direct parsing, to_dict() for serialization, and compare_address for normalized comparison. Aliases include street, street2, zip_code, and postal_code. Version 1.5 uses precomputed street-suffix and state-name lookups to speed up parsing.

PhoneNumber

PhoneNumber(number) normalizes a phone number and exposes e164, domestic_format, and international_format.

Other domain types

Integrations are grouped under the lowercase plugins namespace; instantiate a nested class directly.

Mailgun

mailgun = plugins.Mailgun(api_key, domain,
    default_from="Support <[email protected]>",
    default_archive_to="[email protected]")

response = await mailgun.send_email(
    "Welcome", "[email protected]", html="<p>Hello</p>"
)

send_email accepts one of text, html, or template, plus CC/BCC, reply-to, template variables, and per-message archive_to. The per-message value overrides the constructor default.

Twilio

twilio = plugins.Twilio(account_sid, auth_token, default_from="+15551234567")
await twilio.send_sms("+15557654321", body="Hello")

lookup = await twilio.lookup(
    "+15557654321",
    ["line_status", "line_type_intelligence"]
)
active = lookup.line_status.is_line_active if lookup.line_status else None

lookup(phone_number, lookup_fields, **params) accepts a field string or list and forwards additional query parameters. It returns plugins.Twilio.LookupNumber, which retains response-style properties such as status_code, headers, text, json(), and raise_for_status(), while also exposing parsed caller_name, identity_match, line_status, and line_type_intelligence objects.

identity_match(..., **params) formats supported birth dates as YYYYMMDD and accepts an Address, mapping, or string. Convenience methods line_status(phone) and is_line_active(phone) return parsed status; the latter is tri-state: True, False, or None when unknown.

UserManager

plugins.UserManager(dbm, tbl_name='users', mfa_tbl=None, session_validator=None) integrates with aiosqliteObj. It creates required user/session columns, hashes passwords, caches users, and manages persisted sessions. Nested types include User, UserSession, UserSession.Validation, and MFA.

• add_user, get_user, and get_session are async.
• Session tokens use RandomString.secure_get_alphanum.
• Validation can check session age, user-agent continuity, IP/location distance, and a custom sync or async function.
• Session user-agent data is stored as a structured payload; legacy string payloads remain readable.

Alphabetical glossary of the public toolbox API. “Conditional” entries exist only when their optional dependency is available.

Classes

Functions

This index lists every non-underscore top-level function and class in version 1.5.1.1. Some classes are support types normally created by a higher-level API.

This guide describes the distributed source at 1.5.1.1. Recent additions and retained features include:

• Timestamp-prefixed Peyton IDs, prefix decoding, the day timestamp helper, and base36/base62 conversions.
• Matching.Methods.levenshtein_local alongside the existing positional fuzzy functions.
• Faster normalization and address parsing, random.choices for ordinary alphanumeric generation, and deferred urllib3/concurrent.futures loading.
• The bot6 user-agent pattern and removal of eager installation of a missing aiosqliteObj during startup.
• Twilio Lookup support, parsed lookup response models, identity matching, line status, and forwarding of extra query parameters.
• Mailgun archive_to support at both client and message level.
• Time.parse_date and broader accepted date inputs in Twilio identity matching.
• Faster scan_for_file and expanded FileInfo input handling.
• AttrDict, CacheList, and large-file CSV parsing through by_line.
• Expiring AsyncIpInfo caches and increased aiohttp header-field allowance for large-download endpoints.
• Cryptographically secure random helpers and their use for session-token generation.
• Request-aware User-Agent parsing, structured session user-agent payloads, modern iPhone handling, and malformed-version resilience.

• The file is monolithic and mixes several unrelated concerns.
• Many properties cache values after first computation.
• Some function names imply standard statistics, but behavior may differ from textbook definitions.
• Several helpers are optimized for convenience and internal use over strict abstraction.
• Several APIs use mutable default arguments. New callers should pass explicit lists or dictionaries when they may mutate an input.
• Network helpers do not uniformly provide default timeouts. Pass timeouts through supported request keyword arguments in production code.

DataInfo

di = DataInfo([1, 2, 3, 4, 100])
print(di.info_txt())

Table

t = Table()
t.add_column("Name")
t.add_column("Value")
t.add_row("A", 100)
t.add_row("B", 200)
t.print()

AiohttpResponse

resp = await AiohttpResponse.get("https://example.com")
print(resp.text)

Most toolbox APIs fit one of three calling styles: plain functions return converted values, namespace classes such as CSVReader expose functions called directly on the class, and stateful objects such as TxtTable hold data across method calls. Async methods must be awaited; custom async properties use await object.property.

scan_for_file(path) returns a FileInfo when the path exists, otherwise None. FileInfo.all_files() and all_files_recursive() return filename strings, not FileInfo objects or reliable full paths; recursive results can contain the same basename from different subdirectories.

Many namespace methods omit self; call CSVReader.read(...), Time.parse_time(...), and Matching.match(...) on the class. Constructing and calling them as ordinary instance methods may pass an unexpected argument.

The default CSV result is a ListTable. Its rows support header-based indexing; the report creates separate typed cells so numeric display formatting can apply. No external file or service is required after the import-time updater setup.

import io
import toolbox

csv_input = io.StringIO("name,amount\nAda,1200\nGrace,85.5\n")
data = toolbox.CSVReader.basic_read(csv_input)
report = toolbox.TxtTable(format_numbers=True)
report.add_column("Name")
report.add_column("Amount")

for row in data:
    report.add_row(row["name"], toolbox.to_numb(row["amount"]))

print(report.print(send=False))
plain_records = [row.dict() for row in data]
print(plain_records[0]["name"])  # Ada
csv_output = toolbox.CSVReader.write([
    ["name", "amount"],
    *[[record["name"], record["amount"]] for record in plain_records],
])
print(csv_output)

CSVReader.write(rows) expects an iterable of row sequences and returns text. write_file(path, rows) also writes/closes its destination. condense() rewrites its input file. The custom serializer quotes commas and quotes but does not cover every multiline CSV case; use Python's standard csv module when general CSV interoperability is required.

CacheList stores objects by the key returned by your callback. Looking up a key does not refresh its eviction position. In this example, adding a third distinct key evicts the first because the capacity is two.

import toolbox

cache = toolbox.CacheList(max_cache_size=2)
cache.stringify_func(lambda item: item["id"])
cache.append({"id": "a", "name": "Ada"})
cache.append({"id": "b", "name": "Grace"})
print(cache.get("a")["name"])  # Ada
cache.append({"id": "c", "name": "Linus"})
print(cache.get("a"))  # None
print(cache.pop()["name"])  # Linus

class Report:
    def __init__(self, values):
        self.values = list(values)
        self.calculations = 0

    @toolbox.CachedProperty(expire=60)
    def total(self):
        self.calculations += 1
        return sum(self.values)

report = Report([10, 20])
print(report.total, report.total, report.calculations)  # 30 30 1
print(Report.total.is_cached(report))  # True
report.total = 40  # Explicitly replace the cached value.
print(report.total)  # 40

CachedProperty accepts an expiry in seconds or a timedelta. Read the descriptor through the class, as in Report.total.is_cached(report) and get_cached(report). Assignment replaces the cached value and resets its expiry; a decorated setter may transform the assigned value. There is no public cache-clear method. A changed underlying list does not invalidate the property automatically.

TxtParser captures strings; convert each field explicitly. PhoneNumber strips nondigits, assumes +1 for ten digits, and treats extra leading digits as a country code. It formats values but does not validate real numbering-plan assignments.

import toolbox

parser = toolbox.TxtParser("<name> ; <phone>")
fields = parser("Ada Lovelace ; (212) 555-0100")
if fields is None:
    raise ValueError("Expected name ; phone")

phone = toolbox.PhoneNumber(fields["phone"])
print(phone.e164)  # +12125550100
print(phone.domestic_format)  # (212) 555-0100
print(phone.country_code, phone.area_code)  # +1 212

address = toolbox.Address({
    "address1": "10 Main Street",
    "address2": None,
    "city": "New York",
    "state": "NY",
    "zip": "10001",
}, correct_address=False)
print(address.street)
print(address.to_dict()["zip"])
print(toolbox.Matching.match("Héloise", "HELOISE"))  # True

Address owns mutable address1, address2, city, state, and zip fields. street, street2, zip_code, and postal_code are read aliases. to_dict() returns those five fields; parsed_address and compare_address are cached string properties, not methods. Construct a new Address after changing fields if fresh cached normalization is required. Address normalization does not verify postal deliverability.

This example requires aiohttp and network access. The response helpers consume the response body before returning AiohttpResponse. Then content, text, and json() are synchronous byte/string/decoded-JSON accessors; await the request, not those accessors.

import asyncio
import aiohttp
import toolbox

async def main():
    response = await toolbox.AiohttpResponse.get(
        "https://example.com/",
        timeout=aiohttp.ClientTimeout(total=10),
    )
    response.raise_for_status()
    print(response.status_code, response.url)
    print(len(response.content), response.elapsed.total_seconds())
    print(response.text[:80])

asyncio.run(main())

get, post, and delete forward arguments to aiohttp requests and return AiohttpResponse. headers, cookies, status_code/status, url, encoding, and elapsed provide response metadata. raise_for_status() raises for HTTP errors; methods do not call it automatically. ping() issues a GET, closes the response without loading its body, and returns None.