python: Fix API break by not forcing usage of context manager

caps.get_structure(0).get_name() used to work, but the usage of
StructureWrapper broke it but forcing to wrap it in a "with" statement.

We already have a Structure subclass, simply make it reference its
parent Gst.Caps object.

Part-of: <https://gitlab.freedesktop.org/gstreamer/gstreamer/-/merge_requests/9148>
This commit is contained in:
Xavier Claessens 2025-06-01 16:29:09 -04:00 committed by GStreamer Marge Bot
parent f7c52a22a0
commit bbf60305ed
2 changed files with 51 additions and 16 deletions

View File

@ -142,11 +142,11 @@ __all__.append('NotWritableQuery')
class Query(Gst.Query, MiniObject):
def get_structure(self):
return StructureWrapper(_gi_gst.query_get_structure(self), self,
False)
s = _gi_gst.query_get_structure(self)
return s._set_parent(self) if s is not None else None
def writable_structure(self):
return StructureWrapper(_gi_gst.query_writable_structure(self), self, True)
return StructureWrapper(_gi_gst.query_writable_structure(self)._set_parent(self))
Query = override(Query)
@ -162,11 +162,11 @@ __all__.append('NotWritableEvent')
class Event(Gst.Event, MiniObject):
def get_structure(self):
return StructureWrapper(_gi_gst.event_get_structure(self), self,
False)
s = _gi_gst.event_get_structure(self)
return s._set_parent(self) if s is not None else None
def writable_structure(self):
return StructureWrapper(_gi_gst.event_writable_structure(self), self, True)
return StructureWrapper(_gi_gst.event_writable_structure(self)._set_parent(self))
Event = override(Event)
@ -182,11 +182,11 @@ __all__.append('NotWritableContext')
class Context(Gst.Context, MiniObject):
def get_structure(self):
return StructureWrapper(_gi_gst.context_get_structure(self), self,
False)
s = _gi_gst.context_get_structure(self)
return s._set_parent(self) if s is not None else None
def writable_structure(self):
return StructureWrapper(_gi_gst.context_writable_structure(self), self, True)
return StructureWrapper(_gi_gst.context_writable_structure(self)._set_parent(self))
Context = override(Context)
@ -251,11 +251,11 @@ class Caps(Gst.Caps, MiniObject):
return self.get_size()
def get_structure(self, index):
return StructureWrapper(_gi_gst.caps_get_structure(self, index), self,
False)
s = _gi_gst.caps_get_structure(self, index)
return s._set_parent(self) if s is not None else None
def writable_structure(self, index):
return StructureWrapper(_gi_gst.caps_writable_structure(self, index), self, True)
return StructureWrapper(_gi_gst.caps_writable_structure(self, index)._set_parent(self))
Caps = override(Caps)
@ -477,6 +477,16 @@ class Structure(Gst.Structure):
def __str__(self):
return self.to_string()
def _set_parent(self, parent):
self.__parent__ = parent
return self
def __enter__(self):
return self
def __exit__(self, _type, _value, _tb):
self._set_parent(None)
Structure = override(Structure)
__all__.append('Structure')
@ -851,16 +861,16 @@ def pairwise(iterable):
class StructureWrapper:
def __init__(self, structure, parent, writable):
"""A Gst.Structure wrapper to force usage of a context manager.
"""
def __init__(self, structure):
self.__structure = structure
self.__parent__ = parent
def __enter__(self):
return self.__structure
def __exit__(self, _type, _value, _tb):
self.__parent__ = False
return
self.__structure._set_parent(None)
class MapInfo:

View File

@ -182,6 +182,25 @@ class TestCaps(TestCase):
with caps.get_structure(0) as s:
self.assertNotEqual(ptr, s.__ptr__())
def test_read_no_copy_no_cm(self):
Gst.init(None)
caps = Gst.Caps("audio/x-raw")
s = caps.get_structure(0)
ptr = s.__ptr__()
s2 = caps.get_structure(0)
self.assertEqual(ptr, s2.__ptr__())
c2 = caps.mini_object
self.assertEqual(c2.refcount, 2)
with self.assertRaises(Gst.NotWritableStructure):
caps.get_structure(0).set_value("rate", 44100)
caps.make_writable()
s = caps.get_structure(0)
self.assertNotEqual(ptr, s.__ptr__())
class TestStructure(TestCase):
@ -207,6 +226,12 @@ class TestEvent(TestCase):
self.assertEqual(s["rate"], 44100)
self.assertEqual(s["channels"], 2)
def test_no_structure(self):
Gst.init(None)
event = Gst.Event.new_flush_start()
s = event.get_structure()
self.assertIsNone(s)
class TestQuery(TestCase):
def test_writable(self):