1# coding=utf-8 2# -------------------------------------------------------------------------- 3# Copyright (c) Microsoft Corporation. All rights reserved. 4# Licensed under the MIT License. See License.txt in the project root for license information. 5# Code generated by Microsoft (R) AutoRest Code Generator. 6# Changes may cause incorrect behavior and will be lost if the code is regenerated. 7# -------------------------------------------------------------------------- 8from typing import Any, AsyncIterable, Callable, Dict, Generic, Optional, TypeVar, Union 9import warnings 10 11from azure.core.async_paging import AsyncItemPaged, AsyncList 12from azure.core.exceptions import ClientAuthenticationError, HttpResponseError, ResourceExistsError, ResourceNotFoundError, map_error 13from azure.core.pipeline import PipelineResponse 14from azure.core.pipeline.transport import AsyncHttpResponse, HttpRequest 15from azure.core.polling import AsyncLROPoller, AsyncNoPolling, AsyncPollingMethod 16from azure.mgmt.core.exceptions import ARMErrorFormat 17from azure.mgmt.core.polling.async_arm_polling import AsyncARMPolling 18 19from ... import models as _models 20 21T = TypeVar('T') 22ClsType = Optional[Callable[[PipelineResponse[HttpRequest, AsyncHttpResponse], T, Dict[str, Any]], Any]] 23 24class TaskRunsOperations: 25 """TaskRunsOperations async operations. 26 27 You should not instantiate this class directly. Instead, you should create a Client instance that 28 instantiates it for you and attaches it as an attribute. 29 30 :ivar models: Alias to model classes used in this operation group. 31 :type models: ~azure.mgmt.containerregistry.v2019_06_01_preview.models 32 :param client: Client for service requests. 33 :param config: Configuration of service client. 34 :param serializer: An object model serializer. 35 :param deserializer: An object model deserializer. 36 """ 37 38 models = _models 39 40 def __init__(self, client, config, serializer, deserializer) -> None: 41 self._client = client 42 self._serialize = serializer 43 self._deserialize = deserializer 44 self._config = config 45 46 async def get( 47 self, 48 resource_group_name: str, 49 registry_name: str, 50 task_run_name: str, 51 **kwargs: Any 52 ) -> "_models.TaskRun": 53 """Gets the detailed information for a given task run. 54 55 :param resource_group_name: The name of the resource group to which the container registry 56 belongs. 57 :type resource_group_name: str 58 :param registry_name: The name of the container registry. 59 :type registry_name: str 60 :param task_run_name: The name of the task run. 61 :type task_run_name: str 62 :keyword callable cls: A custom type or function that will be passed the direct response 63 :return: TaskRun, or the result of cls(response) 64 :rtype: ~azure.mgmt.containerregistry.v2019_06_01_preview.models.TaskRun 65 :raises: ~azure.core.exceptions.HttpResponseError 66 """ 67 cls = kwargs.pop('cls', None) # type: ClsType["_models.TaskRun"] 68 error_map = { 69 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 70 } 71 error_map.update(kwargs.pop('error_map', {})) 72 api_version = "2019-06-01-preview" 73 accept = "application/json" 74 75 # Construct URL 76 url = self.get.metadata['url'] # type: ignore 77 path_format_arguments = { 78 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 79 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str', min_length=1), 80 'registryName': self._serialize.url("registry_name", registry_name, 'str', max_length=50, min_length=5, pattern=r'^[a-zA-Z0-9]*$'), 81 'taskRunName': self._serialize.url("task_run_name", task_run_name, 'str', max_length=50, min_length=5, pattern=r'^[a-zA-Z0-9-]*$'), 82 } 83 url = self._client.format_url(url, **path_format_arguments) 84 85 # Construct parameters 86 query_parameters = {} # type: Dict[str, Any] 87 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 88 89 # Construct headers 90 header_parameters = {} # type: Dict[str, Any] 91 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 92 93 request = self._client.get(url, query_parameters, header_parameters) 94 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 95 response = pipeline_response.http_response 96 97 if response.status_code not in [200]: 98 map_error(status_code=response.status_code, response=response, error_map=error_map) 99 error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, response) 100 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) 101 102 deserialized = self._deserialize('TaskRun', pipeline_response) 103 104 if cls: 105 return cls(pipeline_response, deserialized, {}) 106 107 return deserialized 108 get.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerRegistry/registries/{registryName}/taskRuns/{taskRunName}'} # type: ignore 109 110 async def _create_initial( 111 self, 112 resource_group_name: str, 113 registry_name: str, 114 task_run_name: str, 115 task_run: "_models.TaskRun", 116 **kwargs: Any 117 ) -> "_models.TaskRun": 118 cls = kwargs.pop('cls', None) # type: ClsType["_models.TaskRun"] 119 error_map = { 120 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 121 } 122 error_map.update(kwargs.pop('error_map', {})) 123 api_version = "2019-06-01-preview" 124 content_type = kwargs.pop("content_type", "application/json") 125 accept = "application/json" 126 127 # Construct URL 128 url = self._create_initial.metadata['url'] # type: ignore 129 path_format_arguments = { 130 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 131 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str', min_length=1), 132 'registryName': self._serialize.url("registry_name", registry_name, 'str', max_length=50, min_length=5, pattern=r'^[a-zA-Z0-9]*$'), 133 'taskRunName': self._serialize.url("task_run_name", task_run_name, 'str', max_length=50, min_length=5, pattern=r'^[a-zA-Z0-9-]*$'), 134 } 135 url = self._client.format_url(url, **path_format_arguments) 136 137 # Construct parameters 138 query_parameters = {} # type: Dict[str, Any] 139 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 140 141 # Construct headers 142 header_parameters = {} # type: Dict[str, Any] 143 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 144 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 145 146 body_content_kwargs = {} # type: Dict[str, Any] 147 body_content = self._serialize.body(task_run, 'TaskRun') 148 body_content_kwargs['content'] = body_content 149 request = self._client.put(url, query_parameters, header_parameters, **body_content_kwargs) 150 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 151 response = pipeline_response.http_response 152 153 if response.status_code not in [200, 201]: 154 map_error(status_code=response.status_code, response=response, error_map=error_map) 155 error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, response) 156 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) 157 158 if response.status_code == 200: 159 deserialized = self._deserialize('TaskRun', pipeline_response) 160 161 if response.status_code == 201: 162 deserialized = self._deserialize('TaskRun', pipeline_response) 163 164 if cls: 165 return cls(pipeline_response, deserialized, {}) 166 167 return deserialized 168 _create_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerRegistry/registries/{registryName}/taskRuns/{taskRunName}'} # type: ignore 169 170 async def begin_create( 171 self, 172 resource_group_name: str, 173 registry_name: str, 174 task_run_name: str, 175 task_run: "_models.TaskRun", 176 **kwargs: Any 177 ) -> AsyncLROPoller["_models.TaskRun"]: 178 """Creates a task run for a container registry with the specified parameters. 179 180 :param resource_group_name: The name of the resource group to which the container registry 181 belongs. 182 :type resource_group_name: str 183 :param registry_name: The name of the container registry. 184 :type registry_name: str 185 :param task_run_name: The name of the task run. 186 :type task_run_name: str 187 :param task_run: The parameters of a run that needs to scheduled. 188 :type task_run: ~azure.mgmt.containerregistry.v2019_06_01_preview.models.TaskRun 189 :keyword callable cls: A custom type or function that will be passed the direct response 190 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 191 :keyword polling: By default, your polling method will be AsyncARMPolling. 192 Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy. 193 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 194 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 195 :return: An instance of AsyncLROPoller that returns either TaskRun or the result of cls(response) 196 :rtype: ~azure.core.polling.AsyncLROPoller[~azure.mgmt.containerregistry.v2019_06_01_preview.models.TaskRun] 197 :raises ~azure.core.exceptions.HttpResponseError: 198 """ 199 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 200 cls = kwargs.pop('cls', None) # type: ClsType["_models.TaskRun"] 201 lro_delay = kwargs.pop( 202 'polling_interval', 203 self._config.polling_interval 204 ) 205 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 206 if cont_token is None: 207 raw_result = await self._create_initial( 208 resource_group_name=resource_group_name, 209 registry_name=registry_name, 210 task_run_name=task_run_name, 211 task_run=task_run, 212 cls=lambda x,y,z: x, 213 **kwargs 214 ) 215 216 kwargs.pop('error_map', None) 217 kwargs.pop('content_type', None) 218 219 def get_long_running_output(pipeline_response): 220 deserialized = self._deserialize('TaskRun', pipeline_response) 221 222 if cls: 223 return cls(pipeline_response, deserialized, {}) 224 return deserialized 225 226 path_format_arguments = { 227 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 228 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str', min_length=1), 229 'registryName': self._serialize.url("registry_name", registry_name, 'str', max_length=50, min_length=5, pattern=r'^[a-zA-Z0-9]*$'), 230 'taskRunName': self._serialize.url("task_run_name", task_run_name, 'str', max_length=50, min_length=5, pattern=r'^[a-zA-Z0-9-]*$'), 231 } 232 233 if polling is True: polling_method = AsyncARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) 234 elif polling is False: polling_method = AsyncNoPolling() 235 else: polling_method = polling 236 if cont_token: 237 return AsyncLROPoller.from_continuation_token( 238 polling_method=polling_method, 239 continuation_token=cont_token, 240 client=self._client, 241 deserialization_callback=get_long_running_output 242 ) 243 else: 244 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 245 begin_create.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerRegistry/registries/{registryName}/taskRuns/{taskRunName}'} # type: ignore 246 247 async def _delete_initial( 248 self, 249 resource_group_name: str, 250 registry_name: str, 251 task_run_name: str, 252 **kwargs: Any 253 ) -> None: 254 cls = kwargs.pop('cls', None) # type: ClsType[None] 255 error_map = { 256 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 257 } 258 error_map.update(kwargs.pop('error_map', {})) 259 api_version = "2019-06-01-preview" 260 accept = "application/json" 261 262 # Construct URL 263 url = self._delete_initial.metadata['url'] # type: ignore 264 path_format_arguments = { 265 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 266 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str', min_length=1), 267 'registryName': self._serialize.url("registry_name", registry_name, 'str', max_length=50, min_length=5, pattern=r'^[a-zA-Z0-9]*$'), 268 'taskRunName': self._serialize.url("task_run_name", task_run_name, 'str', max_length=50, min_length=5, pattern=r'^[a-zA-Z0-9-]*$'), 269 } 270 url = self._client.format_url(url, **path_format_arguments) 271 272 # Construct parameters 273 query_parameters = {} # type: Dict[str, Any] 274 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 275 276 # Construct headers 277 header_parameters = {} # type: Dict[str, Any] 278 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 279 280 request = self._client.delete(url, query_parameters, header_parameters) 281 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 282 response = pipeline_response.http_response 283 284 if response.status_code not in [200, 202, 204]: 285 map_error(status_code=response.status_code, response=response, error_map=error_map) 286 error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, response) 287 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) 288 289 if cls: 290 return cls(pipeline_response, None, {}) 291 292 _delete_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerRegistry/registries/{registryName}/taskRuns/{taskRunName}'} # type: ignore 293 294 async def begin_delete( 295 self, 296 resource_group_name: str, 297 registry_name: str, 298 task_run_name: str, 299 **kwargs: Any 300 ) -> AsyncLROPoller[None]: 301 """Deletes a specified task run resource. 302 303 :param resource_group_name: The name of the resource group to which the container registry 304 belongs. 305 :type resource_group_name: str 306 :param registry_name: The name of the container registry. 307 :type registry_name: str 308 :param task_run_name: The name of the task run. 309 :type task_run_name: str 310 :keyword callable cls: A custom type or function that will be passed the direct response 311 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 312 :keyword polling: By default, your polling method will be AsyncARMPolling. 313 Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy. 314 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 315 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 316 :return: An instance of AsyncLROPoller that returns either None or the result of cls(response) 317 :rtype: ~azure.core.polling.AsyncLROPoller[None] 318 :raises ~azure.core.exceptions.HttpResponseError: 319 """ 320 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 321 cls = kwargs.pop('cls', None) # type: ClsType[None] 322 lro_delay = kwargs.pop( 323 'polling_interval', 324 self._config.polling_interval 325 ) 326 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 327 if cont_token is None: 328 raw_result = await self._delete_initial( 329 resource_group_name=resource_group_name, 330 registry_name=registry_name, 331 task_run_name=task_run_name, 332 cls=lambda x,y,z: x, 333 **kwargs 334 ) 335 336 kwargs.pop('error_map', None) 337 kwargs.pop('content_type', None) 338 339 def get_long_running_output(pipeline_response): 340 if cls: 341 return cls(pipeline_response, None, {}) 342 343 path_format_arguments = { 344 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 345 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str', min_length=1), 346 'registryName': self._serialize.url("registry_name", registry_name, 'str', max_length=50, min_length=5, pattern=r'^[a-zA-Z0-9]*$'), 347 'taskRunName': self._serialize.url("task_run_name", task_run_name, 'str', max_length=50, min_length=5, pattern=r'^[a-zA-Z0-9-]*$'), 348 } 349 350 if polling is True: polling_method = AsyncARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) 351 elif polling is False: polling_method = AsyncNoPolling() 352 else: polling_method = polling 353 if cont_token: 354 return AsyncLROPoller.from_continuation_token( 355 polling_method=polling_method, 356 continuation_token=cont_token, 357 client=self._client, 358 deserialization_callback=get_long_running_output 359 ) 360 else: 361 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 362 begin_delete.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerRegistry/registries/{registryName}/taskRuns/{taskRunName}'} # type: ignore 363 364 async def _update_initial( 365 self, 366 resource_group_name: str, 367 registry_name: str, 368 task_run_name: str, 369 update_parameters: "_models.TaskRunUpdateParameters", 370 **kwargs: Any 371 ) -> "_models.TaskRun": 372 cls = kwargs.pop('cls', None) # type: ClsType["_models.TaskRun"] 373 error_map = { 374 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 375 } 376 error_map.update(kwargs.pop('error_map', {})) 377 api_version = "2019-06-01-preview" 378 content_type = kwargs.pop("content_type", "application/json") 379 accept = "application/json" 380 381 # Construct URL 382 url = self._update_initial.metadata['url'] # type: ignore 383 path_format_arguments = { 384 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 385 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str', min_length=1), 386 'registryName': self._serialize.url("registry_name", registry_name, 'str', max_length=50, min_length=5, pattern=r'^[a-zA-Z0-9]*$'), 387 'taskRunName': self._serialize.url("task_run_name", task_run_name, 'str', max_length=50, min_length=5, pattern=r'^[a-zA-Z0-9-]*$'), 388 } 389 url = self._client.format_url(url, **path_format_arguments) 390 391 # Construct parameters 392 query_parameters = {} # type: Dict[str, Any] 393 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 394 395 # Construct headers 396 header_parameters = {} # type: Dict[str, Any] 397 header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') 398 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 399 400 body_content_kwargs = {} # type: Dict[str, Any] 401 body_content = self._serialize.body(update_parameters, 'TaskRunUpdateParameters') 402 body_content_kwargs['content'] = body_content 403 request = self._client.patch(url, query_parameters, header_parameters, **body_content_kwargs) 404 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 405 response = pipeline_response.http_response 406 407 if response.status_code not in [200, 201]: 408 map_error(status_code=response.status_code, response=response, error_map=error_map) 409 error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, response) 410 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) 411 412 if response.status_code == 200: 413 deserialized = self._deserialize('TaskRun', pipeline_response) 414 415 if response.status_code == 201: 416 deserialized = self._deserialize('TaskRun', pipeline_response) 417 418 if cls: 419 return cls(pipeline_response, deserialized, {}) 420 421 return deserialized 422 _update_initial.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerRegistry/registries/{registryName}/taskRuns/{taskRunName}'} # type: ignore 423 424 async def begin_update( 425 self, 426 resource_group_name: str, 427 registry_name: str, 428 task_run_name: str, 429 update_parameters: "_models.TaskRunUpdateParameters", 430 **kwargs: Any 431 ) -> AsyncLROPoller["_models.TaskRun"]: 432 """Updates a task run with the specified parameters. 433 434 :param resource_group_name: The name of the resource group to which the container registry 435 belongs. 436 :type resource_group_name: str 437 :param registry_name: The name of the container registry. 438 :type registry_name: str 439 :param task_run_name: The name of the task run. 440 :type task_run_name: str 441 :param update_parameters: The parameters for updating a task run. 442 :type update_parameters: ~azure.mgmt.containerregistry.v2019_06_01_preview.models.TaskRunUpdateParameters 443 :keyword callable cls: A custom type or function that will be passed the direct response 444 :keyword str continuation_token: A continuation token to restart a poller from a saved state. 445 :keyword polling: By default, your polling method will be AsyncARMPolling. 446 Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy. 447 :paramtype polling: bool or ~azure.core.polling.AsyncPollingMethod 448 :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. 449 :return: An instance of AsyncLROPoller that returns either TaskRun or the result of cls(response) 450 :rtype: ~azure.core.polling.AsyncLROPoller[~azure.mgmt.containerregistry.v2019_06_01_preview.models.TaskRun] 451 :raises ~azure.core.exceptions.HttpResponseError: 452 """ 453 polling = kwargs.pop('polling', True) # type: Union[bool, AsyncPollingMethod] 454 cls = kwargs.pop('cls', None) # type: ClsType["_models.TaskRun"] 455 lro_delay = kwargs.pop( 456 'polling_interval', 457 self._config.polling_interval 458 ) 459 cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] 460 if cont_token is None: 461 raw_result = await self._update_initial( 462 resource_group_name=resource_group_name, 463 registry_name=registry_name, 464 task_run_name=task_run_name, 465 update_parameters=update_parameters, 466 cls=lambda x,y,z: x, 467 **kwargs 468 ) 469 470 kwargs.pop('error_map', None) 471 kwargs.pop('content_type', None) 472 473 def get_long_running_output(pipeline_response): 474 deserialized = self._deserialize('TaskRun', pipeline_response) 475 476 if cls: 477 return cls(pipeline_response, deserialized, {}) 478 return deserialized 479 480 path_format_arguments = { 481 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 482 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str', min_length=1), 483 'registryName': self._serialize.url("registry_name", registry_name, 'str', max_length=50, min_length=5, pattern=r'^[a-zA-Z0-9]*$'), 484 'taskRunName': self._serialize.url("task_run_name", task_run_name, 'str', max_length=50, min_length=5, pattern=r'^[a-zA-Z0-9-]*$'), 485 } 486 487 if polling is True: polling_method = AsyncARMPolling(lro_delay, path_format_arguments=path_format_arguments, **kwargs) 488 elif polling is False: polling_method = AsyncNoPolling() 489 else: polling_method = polling 490 if cont_token: 491 return AsyncLROPoller.from_continuation_token( 492 polling_method=polling_method, 493 continuation_token=cont_token, 494 client=self._client, 495 deserialization_callback=get_long_running_output 496 ) 497 else: 498 return AsyncLROPoller(self._client, raw_result, get_long_running_output, polling_method) 499 begin_update.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerRegistry/registries/{registryName}/taskRuns/{taskRunName}'} # type: ignore 500 501 async def get_details( 502 self, 503 resource_group_name: str, 504 registry_name: str, 505 task_run_name: str, 506 **kwargs: Any 507 ) -> "_models.TaskRun": 508 """Gets the detailed information for a given task run that includes all secrets. 509 510 :param resource_group_name: The name of the resource group to which the container registry 511 belongs. 512 :type resource_group_name: str 513 :param registry_name: The name of the container registry. 514 :type registry_name: str 515 :param task_run_name: The name of the task run. 516 :type task_run_name: str 517 :keyword callable cls: A custom type or function that will be passed the direct response 518 :return: TaskRun, or the result of cls(response) 519 :rtype: ~azure.mgmt.containerregistry.v2019_06_01_preview.models.TaskRun 520 :raises: ~azure.core.exceptions.HttpResponseError 521 """ 522 cls = kwargs.pop('cls', None) # type: ClsType["_models.TaskRun"] 523 error_map = { 524 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 525 } 526 error_map.update(kwargs.pop('error_map', {})) 527 api_version = "2019-06-01-preview" 528 accept = "application/json" 529 530 # Construct URL 531 url = self.get_details.metadata['url'] # type: ignore 532 path_format_arguments = { 533 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 534 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str', min_length=1), 535 'registryName': self._serialize.url("registry_name", registry_name, 'str', max_length=50, min_length=5, pattern=r'^[a-zA-Z0-9]*$'), 536 'taskRunName': self._serialize.url("task_run_name", task_run_name, 'str', max_length=50, min_length=5, pattern=r'^[a-zA-Z0-9-]*$'), 537 } 538 url = self._client.format_url(url, **path_format_arguments) 539 540 # Construct parameters 541 query_parameters = {} # type: Dict[str, Any] 542 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 543 544 # Construct headers 545 header_parameters = {} # type: Dict[str, Any] 546 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 547 548 request = self._client.post(url, query_parameters, header_parameters) 549 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 550 response = pipeline_response.http_response 551 552 if response.status_code not in [200]: 553 map_error(status_code=response.status_code, response=response, error_map=error_map) 554 error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, response) 555 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) 556 557 deserialized = self._deserialize('TaskRun', pipeline_response) 558 559 if cls: 560 return cls(pipeline_response, deserialized, {}) 561 562 return deserialized 563 get_details.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerRegistry/registries/{registryName}/taskRuns/{taskRunName}/listDetails'} # type: ignore 564 565 def list( 566 self, 567 resource_group_name: str, 568 registry_name: str, 569 **kwargs: Any 570 ) -> AsyncIterable["_models.TaskRunListResult"]: 571 """Lists all the task runs for a specified container registry. 572 573 :param resource_group_name: The name of the resource group to which the container registry 574 belongs. 575 :type resource_group_name: str 576 :param registry_name: The name of the container registry. 577 :type registry_name: str 578 :keyword callable cls: A custom type or function that will be passed the direct response 579 :return: An iterator like instance of either TaskRunListResult or the result of cls(response) 580 :rtype: ~azure.core.async_paging.AsyncItemPaged[~azure.mgmt.containerregistry.v2019_06_01_preview.models.TaskRunListResult] 581 :raises: ~azure.core.exceptions.HttpResponseError 582 """ 583 cls = kwargs.pop('cls', None) # type: ClsType["_models.TaskRunListResult"] 584 error_map = { 585 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError 586 } 587 error_map.update(kwargs.pop('error_map', {})) 588 api_version = "2019-06-01-preview" 589 accept = "application/json" 590 591 def prepare_request(next_link=None): 592 # Construct headers 593 header_parameters = {} # type: Dict[str, Any] 594 header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') 595 596 if not next_link: 597 # Construct URL 598 url = self.list.metadata['url'] # type: ignore 599 path_format_arguments = { 600 'subscriptionId': self._serialize.url("self._config.subscription_id", self._config.subscription_id, 'str'), 601 'resourceGroupName': self._serialize.url("resource_group_name", resource_group_name, 'str', min_length=1), 602 'registryName': self._serialize.url("registry_name", registry_name, 'str', max_length=50, min_length=5, pattern=r'^[a-zA-Z0-9]*$'), 603 } 604 url = self._client.format_url(url, **path_format_arguments) 605 # Construct parameters 606 query_parameters = {} # type: Dict[str, Any] 607 query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') 608 609 request = self._client.get(url, query_parameters, header_parameters) 610 else: 611 url = next_link 612 query_parameters = {} # type: Dict[str, Any] 613 request = self._client.get(url, query_parameters, header_parameters) 614 return request 615 616 async def extract_data(pipeline_response): 617 deserialized = self._deserialize('TaskRunListResult', pipeline_response) 618 list_of_elem = deserialized.value 619 if cls: 620 list_of_elem = cls(list_of_elem) 621 return deserialized.next_link or None, AsyncList(list_of_elem) 622 623 async def get_next(next_link=None): 624 request = prepare_request(next_link) 625 626 pipeline_response = await self._client._pipeline.run(request, stream=False, **kwargs) 627 response = pipeline_response.http_response 628 629 if response.status_code not in [200]: 630 error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, response) 631 map_error(status_code=response.status_code, response=response, error_map=error_map) 632 raise HttpResponseError(response=response, model=error, error_format=ARMErrorFormat) 633 634 return pipeline_response 635 636 return AsyncItemPaged( 637 get_next, extract_data 638 ) 639 list.metadata = {'url': '/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ContainerRegistry/registries/{registryName}/taskRuns'} # type: ignore 640