v0.1.3; add several functions in utils

PandaPkgInfo.py

@@ -1 +1 @@
-release_version = "0.1.2"
+release_version = "0.1.3"

pandacommon/pandautils/PandaUtils.py

@@ -1,4 +1,5 @@
 import datetime
+import itertools
 
 import pytz
 
@@ -33,17 +34,17 @@ def isLogRotating(before_limit, after_limit):
     return False
 
 
-def aware_utcnow() -> datetime:
+def aware_utcnow() -> datetime.datetime:
     """
     Return the current UTC date and time, with tzinfo timezone.utc
 
     Returns:
         datetime: current UTC date and time, with tzinfo timezone.utc
     """
-    return datetime.now(timezone.utc)
+    return datetime.datetime.now(datetime.timezone.utc)
 
 
-def aware_utcfromtimestamp(timestamp: float) -> datetime:
+def aware_utcfromtimestamp(timestamp: float) -> datetime.datetime:
     """
     Return the local date and time, with tzinfo timezone.utc, corresponding to the POSIX timestamp
 
@@ -53,10 +54,10 @@ def aware_utcfromtimestamp(timestamp: float) -> datetime:
     Returns:
         datetime: current UTC date and time, with tzinfo timezone.utc
     """
-    return datetime.fromtimestamp(timestamp, timezone.utc)
+    return datetime.datetime.fromtimestamp(timestamp, datetime.timezone.utc)
 
 
-def naive_utcnow() -> datetime:
+def naive_utcnow() -> datetime.datetime:
     """
     Return the current UTC date and time, without tzinfo
 
@@ -66,7 +67,7 @@ def naive_utcnow() -> datetime:
     return aware_utcnow().replace(tzinfo=None)
 
 
-def naive_utcfromtimestamp(timestamp: float) -> datetime:
+def naive_utcfromtimestamp(timestamp: float) -> datetime.datetime:
     """
     Return the local date and time, without tzinfo, corresponding to the POSIX timestamp
 
@@ -77,3 +78,18 @@ def naive_utcfromtimestamp(timestamp: float) -> datetime:
         datetime: current UTC date and time, without tzinfo
     """
     return aware_utcfromtimestamp(timestamp).replace(tzinfo=None)
+
+
+def batched(iterable, n, *, strict=False):
+    """
+    Batch data from the iterable into tuples of length n. The last batch may be shorter than n
+    If strict is true, will raise a ValueError if the final batch is shorter than n
+    Note this function is for Python <= 3.11 as it mimics itertools.batched() in Python 3.13
+    """
+    if n < 1:
+        raise ValueError("n must be at least one")
+    iterator = iter(iterable)
+    while batch := tuple(itertools.islice(iterator, n)):
+        if strict and len(batch) != n:
+            raise ValueError("batched(): incomplete batch")
+        yield batch

pandacommon/pandautils/base.py

@@ -0,0 +1,125 @@
+##############################
+# Base classes in PanDA/JEDI #
+##############################
+
+
+class SpecBase(object):
+    """
+    Base class of specification
+    """
+    # attributes
+    attributes = ()
+    # attributes which have 0 by default
+    _zeroAttrs = ()
+    # attributes to force update
+    _forceUpdateAttrs = ()
+    # mapping between sequence and attr
+    _seqAttrMap = {}
+
+    # constructor
+    def __init__(self):
+        # install attributes
+        for attr in self.attributes:
+            self._orig_setattr(attr, None)
+        # map of changed attributes
+        self._orig_setattr("_changedAttrs", {})
+
+    # override __setattr__ to collect the changed attributes
+    def __setattr__(self, name, value):
+        oldVal = getattr(self, name)
+        self._orig_setattr(name, value)
+        newVal = getattr(self, name)
+        # collect changed attributes
+        if oldVal != newVal or name in self._forceUpdateAttrs:
+            self._changedAttrs[name] = value
+
+    def _orig_setattr(self, name, value):
+        """
+        original setattr method
+        """
+        super().__setattr__(name, value)
+
+    def resetChangedList(self):
+        """
+        reset changed attribute list
+        """
+        self._orig_setattr("_changedAttrs", {})
+
+    def forceUpdate(self, name):
+        """
+        force update the attribute
+        """
+        if name in self.attributes:
+            self._changedAttrs[name] = getattr(self, name)
+
+    def valuesMap(self, useSeq=False, onlyChanged=False):
+        """
+        return map of values
+        """
+        ret = {}
+        for attr in self.attributes:
+            # use sequence
+            if useSeq and attr in self._seqAttrMap:
+                continue
+            # only changed attributes
+            if onlyChanged:
+                if attr not in self._changedAttrs:
+                    continue
+            val = getattr(self, attr)
+            if val is None:
+                if attr in self._zeroAttrs:
+                    val = 0
+                else:
+                    val = None
+            ret[f":{attr}"] = val
+        return ret
+
+    def pack(self, values):
+        """
+        pack tuple into spec
+        """
+        for i in range(len(self.attributes)):
+            attr = self.attributes[i]
+            val = values[i]
+            self._orig_setattr(attr, val)
+
+    @classmethod
+    def columnNames(cls, prefix=None):
+        """
+        return column names for INSERT
+        """
+        attr_list = []
+        for attr in cls.attributes:
+            if prefix is not None:
+                attr_list.append(f"{prefix}.{attr}")
+            else:
+                attr_list.append(f"{attr}")
+        ret = ",".join(attr_list)
+        return ret
+
+    @classmethod
+    def bindValuesExpression(cls, useSeq=True):
+        """
+        return expression of bind variables for INSERT
+        """
+        attr_list = []
+        for attr in cls.attributes:
+            if useSeq and attr in cls._seqAttrMap:
+                attr_list.append(f"{cls._seqAttrMap[attr]}")
+            else:
+                attr_list.append(f":{attr}")
+        attrs_str = ",".join(attr_list)
+        ret = f"VALUES({attrs_str}) "
+        return ret
+
+    def bindUpdateChangesExpression(self):
+        """
+        return an expression of bind variables for UPDATE to update only changed attributes
+        """
+        attr_list = []
+        for attr in self.attributes:
+            if attr in self._changedAttrs:
+                attr_list.append(f"{attr}=:{attr}")
+        attrs_str = ",".join(attr_list)
+        ret = f"{attrs_str} "
+        return ret