CUP (Common Useful Python): Building Reliable Python Workflows with Baidu’s Utility Toolkit

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)