1+ from typing import Any ,cast ,Dict ,Optional ,TYPE_CHECKING ,Union
2+
3+ import requests
4+
15from gitlab import cli
26from gitlab import exceptions as exc
37from gitlab .base import RequiredOptional ,RESTManager ,RESTObject
@@ -52,7 +56,7 @@ class ProjectPipeline(RefreshMixin, ObjectDeleteMixin, RESTObject):
5256
5357@cli .register_custom_action ("ProjectPipeline" )
5458@exc .on_http_error (exc .GitlabPipelineCancelError )
55- def cancel (self ,** kwargs ) :
59+ def cancel (self ,** kwargs : Any ) -> Union [ Dict [ str , Any ], requests . Response ] :
5660"""Cancel the job.
5761
5862 Args:
@@ -67,7 +71,7 @@ def cancel(self, **kwargs):
6771
6872@cli .register_custom_action ("ProjectPipeline" )
6973@exc .on_http_error (exc .GitlabPipelineRetryError )
70- def retry (self ,** kwargs ) :
74+ def retry (self ,** kwargs : Any ) -> Union [ Dict [ str , Any ], requests . Response ] :
7175"""Retry the job.
7276
7377 Args:
@@ -98,7 +102,14 @@ class ProjectPipelineManager(RetrieveMixin, CreateMixin, DeleteMixin, RESTManage
98102 )
99103_create_attrs = RequiredOptional (required = ("ref" ,))
100104
101- def create (self ,data ,** kwargs ):
105+ def get (
106+ self ,id :Union [str ,int ],lazy :bool = False ,** kwargs :Any
107+ )-> ProjectPipeline :
108+ return cast (ProjectPipeline ,super ().get (id = id ,lazy = lazy ,** kwargs ))
109+
110+ def create (
111+ self ,data :Optional [Dict [str ,Any ]]= None ,** kwargs :Any
112+ )-> ProjectPipeline :
102113"""Creates a new object.
103114
104115 Args:
@@ -114,8 +125,12 @@ def create(self, data, **kwargs):
114125 RESTObject: A new instance of the managed object class build with
115126 the data sent by the server
116127 """
128+ if TYPE_CHECKING :
129+ assert self .path is not None
117130path = self .path [:- 1 ]# drop the 's'
118- return CreateMixin .create (self ,data ,path = path ,** kwargs )
131+ return cast (
132+ ProjectPipeline ,CreateMixin .create (self ,data ,path = path ,** kwargs )
133+ )
119134
120135
121136class ProjectPipelineJob (RESTObject ):
@@ -169,7 +184,7 @@ class ProjectPipelineSchedule(SaveMixin, ObjectDeleteMixin, RESTObject):
169184
170185@cli .register_custom_action ("ProjectPipelineSchedule" )
171186@exc .on_http_error (exc .GitlabOwnershipError )
172- def take_ownership (self ,** kwargs ) :
187+ def take_ownership (self ,** kwargs : Any ) -> None :
173188"""Update the owner of a pipeline schedule.
174189
175190 Args:
@@ -181,11 +196,13 @@ def take_ownership(self, **kwargs):
181196 """
182197path = f"{ self .manager .path } { self .get_id ()}
183198server_data = self .manager .gitlab .http_post (path ,** kwargs )
199+ if TYPE_CHECKING :
200+ assert isinstance (server_data ,dict )
184201self ._update_attrs (server_data )
185202
186203@cli .register_custom_action ("ProjectPipelineSchedule" )
187204@exc .on_http_error (exc .GitlabPipelinePlayError )
188- def play (self ,** kwargs ) :
205+ def play (self ,** kwargs : Any ) -> Dict [ str , Any ] :
189206"""Trigger a new scheduled pipeline, which runs immediately.
190207 The next scheduled run of this pipeline is not affected.
191208
@@ -198,6 +215,8 @@ def play(self, **kwargs):
198215 """
199216path = f"{ self .manager .path } { self .get_id ()}
200217server_data = self .manager .gitlab .http_post (path ,** kwargs )
218+ if TYPE_CHECKING :
219+ assert isinstance (server_data ,dict )
201220self ._update_attrs (server_data )
202221return server_data
203222
@@ -213,6 +232,11 @@ class ProjectPipelineScheduleManager(CRUDMixin, RESTManager):
213232optional = ("description" ,"ref" ,"cron" ,"cron_timezone" ,"active" ),
214233 )
215234
235+ def get (
236+ self ,id :Union [str ,int ],lazy :bool = False ,** kwargs :Any
237+ )-> ProjectPipelineSchedule :
238+ return cast (ProjectPipelineSchedule ,super ().get (id = id ,lazy = lazy ,** kwargs ))
239+
216240
217241class ProjectPipelineTestReport (RESTObject ):
218242_id_attr = None