python-gitlab · isaac-philip · Feb 15, 2024 · JohnVillalovos · Apr 19, 2026 · Copilot
diff --git a/docs/gl_objects/merge_trains.rst b/docs/gl_objects/merge_trains.rst
@@ -9,6 +9,8 @@ Reference
 
   + :class:`gitlab.v4.objects.ProjectMergeTrain`
   + :class:`gitlab.v4.objects.ProjectMergeTrainManager`
+  + :class:`gitlab.v4.objects.ProjectMergeTrainMergeRequest`
+  + :class:`gitlab.v4.objects.ProjectMergeTrainMergeRequestManager`
   + :attr:`gitlab.v4.objects.Project.merge_trains`
 
 * GitLab API: https://docs.gitlab.com/api/merge_trains
@@ -27,3 +29,13 @@ List active merge trains for a project::
 List completed (have been merged) merge trains for a project::
 
     merge_trains = project.merge_trains.list(scope="complete")
+
+Get Merge Request Status for a Merge Train::
+
+    merge_train_mr = project.merge_trains.get(1, lazy=True).merge_requests.get(1)
+    merge_train_mr_status = merge_train_mr.pipeline.get("status")
+
+Add Merge Request to a Merge Train::
+
+    merge_train_to_update = project.merge_trains.get(1, lazy=True)
+    merge_requests_update = merge_train_to_update.merge_requests.update(5, new_data={"sha": "cd22awr721ssds"})
diff --git a/gitlab/base.py b/gitlab/base.py
@@ -7,7 +7,7 @@
 import textwrap
 from collections.abc import Iterable
 from types import ModuleType
-from typing import Any, ClassVar, Generic, TYPE_CHECKING, TypeVar
+from typing import Any, cast, ClassVar, Generic, TYPE_CHECKING, TypeVar
 
 import gitlab
 from gitlab import types as g_types
@@ -351,6 +351,7 @@ class RESTManager(Generic[TObjCls]):
     _path: ClassVar[str]
     _obj_cls: type[TObjCls]
     _from_parent_attrs: dict[str, Any] = {}
+    _parent_ref_attr: ClassVar[str | None] = None
     _types: dict[str, type[g_types.GitlabAttribute]] = {}
 
     _computed_path: str
@@ -389,6 +390,18 @@ def _compute_path(self, path: str | None = None) -> str:
         self._parent_attrs = data
         return path.format(**data)
 
+    def _get_parent_ref_id(self) -> int | str | None:
+        if self._parent is None or not self._parent_ref_attr:
+            return None
+        if not hasattr(self._parent, self._parent_ref_attr):
+            return None
+        parent_ref = getattr(self._parent, self._parent_ref_attr)
+        if parent_ref is None:
+            return None
+        if hasattr(parent_ref, "iid"):
+            return cast(int, parent_ref.iid)
-        if hasattr(parent_ref, "iid"):
-            return cast(int, parent_ref.iid)
+        if isinstance(parent_ref, dict) and "iid" in parent_ref:
+            return parent_ref["iid"]
+        if hasattr(parent_ref, "iid"):
+            return parent_ref.iid
-        if hasattr(parent_ref, "iid"):
-            return cast(int, parent_ref.iid)
+        if isinstance(parent_ref, dict) and "iid" in parent_ref:
+            return parent_ref["iid"]
+        if hasattr(parent_ref, "iid"):
+            return parent_ref.iid
+        return None
+
     @property
     def path(self) -> str:
         return self._computed_path
diff --git a/gitlab/mixins.py b/gitlab/mixins.py
@@ -72,11 +72,14 @@ class GetMixin(HeadMixin[base.TObjCls]):
     _optional_get_attrs: tuple[str, ...] = ()
 
     @exc.on_http_error(exc.GitlabGetError)
-    def get(self, id: str | int, lazy: bool = False, **kwargs: Any) -> base.TObjCls:
+    def get(
+        self, id: str | int | None = None, lazy: bool = False, **kwargs: Any
+    ) -> base.TObjCls:
         """Retrieve a single object.
 
         Args:
-            id: ID of the object to retrieve
+            id: ID of the object to retrieve. If not provided, falls back to
+                _parent_ref_id from the parent object (if available).
             lazy: If True, don't request the server, but create a
                          shallow object giving access to the managers. This is
                          useful if you want to avoid useless calls to the API.
-                _parent_ref_id from the parent object (if available).
-            lazy: If True, don't request the server, but create a
-                         shallow object giving access to the managers. This is
-                         useful if you want to avoid useless calls to the API.
+                the parent reference resolved by ``self._get_parent_ref_id()``
+                when available (for example, when ``_parent_ref_attr`` is set
+                on the manager).
+            lazy: If True, don't request the server, but create a shallow
+                object giving access to the managers. This is useful if you
+                want to avoid useless calls to the API.
-                _parent_ref_id from the parent object (if available).
-            lazy: If True, don't request the server, but create a
-                         shallow object giving access to the managers. This is
-                         useful if you want to avoid useless calls to the API.
+                the parent reference resolved by ``self._get_parent_ref_id()``
+                when available (for example, when ``_parent_ref_attr`` is set
+                on the manager).
+            lazy: If True, don't request the server, but create a shallow
+                object giving access to the managers. This is useful if you
+                want to avoid useless calls to the API.
@@ -89,6 +92,13 @@ def get(self, id: str | int, lazy: bool = False, **kwargs: Any) -> base.TObjCls:
             GitlabAuthenticationError: If authentication is not correct
             GitlabGetError: If the server cannot perform the request
         """
+        if id is None:
+            id = self._get_parent_ref_id()
+        if id is None:
+            raise ValueError(
+                "id is required. Either provide it explicitly or set "
+                "_parent_ref_attr on the manager to use the parent's reference."
+            )
         if isinstance(id, str):
             id = utils.EncodedId(id)
         path = f"{self.path}/{id}"
@@ -310,6 +320,8 @@ def update(
         """
         new_data = new_data or {}
 
+        if id is None:
+            id = self._get_parent_ref_id()
         if id is None:
             path = self.path
         else:

diff --git a/gitlab/v4/objects/merge_trains.py b/gitlab/v4/objects/merge_trains.py
@@ -1,14 +1,44 @@
-from gitlab.base import RESTObject
-from gitlab.mixins import ListMixin
+from gitlab.base import RESTManager, RESTObject
+from gitlab.mixins import GetMixin, ListMixin, UpdateMethod, UpdateMixin
+from gitlab.types import RequiredOptional
 
-__all__ = ["ProjectMergeTrain", "ProjectMergeTrainManager"]
+__all__ = [
+    "ProjectMergeTrain",
+    "ProjectMergeTrainManager",
+    "ProjectMergeTrainMergeRequest",
+    "ProjectMergeTrainMergeRequestManager",
+]
 
 
-class ProjectMergeTrain(RESTObject):
+class ProjectMergeTrainMergeRequest(RESTObject):
     pass
 
 
-class ProjectMergeTrainManager(ListMixin[ProjectMergeTrain]):
+class ProjectMergeTrainMergeRequestManager(
+    GetMixin[ProjectMergeTrainMergeRequest],
+    UpdateMixin[ProjectMergeTrainMergeRequest],
+    RESTManager[ProjectMergeTrainMergeRequest],
+):
+    _path = "/projects/{project_id}/merge_trains/merge_requests"
+    _obj_cls = ProjectMergeTrainMergeRequest
+    _from_parent_attrs = {"project_id": "project_id"}
+    _parent_ref_attr = "merge_request"
+    _update_method: UpdateMethod = UpdateMethod.POST
+
+    _update_attrs = RequiredOptional(
+        optional=("sha", "squash", "when_pipeline_succeeds", "auto_merge")
+    )
+
+
+class ProjectMergeTrain(RESTObject):
+    merge_requests: ProjectMergeTrainMergeRequestManager
+
+
+class ProjectMergeTrainManager(
+    GetMixin[ProjectMergeTrain],
+    ListMixin[ProjectMergeTrain],
+    RESTManager[ProjectMergeTrain],
+):
     _path = "/projects/{project_id}/merge_trains"
     _obj_cls = ProjectMergeTrain
     _from_parent_attrs = {"project_id": "id"}

diff --git a/tests/unit/mixins/test_mixin_methods.py b/tests/unit/mixins/test_mixin_methods.py
@@ -584,7 +584,6 @@ class TestClass(UploadMixin, FakeObject):
         url=url,
         json={"id": 42, "file_name": "test.txt", "file_content": "testing contents"},
         status=200,
-        match=[responses.matchers.query_param_matcher({})],
     )
 
     mgr = FakeManager(gl)
@@ -596,3 +595,101 @@ class TestClass(UploadMixin, FakeObject):
     assert res_only_path["file_name"] == "test.txt"
     assert res_only_path["file_content"] == "testing contents"
     assert responses.assert_call_count(url, 1) is True
+
+
+class MockParentRefWithIID:
+    def __init__(self, iid):
+        self.iid = iid
+
+
+class MockParentWithRef:
+    def __init__(self, parent_ref):
+        self.parent_ref = parent_ref
+
+
+class MockManagerWithRefAttr(base.RESTManager):
+    _path = "/tests/{test_id}/refs"
+    _obj_cls = FakeObject
+    _from_parent_attrs = {"test_id": "id"}
+    _parent_ref_attr = "parent_ref"
+
+
+def test_get_parent_ref_id_no_parent(gl):
+    class M(MockManagerWithRefAttr):
+        pass
+
+    mgr = M(gl)
+    assert mgr._get_parent_ref_id() is None
+
+
+def test_get_parent_ref_id_no_parent_ref_attr(gl):
+    class M(FakeManager):
+        pass
+
+    mgr = M(gl)
+    assert mgr._get_parent_ref_id() is None
+
+
+def test_get_parent_ref_id_parent_has_no_ref_attr(gl):
+    class M(FakeManager):
+        _parent_ref_attr = "nonexistent"
+
+    parent = MockParentWithRef(None)
+    mgr = M(gl, parent=parent)
+    assert mgr._get_parent_ref_id() is None
+
+
+def test_get_parent_ref_id_parent_ref_is_none(gl):
+    class M(MockManagerWithRefAttr):
+        pass
+
+    parent = MockParentWithRef(None)
+    mgr = M(gl, parent=parent)
+    assert mgr._get_parent_ref_id() is None
+
+
+def test_get_parent_ref_id_success(gl):
+    class M(MockManagerWithRefAttr):
+        pass
+
+    parent_ref = MockParentRefWithIID(42)
+    parent = MockParentWithRef(parent_ref)
+    mgr = M(gl, parent=parent)
+    assert mgr._get_parent_ref_id() == 42
+
+
+def test_get_parent_ref_id_no_iid_attribute(gl):
+    class MockParentRefNoIID:
+        pass
+
+    class M(MockManagerWithRefAttr):
+        pass
+
+    parent_ref = MockParentRefNoIID()
+    parent = MockParentWithRef(parent_ref)
+    mgr = M(gl, parent=parent)
+    assert mgr._get_parent_ref_id() is None
+
+
+def test_get_mixin_without_id_raises_error_when_no_parent_ref(gl):
+    class M(GetMixin, MockManagerWithRefAttr):
+        pass
+
+    mgr = M(gl)
+    with pytest.raises(ValueError, match="id is required"):
+        mgr.get()
+
+
+@responses.activate
+def test_update_mixin_without_id_no_parent_ref(gl):
+    class M(UpdateMixin, FakeManager):
+        _update_method = UpdateMethod.POST
+        _obj_cls = FakeObject
+
+    url = "http://localhost/api/v4/tests"
+    responses.add(method=responses.POST, url=url, json={}, status=200)
+
+    mgr = M(gl)
+    result = mgr.update(new_data={"foo": "bar"})
+    assert isinstance(result, dict)
+    assert responses.assert_call_count(url, 1) is True
diff --git a/tests/unit/objects/test_merge_trains.py b/tests/unit/objects/test_merge_trains.py
@@ -3,15 +3,17 @@
 https://docs.gitlab.com/ee/api/merge_trains.html
 """
 
+from copy import deepcopy
+
 import pytest
 import responses
 
-from gitlab.v4.objects import ProjectMergeTrain
+from gitlab.v4.objects import ProjectMergeTrain, ProjectMergeTrainMergeRequest
 
 mr_content = {
     "id": 110,
     "merge_request": {
-        "id": 1,
+        "id": 273,
         "iid": 1,
         "project_id": 3,
         "title": "Test merge train",
@@ -46,6 +48,10 @@
     "duration": 70,
 }
 
+merge_train_update = deepcopy(mr_content)
+merge_train_update["merge_request"]["iid"] = 4
+merge_train_update["pipeline"]["sha"] = "ef33a3zxc3"
+
 
 @pytest.fixture
 def resp_list_merge_trains():
@@ -60,7 +66,62 @@ def resp_list_merge_trains():
         yield rsps
 
 
+@pytest.fixture
+def resp_merge_trains_merge_request_get():
+    with responses.RequestsMock() as rsps:
+        rsps.add(
+            method=responses.GET,
+            url="http://localhost/api/v4/projects/1/merge_trains/merge_requests/1",
+            json=mr_content,
+            content_type="application/json",
+            status=200,
+        )
+        yield rsps
+
+
+@pytest.fixture
+def resp_merge_trains_merge_request_post():
+    with responses.RequestsMock() as rsps:
+        rsps.add(
+            method=responses.POST,
+            url="http://localhost/api/v4/projects/1/merge_trains/merge_requests/4",
+            json=[merge_train_update],
+            content_type="application/json",
+            status=200,
+            match=[responses.matchers.json_params_matcher({"sha": "ef33a3zxc3"})],
+        )
+        yield rsps
+
+
 def test_list_project_merge_requests(project, resp_list_merge_trains):
     merge_trains = project.merge_trains.list()
     assert isinstance(merge_trains[0], ProjectMergeTrain)
     assert merge_trains[0].id == mr_content["id"]
+
+
+def test_merge_trains_status_merge_request(
+    project, resp_merge_trains_merge_request_get
+):
+    merge_train_mr: ProjectMergeTrainMergeRequest = project.merge_trains.get(
+        1, lazy=True
+    ).merge_requests.get(1)
+    assert isinstance(merge_train_mr, ProjectMergeTrainMergeRequest)
+    assert merge_train_mr.get_id() == 110
+    assert merge_train_mr.merge_request["iid"] == mr_content["merge_request"]["iid"]
+    assert merge_train_mr.pipeline.get("status") == mr_content["pipeline"]["status"]
+
+
+def test_merge_train_add_merge_request(project, resp_merge_trains_merge_request_post):
+    merge_train: ProjectMergeTrain = project.merge_trains.get(1, lazy=True)
+    merge_requests_update = merge_train.merge_requests.update(
+        4, new_data={"sha": "ef33a3zxc3"}
+    )
+    assert isinstance(merge_train, ProjectMergeTrain)
+    assert (
+        merge_requests_update[0]["pipeline"]["sha"]
+        == merge_train_update["pipeline"]["sha"]
+    )
+    assert (
+        merge_requests_update[0]["merge_request"]["iid"]
+        == merge_train_update["merge_request"]["iid"]
+    )