In this tutorial, we explore CUP, Baidu’s Common Useful Python library, as a practical utility toolkit for building stronger Python workflows. We begin by setting up the library in a Colab-friendly environment and then move through its major subsystems step by step, including logging, decorators, nested configuration, caching, ID generation, thread pools, interruptible threads, delayed execution, time utilities, Linux resource monitoring, file locking, networking helpers, object storage interfaces, type maps, and built-in testing assertions. As we progress, we do not just call functions at random; we observe how each module fits into real-world development tasks such as monitoring, automation, concurrency, configuration management, and reliability checks.
CUP Setup and Logging
import os
import sys
import time
import threading
import tempfile
import datetime
import subprocess
def banner(title):
line = "=" * 70
print("n" + line + "n" + title + "n" + line)
def skip(exc):
"""Report a gracefully-skipped section without aborting the notebook."""
print(" [skipped — {}: {}]".format(type(exc).__name__, exc))
banner("0. SETUP (install + cup.platforms + cup.version)")
subprocess.run(
[sys.executable, "-m", "pip", "install", "-q", "cup", "pytz"],
check=False,
)
import cup
ver = getattr(cup, "__version__", None)
if ver is None:
try:
from cup import version as _v
ver = getattr(_v, "VERSION", None) or getattr(_v, "__version__", "unknown")
except Exception:
ver = "unknown"
print("CUP version :", ver)
print("Python :", sys.version.split()[0])
try:
from cup import platforms
print("is_linux :", platforms.is_linux())
print("is_mac :", platforms.is_mac())
print("is_windows :", platforms.is_windows())
print("is_py3 :", platforms.is_py3())
except Exception as e:
skip(e)
banner("1. LOGGING (cup.log)")
LOGFILE = os.path.join(tempfile.gettempdir(), "cup_tutorial.log")
try:
from cup import log
log.init_comlog(
"cup_tutorial",
log.INFO,
LOGFILE,
log.ROTATION,
10 * 1024 * 1024,
True,
False,
)
log.info("hello from cup.log — written to file AND stdout")
log.warning("a warning line")
log.info_if(2 > 1, "info_if(True) -> emitted")
log.info_if(1 > 2, "info_if(False) -> you will NOT see this")
log.setloglevel(log.DEBUG)
log.debug("debug visible after setloglevel(DEBUG)")
try:
with open(LOGFILE) as fh:
last = [ln for ln in fh.read().splitlines() if ln.strip()][-1]
parsed = log.parse(last)
print("parsed last log line ->")
for k in ("loglevel", "date", "time", "pid", "srcline", "msg"):
if isinstance(parsed, dict) and k in parsed:
print(" {:8}: {}".format(k, parsed[k]))
except Exception as e:
skip(e)
except Exception as e:
skip(e)
We begin by setting up the CUP tutorial environment and installing the required packages directly from Python. We define helper functions that keep the notebook readable and allow failed sections to be skipped safely. We then explore CUP version details, platform checks, and structured logging to understand the library’s foundation.
Decorators and Nested Config
banner("2. DECORATORS (cup.decorators)")
try:
from cup import decorators
@decorators.Singleton
class AppConfig(object):
def __init__(self):
self.created_at = time.time()
a, b = AppConfig(), AppConfig()
print("Singleton: a is b ->", a is b, "(same created_at:",
a.created_at == b.created_at, ")")
@decorators.TraceUsedTime(
b_print_stdout=True,
enter_msg="event_id=0xABCDE enter",
leave_msg="event_id=0xABCDE leave",
)
def heavy_compute():
time.sleep(0.2)
return sum(range(200000))
print("heavy_compute() =", heavy_compute())
@decorators.needlinux
def linux_only():
return "this body is allowed to run on Linux"
print("needlinux ->", linux_only())
except Exception as e:
skip(e)
banner("3. RICH NESTED CONFIG (cup.util.conf)")
CONF_PATH = os.path.join(tempfile.gettempdir(), "cup_demo.conf")
CONF_TEXT = """
# ---- global scalars (layer 0) ----
host: abc.com
port: 12345
debug: false
[monitor]
enabled: true
interval: 60
regex: sshd
[.thresholds]
cpu_max: 90
mem_max: 80
[..actions]
on_breach: alert
[storage]
@path: /data/disk1
@path: /data/disk2
@path: /data/disk3
"""
try:
from cup.util import conf
with open(CONF_PATH, "w") as fh:
fh.write(CONF_TEXT)
cfg = conf.Configure2Dict(CONF_PATH, separator=":").get_dict()
print("host :", cfg["host"])
print("port :", cfg["port"])
print("monitor.enabled :", cfg["monitor"]["enabled"])
print("monitor.regex :", cfg["monitor"]["regex"])
print("monitor.thresholds.cpu_max :", cfg["monitor"]["thresholds"]["cpu_max"])
print("monitor.thresholds.actions :",
cfg["monitor"]["thresholds"]["actions"]["on_breach"])
print("storage.path (repeated @ -> list):", list(cfg["storage"]["path"]))
cfg["port"] = "10085"
cfg["monitor"]["thresholds"]["actions"]["on_breach"] = "restart"
NEW_PATH = CONF_PATH + ".new"
conf.Dict2Configure(cfg, separator=":").write_conf(NEW_PATH)
re_read = conf.Configure2Dict(NEW_PATH, separator=":").get_dict()
print("round-trip port :", re_read["port"], "(was 12345)")
print("round-trip on_breach :",
re_read["monitor"]["thresholds"]["actions"]["on_breach"], "(was alert)")
except Exception as e:
skip(e)
We move on to CUP decorators and see how they help us create single-instance classes, track execution time, and protect Linux-only functions. We then work with CUP’s rich configuration system and load a nested configuration file with sections, child sections, and repeated values. We also update the configuration and write it back to disk to confirm that the read-modify-write flow works correctly.
Caching, IDs, Thread Pools
banner("4. IN-MEMORY KV CACHE (cup.cache)")
try:
from cup import cache
kv = cache.KVCache(name="demo")
kv.set({"user:1": "alice", "user:2": "bob"}, expire_sec=2)
kv.set({"config:flag": "on"}, expire_sec=None)
print("size after sets :", kv.size())
print("get user:1 :", kv.get("user:1"))
print("get missing key :", kv.get("nope"))
print("sleeping 2.2s to let the 2s-TTL keys expire ...")
time.sleep(2.2)
print("get user:1 (expired) :", kv.get("user:1"))
print("get config:flag (eternal):", kv.get("config:flag"))
reclaimed = kv.pop_n_expired(0)
print("pop_n_expired reclaimed :", list(reclaimed.keys()) if reclaimed else [])
except Exception as e:
skip(e)
banner("5. UNIQUE ID GENERATION (cup.services.generator)")
try:
from cup.services import generator
gman = generator.CGeneratorMan()
print("uniqname :", gman.get_uniqname())
print("next_uniq_num :", gman.get_next_uniq_num())
print("next_uniq_num (again) :", gman.get_next_uniq_num(), "(monotonic)")
if hasattr(gman, "get_uuid"):
try:
print("get_uuid :", gman.get_uuid())
except Exception as e:
skip(e)
if hasattr(gman, "get_random_str"):
try:
print("get_random_str(16) :", gman.get_random_str(16))
except Exception as e:
skip(e)
print("singleton check :", generator.CGeneratorMan() is gman)
try:
cyc = generator.CycleIDGenerator("127.0.0.1", 8080)
i1, i2 = cyc.next_id(), cyc.next_id()
print("CycleIDGenerator id #1 :", i1)
print("CycleIDGenerator id #2 :", i2, "(incremented)")
print("id #1 as hex :", generator.CycleIDGenerator.id2_hexstring(i1))
except Exception as e:
skip(e)
except Exception as e:
skip(e)
banner("6. THREAD POOL (cup.services.threadpool)")
try:
from cup.services import threadpool
pool = threadpool.ThreadPool(minthreads=2, maxthreads=4, name="demo-pool")
pool.start()
results, rlock = [], threading.Lock()
def square(n):
time.sleep(0.03)
with rlock:
results.append(n * n)
return n * n
for i in range(8):
pool.add_1job(square, i)
callback_log = []
def on_done(ok, result):
callback_log.append((ok, result))
pool.add_1job_with_callback(on_done, square, 100)
def will_fail():
raise RuntimeError("boom inside worker")
pool.add_1job_with_callback(on_done, will_fail)
time.sleep(0.5)
print("live stats :", pool.get_stats())
pool.stop()
print("squares collected :", sorted(results))
print("callback results :", callback_log)
except Exception as e:
skip(e)
We use CUP’s in-memory cache to store key-value pairs with temporary and permanent lifetimes. We then generate unique names, counters, UUID-style values, random strings, and cycling IDs for distributed-style identifiers. We also create a thread pool, submit jobs, collect results, and observe callback behavior for both successful and failed tasks.
Threads, Scheduling, Time Utilities
banner("7. INTERRUPTIBLE THREADS + RW LOCK (cup.thread)")
try:
from cup import thread as cupthread
rw = cupthread.RWLock()
rw.acquire_readlock()
rw.acquire_readlock()
print("acquired 2 read locks concurrently")
rw.release_readlock()
rw.release_readlock()
rw.acquire_writelock()
print("acquired exclusive write lock")
rw.release_writelock()
stop_flag = {"running": True}
def busy_loop():
while stop_flag["running"]:
time.sleep(0.05)
t = cupthread.CupThread(target=busy_loop)
t.daemon = True
t.start()
print("worker python tid :", t.get_my_tid())
stop_flag["running"] = False
ok = t.terminate()
print("CupThread.terminate -> :", ok)
except Exception as e:
skip(e)
banner("8. DELAY / CRON EXECUTOR (cup.services.executor)")
try:
from cup.services import executor
URGENCY = getattr(executor, "URGENCY_NORMAL", 0)
svc = executor.ExecutionService(
delay_exe_thdnum=2, queue_exec_thdnum=2, name="exec-demo"
)
svc.start()
fired = []
def task(tag):
fired.append((tag, round(time.time(), 3)))
svc.delay_exec(0.3, task, URGENCY, "delayed-0.3s")
svc.queue_exec(task, URGENCY, "queued-now")
time.sleep(0.8)
svc.stop()
print("fired (tag, ts) :", fired)
try:
import pytz
import hashlib
tz = pytz.timezone("Asia/Shanghai")
timer = {"minute": [0, 30], "hour": None,
"weekday": None, "monthday": None, "month": None}
task_uuid = hashlib.md5(b"demo-cron").hexdigest()
ctask = executor.CronTask("demo-cron", tz, timer, task_uuid, task, "cron-arg")
print("cron next_schedtime :", ctask.next_schedtime())
print("cron 32-byte taskid :", ctask.taskid())
except Exception as e:
skip(e)
except Exception as e:
skip(e)
banner("9. TIME HELPERS (cup.timeplus)")
try:
from cup import timeplus
import pytz
print("get_str_now (default) :", timeplus.get_str_now())
print("get_str_now (custom fmt):", timeplus.get_str_now("%Y/%m/%d %H:%M:%S"))
tp = timeplus.TimePlus(pytz.timezone("Asia/Shanghai"))
now_utc = datetime.datetime.utcnow()
print("naive utc now :", now_utc)
print("-> Shanghai local :", tp.utc2local(now_utc))
except Exception as e:
skip(e)
We explore CUP’s thread utilities by using a reader-writer lock and an interruptible thread. We then run delayed and queued tasks through the execution service to understand simple scheduling behavior. We also create a cron-style task and inspect its next scheduled run time without waiting for the real clock to tick.
System Resources and Storage
banner("10. SYSTEM RESOURCES (cup.res.linux)")
try:
from cup.res import linux as reslinux
try:
print("cpu cores :", reslinux.get_cpu_nums())
except Exception as e:
skip(e)
try:
cpu = reslinux.get_cpu_usage(intvl_in_sec=1)
print("cpu usage sample :", cpu, "| usr=", getattr(cpu, "usr", "?"))
except Exception as e:
skip(e)
try:
mem = reslinux.get_meminfo()
gb = 1024.0 ** 3
print("mem total / available : {:.2f} GB / {:.2f} GB".format(
mem.total / gb, mem.available / gb))
except Exception as e:
skip(e)
try:
print("kernel version :", reslinux.get_kernel_version())
except Exception as e:
skip(e)
try:
all_pids = reslinux.pids()
print("live process count :", len(all_pids))
except Exception as e:
skip(e)
try:
me = reslinux.Process(os.getpid())
name = me.get_process_name() if hasattr(me, "get_process_name") else "?"
status = me.get_process_status() if hasattr(me, "get_process_status") else "?"
print("this process name/status:", name, "/", status)
except Exception as e:
skip(e)
except Exception as e:
skip(e)
banner("11. FILE UTILITIES (cup.exfile)")
try:
from cup import exfile
lock_path = os.path.join(tempfile.gettempdir(), "cup_demo.lock")
flock = exfile.LockFile(lock_path)
got = flock.lock(blocking=True)
print("acquired file lock :", got)
print("lock file path :", flock.filepath())
flock.unlock()
print("released file lock : True")
except Exception as e:
skip(e)
banner("12. NETWORKING HELPERS (cup.net)")
try:
from cup import net as cupnet
try:
print("local hostname :", cupnet.get_local_hostname())
except Exception as e:
skip(e)
try:
print("host ip :", cupnet.get_hostip())
except Exception as e:
skip(e)
try:
print("local port 9999 free? :", cupnet.localport_free(9999))
except Exception as e:
skip(e)
except Exception as e:
skip(e)
banner("13. UNIFIED OBJECT STORAGE (cup.storage.obj)")
try:
from cup.storage import obj as cupobj
print("available backends :",
[n for n in dir(cupobj) if n.endswith("ObjectSystem")])
los = None
for ctor in (lambda: cupobj.LocalObjectSystem(tempfile.gettempdir()),
lambda: cupobj.LocalObjectSystem({}),
lambda: cupobj.LocalObjectSystem()):
try:
los = ctor()
break
except Exception:
continue
if los is None:
print(" LocalObjectSystem ctor varies by version — see docs for "
"your release's exact signature.")
else:
print("LocalObjectSystem ready :", type(los).__name__,
"(put/get/head/delete available)")
except Exception as e:
skip(e)
We inspect Linux system resources, including CPU cores, CPU usage, memory, kernel version, process count, and current process status. We then use CUP’s file utility module to acquire and release an advisory lock for safe single-instance execution. We also test networking helpers and probe the object storage interface to see which storage backends are available.
Type Maps and Assertions
banner("14. BIDIRECTIONAL TYPE MAPS (cup.flag)")
try:
from cup import flag
tman = flag.TypeMan()
tman.register_types({"LOGIN": 1, "LOGOUT": 2, "HEARTBEAT": 3, "DATA": 4})
print("LOGIN -> number :", tman.getnumber_bykey("LOGIN"))
print("number 2 -> key :", tman.getkey_bynumber(2))
print("all registered keys :", list(tman.get_key_list()))
except Exception as e:
skip(e)
banner("15. BUILT-IN TEST ASSERTIONS (cup.unittest)")
try:
from cup import unittest as cup_unittest
cup_unittest.assert_eq(2 + 2, 4)
cup_unittest.assert_ne(1, 2)
cup_unittest.assert_true(isinstance([], list))
cup_unittest.assert_gt(5, 3)
cup_unittest.assert_eq_one("b", ["a", "b", "c"])
def raises_value_error():
raise ValueError("expected")
cup_unittest.expect_raise(raises_value_error, ValueError)
print("all passing assertions : OK")
try:
cup_unittest.assert_eq(1, 2, "intentional failure demo")
except Exception as e:
print("caught intended failure :", type(e).__name__)
except Exception as e:
skip(e)
banner("DONE — you exercised 15 CUP subsystems")
print("""
What you just used, by module:
cup.log .................. structured, rotating, parseable logging
cup.decorators ........... Singleton, TraceUsedTime, platform guards
cup.util.conf ............ nested config with sections + @arrays (round-trip)
cup.cache ................ TTL KV cache (get() extends lifetime)
cup.services.generator ... host-unique names, counters, cycling 128-bit IDs
cup.services.threadpool .. pool with result/failure callbacks + live stats
cup.thread ............... interruptible CupThread + reader/writer RWLock
cup.services.executor .... delay_exec / queue_exec + crontab-style CronTask
cup.timeplus ............. now-formatting + safe tz<->UTC conversion
cup.res.linux ............ /proc cpu, memory, kernel, process introspection
cup.exfile ............... advisory single-instance file locking
cup.net .................. hostname / ip / free-port helpers
cup.storage.obj .......... one API over Local / FTP / S3 / AFS backends
cup.flag ................. bidirectional key<->number type tables
cup.unittest ............. assertion toolkit that plays well with try/except
Next steps: API reference at https://cup.iobusy.com • source at
https://github.com/baidu/CUP (browse src/cup/ and the cup_test/ suite).
""")
We use CUP’s flag utilities to create bidirectional mappings between message names and numeric codes. We then run CUP’s built-in assertion helpers to validate equality, inequality, truth checks, comparisons, membership, and expected exceptions. We finish the tutorial by summarizing every CUP subsystem we exercise and connecting each one to a practical development use case.
Conclusion
In conclusion, we completed a broad hands-on tour of CUP and understand how its utilities can support production-style Python applications. We saw how CUP helps us structure logs, manage runtime configuration, cache temporary data, generate unique identifiers, coordinate threads, inspect system resources, protect files with locks, and validate behavior through assertions. It sets us a compact but useful foundation for using CUP in scripts, backend services, automation jobs, and system-level Python projects where reliability, observability, and practical tooling matter.
Check out the FULL CODES here. Also, feel free to follow us on Twitter and don’t forget to join our 150k+ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
Need to partner with us for promoting your GitHub Repo OR Hugging Face Page OR Product Release OR Webinar etc.? Connect with us
The post CUP (Common Useful Python): Building Reliable Python Workflows with Baidu’s Utility Toolkit appeared first on MarkTechPost.
