python: Fix pulling events from appsink

appsink.pull_object() is introspectable, but it needs a way to convert
the GstMiniObject to its GstEvent/GstSample subclass.

Part-of: <https://gitlab.freedesktop.org/gstreamer/gstreamer/-/merge_requests/9148>

girs/GstApp-1.0.gir

@@ -205,7 +205,7 @@ If an EOS event was received before any buffers, this function returns
           </instance-parameter>
         </parameters>
       </virtual-method>
-      <virtual-method name="try_pull_object" invoker="try_pull_object" version="1.20" introspectable="0">
+      <virtual-method name="try_pull_object" invoker="try_pull_object" version="1.20">
         <doc xml:space="preserve" filename="../subprojects/gst-plugins-base/gst-libs/gst/app/gstappsink.c">This function blocks until a sample or an event or EOS becomes available or the appsink
 element is set to the READY/NULL state or the timeout expires.
 
@@ -502,7 +502,7 @@ PLAYING state.</doc>
           </instance-parameter>
         </parameters>
       </method>
-      <method name="pull_object" c:identifier="gst_app_sink_pull_object" version="1.20" introspectable="0">
+      <method name="pull_object" c:identifier="gst_app_sink_pull_object" version="1.20">
         <doc xml:space="preserve" filename="../subprojects/gst-plugins-base/gst-libs/gst/app/gstappsink.c">This function blocks until a sample or an event becomes available or the appsink
 element is set to the READY/NULL state.
 
@@ -800,7 +800,7 @@ case new buffers will be discarded.</doc>
           </parameter>
         </parameters>
       </method>
-      <method name="try_pull_object" c:identifier="gst_app_sink_try_pull_object" version="1.20" introspectable="0">
+      <method name="try_pull_object" c:identifier="gst_app_sink_try_pull_object" version="1.20">
         <doc xml:space="preserve" filename="../subprojects/gst-plugins-base/gst-libs/gst/app/gstappsink.c">This function blocks until a sample or an event or EOS becomes available or the appsink
 element is set to the READY/NULL state or the timeout expires.
 

subprojects/gst-plugins-base/gst-libs/gst/app/gstappsink.c

@@ -2160,7 +2160,7 @@ gst_app_sink_pull_sample (GstAppSink * appsink)
 }
 
 /**
- * gst_app_sink_pull_object: (skip)
+ * gst_app_sink_pull_object:
  * @appsink: a #GstAppSink
  *
  * This function blocks until a sample or an event becomes available or the appsink
@@ -2361,7 +2361,7 @@ gst_app_sink_try_pull_sample (GstAppSink * appsink, GstClockTime timeout)
 }
 
 /**
- * gst_app_sink_try_pull_object: (skip)
+ * gst_app_sink_try_pull_object:
  * @appsink: a #GstAppSink
  * @timeout: the maximum amount of time to wait for a sample
  *

subprojects/gst-python/gi/overrides/GstApp.py

@@ -0,0 +1,40 @@
+# -*- Mode: Python -*-
+# vi:si:et:sw=4:sts=4:ts=4
+#
+# Copyright (C) 2025 Netflix Inc.
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
+
+from gi.overrides import _gi_gst
+from ..overrides import override
+from ..module import get_introspection_module
+
+
+GstApp = get_introspection_module('GstApp')
+__all__ = []
+
+
+class AppSink(GstApp.AppSink):
+    def pull_object(self):
+        obj = super().pull_object()
+        return _gi_gst.mini_object_to_subclass(obj) if obj else None
+
+    def try_pull_object(self, timeout):
+        obj = super().try_pull_object(timeout)
+        return _gi_gst.mini_object_to_subclass(obj) if obj else None
+
+
+AppSink = override(AppSink)
+__all__.append('AppSink')

subprojects/gst-python/gi/overrides/gstmodule.c

@@ -1074,6 +1074,24 @@ _gst_mini_object_is_writable (PyObject * self, PyObject * args)
   return res;
 }
 
+static PyObject *
+_gst_mini_object_to_subclass (PyObject * self, PyObject * args)
+{
+  PyObject *py_miniobj;
+  GstMiniObject *mini_object;
+
+  py_miniobj = PyTuple_GetItem (args, 0);
+  if (py_miniobj == NULL) {
+    PyErr_SetString (PyExc_TypeError, "Expected a PyGObject");
+    return NULL;
+  }
+
+  mini_object = pyg_boxed_get (py_miniobj, GstMiniObject);
+
+  return pyg_boxed_new (GST_MINI_OBJECT_TYPE (mini_object), mini_object, TRUE,
+      TRUE);
+}
+
 static PyObject *
 _gst_caps_get_structure (PyObject * self, PyObject * args)
 {
@@ -1684,6 +1702,7 @@ static PyMethodDef _gi_gst_functions[] = {
   {"mini_object_is_writable", (PyCFunction) _gst_mini_object_is_writable, METH_VARARGS, NULL},
   {"mini_object_flags", (PyCFunction) _gst_mini_object_flags, METH_VARARGS, NULL},
   {"mini_object_set_flags", (PyCFunction) _gst_mini_object_set_flags, METH_VARARGS, NULL},
+  {"mini_object_to_subclass", (PyCFunction) _gst_mini_object_to_subclass, METH_VARARGS, NULL},
 
   {"event_get_structure", (PyCFunction) _gst_event_get_structure, METH_VARARGS, NULL},
   {"event_writable_structure", (PyCFunction) _gst_event_writable_structure, METH_VARARGS, NULL},

subprojects/gst-python/gi/overrides/meson.build

@@ -1,4 +1,4 @@
-pysources = ['Gst.py', 'GstPbutils.py', 'GstVideo.py', 'GstAudio.py','GstAnalytics.py']
+pysources = ['Gst.py', 'GstPbutils.py', 'GstVideo.py', 'GstAudio.py','GstAnalytics.py', 'GstApp.py']
 python.install_sources(pysources,
   pure : false,
   subdir: 'gi/overrides',

subprojects/gst-python/testsuite/meson.build

@@ -5,6 +5,7 @@ tests = [
     ['Test fundamentals', 'test_types.py'],
     ['Test plugins', 'test_plugin.py'],
     ['Test analytics', 'test_analytics.py', ['gst-plugins-bad/gst-libs/gst/analytics', 'gst-plugins-base/gst-libs/gst/video']],
+    ['Test appsink', 'test_appsink.py', ['gst-plugins-base/gst-libs/gst/app']],
 ]
 
 runcmd = run_command(python, '-c', '''with open("@0@/mesonconfig.py", "w") as f:

subprojects/gst-python/testsuite/test_appsink.py

@@ -0,0 +1,62 @@
+# -*- Mode: Python -*-
+# vi:si:et:sw=4:sts=4:ts=4
+#
+# Copyright (C) 2025 Netflix Inc.
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
+
+import overrides_hack
+overrides_hack
+
+from common import TestCase
+
+import gi
+gi.require_version("Gst", "1.0")
+gi.require_version("GstApp", "1.0")
+from gi.repository import Gst
+from gi.repository import GstApp
+Gst.init(None)
+
+
+class TestAppSink(TestCase):
+
+    def test_appsink_object(self):
+        # Create an appsink
+        appsink = Gst.ElementFactory.make("appsink", None)
+        self.assertIsNotNone(appsink)
+        self.assertTrue(isinstance(appsink, GstApp.AppSink))
+        appsink.set_state(Gst.State.PLAYING)
+
+        # Send an event to the appsink
+        pad = appsink.get_static_pad("sink")
+        caps = Gst.Caps("audio/x-raw")
+        pad.send_event(Gst.Event.new_caps(caps))
+
+        # Send a buffer to the appsink
+        pad.chain(Gst.Buffer.new_wrapped([42]))
+
+        # 1st object pulled must be the event
+        event = appsink.pull_object()
+        self.assertTrue(isinstance(event, Gst.Event))
+        self.assertTrue(caps.is_equal(event.parse_caps()))
+
+        # 2nd object pulled must be the buffer
+        sample = appsink.pull_object()
+        self.assertTrue(isinstance(sample, Gst.Sample))
+        buf = sample.get_buffer()
+        with buf.map(Gst.MapFlags.READ) as info:
+            self.assertEqual(info.data[0], 42)
+
+        appsink.set_state(Gst.State.NULL)