commons

Small sync helpers shared across projects. Base is stdlib only — no dependencies.

• timing — unix-timestamp deltas + timezone-aware datetime conversions
• paths — nested dict/list access by dotted path
• masking — display masking for cards / cvv / tokens
• retry — exponential-backoff retry, sync (retry) and async (aretry)
• addr — ip/address tooling: pure stdlib ip utils in base, async geo lookups behind the commons[addr] extra

Install

commons @ git+ssh://git@git.rethinkstudios.io/rethink-public/commons.git@v0.2.3
# async address/geo lookups (fetch_ip / ip_location / fetch_location) need the extra:
commons[addr] @ git+ssh://git@git.rethinkstudios.io/rethink-public/commons.git@v0.2.3

The base install pulls nothing (stdlib). Only commons[addr] adds aiohttp, and only for the geo lookups — the pure commons.addr.ip utilities ship in base.

Drop the @v0.2.3 suffix from the line above to install the latest unpinned.

timing

Unix ints stay the storable value; datetimes are produced on demand in whatever timezone you ask for. One engine (_delta_seconds) backs both the bare functions and Clock.

Deltas (bare functions)

from commons import now, add, ahead, ago, is_expired

now()                       # current unix ts (int)
add(ts, days=1, hours=-3)   # shift a ts by signed units
ahead(days=7)               # now + delta  (replaces the old in_days/in_hours/...)
ago(hours=2)                # now - delta
is_expired(deadline)        # True if past; None/0 never expires

add/ahead/ago take days/hours/minutes/seconds as keyword units — one call, any combination, signed.

Datetime + timezone

from commons import to_dt, to_unix, now_dt, convert, fmt, date

to_dt(ts)                       # aware datetime in UTC
to_dt(ts, "America/New_York")   # same instant, eastern wall clock
to_unix(some_datetime)          # datetime -> unix (naive read as UTC, or pass tz)
now_dt("Asia/Tokyo")            # current time as an aware datetime in a tz
convert(dt, "Asia/Tokyo")       # re-express any datetime in another tz (same instant)
fmt(ts, "America/New_York", "%H:%M")   # formatted string in a tz
date(ts, "America/New_York")    # "mm/dd/YYYY" in a tz (ts optional -> now)

Timezones accept an IANA name ("America/New_York"), a tzinfo, or None (UTC).

Clock

A Clock binds a timezone + fast flag and delegates to the same functions, so you don't repeat the tz on every call:

from commons import Clock

clock = Clock("America/New_York")     # any IANA tz; defaults to UTC
clock.now()                            # unix ts
clock.add(ts, days=1)                  # delta (honors the clock's fast flag)
clock.ahead(days=7) / clock.ago(hours=2)
clock.to_dt(ts) / clock.now_dt()       # datetimes in the clock's tz
clock.convert(dt)                      # re-express dt in the clock's tz
clock.fmt(ts) / clock.date(ts)         # formatted in the clock's tz

Test mode

fast collapses every unit to one second so time-based flows run quickly in tests. Per-call (add(ts, days=1, fast=True)), per-clock (Clock(fast=True)), or as the module default for bare calls:

from commons import timing
timing.FAST_MODE = True     # in test setup

paths

from commons import deep_get, deep_set

data = {"in": {"this": {"old": {"notation": 42}}}, "items": [{"id": "a"}, {"id": "b"}]}

deep_get(data, "in.this.old.notation")   # 42
deep_get(data, "items.1.id")             # "b"  (numeric segment indexes a list)
deep_get(data, "in.nope.here", "DEF")    # "DEF" (missing -> default, no raise)

deep_set({}, "a.b.c", 9)                 # {"a": {"b": {"c": 9}}}

masking

Display helpers only — they format a value for showing; they are not a security control (the underlying value is unchanged and still needs proper handling).

from commons import credit, cvv, phantom, provider

credit("4111 1111 1111 1234")   # "•••• •••• •••• 1234"
cvv("123")                       # "•••"
phantom("abcdef1234567890")      # "abcdef...7890"
provider("4111111111111111")     # "VISA"  (VISA/MC/AMEX/UPAY/DISC/JCB/DNRS/UNKW)

retry

Exponential-backoff retry, sync (retry) and async (aretry). Each works as a call form or a decorator, with the same kwargs. After the attempts are exhausted the last exception is re-raised — it never swallows or returns a default.

from commons import retry, aretry

# call form
rows = retry(lambda: read_db(), attempts=5, on=(IOError,))
data = await aretry(lambda: fetch(url), attempts=3, backoff=0.5, on=(TimeoutError,))

# decorator form (same kwargs)
@aretry(attempts=4, backoff=0.5, factor=2.0, on=(ConnectionError,))
async def pull():
    ...

Knobs: attempts (total tries), backoff / factor / max_backoff (delay is min(backoff * factor**n, max_backoff)), jitter (full jitter, on by default), on= (tuple of retryable exception types), and give_up=lambda exc: ... to stop early on a non-retryable error (e.g. a 400 vs a 429):

# retry 429/5xx but give up immediately on a 4xx
await aretry(send, attempts=4, on=(HTTPError,),
             give_up=lambda e: 400 <= e.status < 500 and e.status != 429)

Each retry is logged (emit-only). sleep= and rand= are injectable for deterministic tests (no real waits).

addr

IP/address tooling, exposed as a submodule. The pure ip utilities ship in the base install (stdlib ipaddress); the async geo lookups need commons[addr].

ip (pure, base install)

from commons import addr

addr.ip.is_valid("8.8.8.8")            # True   (False on bad input, never raises)
addr.ip.version("2001:db8::1")         # 6      (None if invalid)
addr.ip.to_int("0.0.0.1")              # 1
addr.ip.from_int(1)                     # "0.0.0.1"   (version=6 for ipv6)
addr.ip.is_private("10.0.0.5")         # True
addr.ip.is_global("8.8.8.8")           # True

addr.ip.in_network("10.0.0.5", "10.0.0.0/24")          # True
addr.ip.in_any("10.0.0.5", ["1.2.3.0/24", "10.0.0.0/8"])  # True (allow/blocklists)
# in_network / in_any return False on bad address OR bad cidr — they never raise

addr.ip.network_address("10.0.0.5/24")  # "10.0.0.0"   (strict=False tolerates host bits)
addr.ip.broadcast_address("10.0.0.5/24")# "10.0.0.255"
addr.ip.netmask("10.0.0.0/24")          # "255.255.255.0"
addr.ip.prefix_bits("10.0.0.0/24")      # 24
addr.ip.host_bits("10.0.0.0/24")        # 8
addr.ip.num_addresses("10.0.0.0/24")    # 256
addr.ip.set_bits("0.0.0.3")             # 2   (popcount of the address int)
addr.ip.hosts("10.0.0.0/29")            # ["10.0.0.1", ... "10.0.0.6"]
addr.ip.hosts("10.0.0.0/8", limit=100)  # cap materialization on huge ranges

geo (async, needs commons[addr])

Each lookup is async, accepts an optional session= (reuse an aiohttp ClientSession; one is created and closed internally if omitted) and timeout=15, and returns None on any request/parse failure.

from commons.addr import fetch_ip, ip_location, fetch_location

await fetch_ip()                                  # your public ip via ipify
await ip_location("8.8.8.8", api_key=cfg_key)     # geo.ipify; api_key REQUIRED, injected
await fetch_location(40.7, -74.0)                 # nominatim reverse -> {"country", "state"}

• ip_location's api_key is a required keyword you inject — there is no default and nothing hardcoded. (The old code shipped a hardcoded key; it's gone.)
• fetch_location sets a User-Agent (Nominatim's terms require one); override via user_agent="your-app/1.0".
• Without the [addr] extra installed, the package still imports — but calling a geo function raises a clear RuntimeError telling you to install commons[addr].