Fix serialization order of ops

airflow-core/src/airflow/serialization/serde.py

@@ -120,13 +120,6 @@ def serialize(o: object, depth: int = 0) -> U | None:
     if o is None:
         return o
 
-    # primitive types are returned as is
-    if isinstance(o, _primitives):
-        if isinstance(o, enum.Enum):
-            return o.value
-
-        return o
-
     if isinstance(o, list):
         return [serialize(d, depth + 1) for d in o]
 
@@ -159,6 +152,13 @@ def serialize(o: object, depth: int = 0) -> U | None:
         if is_serialized:
             return encode(classname or serialized_classname, version, serialize(data, depth + 1))
 
+    # primitive types are returned as is
+    if isinstance(o, _primitives):
+        if isinstance(o, enum.Enum):
+            return o.value
+
+        return o
+
     # custom serializers
     dct = {
         CLASSNAME: qn,

airflow-core/src/airflow/serialization/serializers/numpy.py

@@ -53,6 +53,7 @@ def serialize(o: object) -> tuple[U, str, int, bool]:
         return "", "", 0, False
 
     name = qualname(o)
+    metadata = (name, __version__, True)
     if isinstance(
         o,
         np.int_
@@ -67,13 +68,13 @@ def serialize(o: object) -> tuple[U, str, int, bool]:
         | np.uint32
         | np.uint64,
     ):
-        return int(o), name, __version__, True
+        return int(o), *metadata
 
     if isinstance(o, np.bool_):
-        return bool(o), name, __version__, True
+        return bool(o), *metadata
 
     if isinstance(o, (np.float16, np.float32, np.float64, np.complex64, np.complex128)):
-        return float(o), name, __version__, True
+        return float(o), *metadata
 
     return "", "", 0, False
 

airflow-core/tests/unit/serialization/serializers/test_serializers.py

@@ -227,11 +227,22 @@ def test_bignum_deserialize_errors(self, klass, version, payload, msg):
         with pytest.raises(TypeError, match=msg):
             deserialize(klass, version, payload)
 
-    def test_numpy(self):
-        i = np.int16(10)
-        e = serialize(i)
+    @pytest.mark.parametrize(
+        "value",
+        [
+            np.int8(1),
+            np.int16(2),
+            np.int64(4),
+            np.float16(4.5),
+            np.float64(123.11241231351),
+        ],
+    )
+    def test_numpy(self, value):
+        e = serialize(value)
+        assert isinstance(e, dict)
         d = deserialize(e)
-        assert i == d
+        assert value == d
+        assert type(value) is type(d)
 
     def test_numpy_serializers(self):
         from airflow.serialization.serializers.numpy import serialize