higher-level APIs. To lessen the chances of a change to the interface impacting your code, we

recommend using keyword arguments when calling the interfaces.

At the lowest level, HTTP methods call ``http_request()``, which performs the actual request and takes

At the lowest level, HTTP methods call ``backend_request()``, which performs the actual request and takes

care of details such as timeouts, retries, and handling rate-limits.

This method can be invoked directly to or customize this behavior for a single request, or to call custom

Or get the total number of a user's events with a customized HEAD request:

..code-block::python

)

)

defhttp_request(

defbackend_request(

self,

verb:str,

path:str,

retry_transient_errors:Optional[bool]=None,

max_retries:int=10,

**kwargs:Any,

)->requests.Response:

)->_backends.DefaultResponse:

whileTrue:

try:

result=self._backend.http_request(

backend_response=self._backend.http_request(

method=verb,

url=url,

json=send_data.json,

self._check_redirects(result.response)

self._check_redirects(backend_response.response)

if200<=result.status_code<300:

if200<=backend_response.status_code<300:

defshould_retry()->bool:

ifresult.status_code==429andobey_rate_limit:

ifbackend_response.status_code==429andobey_rate_limit:

ifnotretry_transient_errors:

ifresult.status_codeingitlab.const.RETRYABLE_TRANSIENT_ERROR_CODES:

if (

backend_response.status_code

ingitlab.const.RETRYABLE_TRANSIENT_ERROR_CODES

):

ifresult.status_code==409and"Resource lock"inresult.reason:

if (

backend_response.status_code==409

and"Resource lock"inbackend_response.reason

):

ifmax_retries==-1orcur_retries<max_retries:

if"Retry-After"inresult.headers:

wait_time=int(result.headers["Retry-After"])

elif"RateLimit-Reset"inresult.headers:

wait_time=int(result.headers["RateLimit-Reset"])-time.time()

if"Retry-After"inbackend_response.headers:

wait_time=int(backend_response.headers["Retry-After"])

elif"RateLimit-Reset"inbackend_response.headers:

wait_time= (

int(backend_response.headers["RateLimit-Reset"])

-time.time()

)

time.sleep(wait_time)

error_message=result.content

error_message=backend_response.content

try:

error_json=result.json()

error_json=backend_response.json()

forkin ("message","error"):

ifkinerror_json:

error_message=error_json[k]

except (KeyError,ValueError,TypeError):

ifresult.status_code==401:

ifbackend_response.status_code==401:

raisegitlab.exceptions.GitlabAuthenticationError(

response_code=result.status_code,

response_code=backend_response.status_code,

error_message=error_message,

response_body=result.content,

response_body=backend_response.content,

)

raisegitlab.exceptions.GitlabHttpError(

response_code=result.status_code,

response_code=backend_response.status_code,

error_message=error_message,

response_body=result.content,

response_body=backend_response.content,

)

defhttp_request(

self,

verb:str,

path:str,

query_data:Optional[Dict[str,Any]]=None,

post_data:Optional[Union[Dict[str,Any],bytes]]=None,

raw:bool=False,

streamed:bool=False,

files:Optional[Dict[str,Any]]=None,

timeout:Optional[float]=None,

obey_rate_limit:bool=True,

retry_transient_errors:Optional[bool]=None,

max_retries:int=10,

**kwargs:Any,

)->requests.Response:

utils.warn(

"Please use `backend_request()` instead.",

category=DeprecationWarning,

)

returnself.backend_request(

verb,

path,

query_data,

post_data,

raw,

streamed,

files,

timeout,

obey_rate_limit,

retry_transient_errors,

max_retries,

**kwargs,

).response

defhttp_get(

self,

path:str,

query_data=query_dataor {}

result=self.http_request(

backend_response=self.backend_request(

"get",path,query_data=query_data,streamed=streamed,**kwargs

)

content_type=utils.get_content_type(result.headers.get("Content-Type"))

content_type=utils.get_content_type(

backend_response.headers.get("Content-Type")

)

result=backend_response.response

ifcontent_type=="application/json"andnotstreamedandnotraw:

try:

query_data=query_dataor {}

result=self.http_request("head",path,query_data=query_data,**kwargs)

returnresult.headers

backend_response=self.http_request(

"head",path,query_data=query_data,**kwargs

)

returnbackend_response.headers

defhttp_list(

self,

query_data=query_dataor {}

post_data=post_dataor {}

result=self.http_request(

backend_response=self.backend_request(

"post",

path,

query_data=query_data,

raw=raw,

**kwargs,

)

content_type=utils.get_content_type(result.headers.get("Content-Type"))

content_type=utils.get_content_type(

backend_response.headers.get("Content-Type")

)

result=backend_response.response

try:

ifcontent_type=="application/json":

query_data=query_dataor {}

post_data=post_dataor {}

result=self.http_request(

backend_response=self.http_request(

"put",

path,

query_data=query_data,

**kwargs,

)

try:

json_result=result.json()

json_result=backend_response.json()

ifTYPE_CHECKING:

assertisinstance(json_result,dict)

query_data=query_dataor {}

post_data=post_dataor {}

result=self.http_request(

backend_response=self.http_request(

"patch",

path,

query_data=query_data,

**kwargs,

)

try:

json_result=result.json()

json_result=backend_response.json()

ifTYPE_CHECKING:

assertisinstance(json_result,dict)

returnself.http_request("delete",path,**kwargs)

backend_response=self.backend_request("delete",path,**kwargs)

returnbackend_response.response

defsearch(

self,url:str,query_data:Optional[Dict[str,Any]]=None,**kwargs:Any

)->None:

query_data=query_dataor {}

result=self._gl.http_request("get",url,query_data=query_data,**kwargs)

backend_response=self._gl.backend_request(

"get",url,query_data=query_data,**kwargs

)

result=backend_response.response

try:

next_url=result.links["next"]["url"]

exceptKeyError: