CUP（通用實用 Python）：用百度工具套件打造可靠的 Python 工作流程

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 Copy CodeCopiedUse a different Browserimport 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 Copy CodeCopiedUse a different Browserbanner("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 Copy CodeCopiedUse a different Browserbanner("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 tem