service_endpoint_client.py

# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------
# Generated file, DO NOT EDIT
# Changes may cause incorrect behavior and will be lost if the code is regenerated.
# --------------------------------------------------------------------------------------------

from azure.core.rest import HttpRequest, HttpResponse
from ...client import Client
from ...v7_0.service_endpoint import models


class ServiceEndpointClient(Client):
    """ServiceEndpoint
    :param str base_url: Service URL
    :param Authentication creds: Authenticated credentials.
    """

    def __init__(self, base_url=None, creds=None):
        super(ServiceEndpointClient, self).__init__(base_url, creds)
        client_models = {k: v for k, v in models.__dict__.items() if isinstance(v, type)}
        self._serialize = Serializer(client_models)
        self._deserialize = Deserializer(client_models)

    resource_area_identifier = '1814ab31-2f4f-4a9f-8761-f4d77dc5a5d7'

    def execute_service_endpoint_request(self, service_endpoint_request, project, endpoint_id):
        """ExecuteServiceEndpointRequest.
        Proxy for a GET request defined by a service endpoint.
        :param :class:`<ServiceEndpointRequest> <azure.devops.v7_0.service_endpoint.models.ServiceEndpointRequest>` service_endpoint_request: Service endpoint request.
        :param str project: Project ID or project name
        :param str endpoint_id: Id of the service endpoint.
        :rtype: :class:`<ServiceEndpointRequestResult> <azure.devops.v7_0.service_endpoint.models.ServiceEndpointRequestResult>`
        """
        route_values = {}
        if project is not None:
            route_values['project'] = self._serialize.url('project', project, 'str')
        query_parameters = {}
        if endpoint_id is not None:
            query_parameters['endpointId'] = self._serialize.query('endpoint_id', endpoint_id, 'str')
        content = self._serialize.body(service_endpoint_request, 'ServiceEndpointRequest')
        response = self._send(http_method='POST',
                              location_id='cc63bb57-2a5f-4a7a-b79c-c142d308657e',
                              version='7.0',
                              route_values=route_values,
                              query_parameters=query_parameters,
                              content=content)
        return self._deserialize('ServiceEndpointRequestResult', response)

    def create_service_endpoint(self, endpoint):
        """CreateServiceEndpoint.
        Creates a new service endpoint
        :param :class:`<ServiceEndpoint> <azure.devops.v7_0.service_endpoint.models.ServiceEndpoint>` endpoint: Service endpoint to create
        :rtype: :class:`<ServiceEndpoint> <azure.devops.v7_0.service_endpoint.models.ServiceEndpoint>`
        """
        content = self._serialize.body(endpoint, 'ServiceEndpoint')
        response = self._send(http_method='POST',
                              location_id='14e48fdc-2c8b-41ce-a0c3-e26f6cc55bd0',
                              version='7.0',
                              content=content)
        return self._deserialize('ServiceEndpoint', response)

    def delete_service_endpoint(self, endpoint_id, project_ids, deep=None):
        """DeleteServiceEndpoint.
        Delete a service endpoint
        :param str endpoint_id: Endpoint Id of endpoint to delete
        :param [str] project_ids: project Ids from which endpoint needs to be deleted
        :param bool deep: delete the spn created by endpoint
        """
        route_values = {}
        if endpoint_id is not None:
            route_values['endpointId'] = self._serialize.url('endpoint_id', endpoint_id, 'str')
        query_parameters = {}
        if project_ids is not None:
            project_ids = ",".join(project_ids)
            query_parameters['projectIds'] = self._serialize.query('project_ids', project_ids, 'str')
        if deep is not None:
            query_parameters['deep'] = self._serialize.query('deep', deep, 'bool')
        self._send(http_method='DELETE',
                   location_id='14e48fdc-2c8b-41ce-a0c3-e26f6cc55bd0',
                   version='7.0',
                   route_values=route_values,
                   query_parameters=query_parameters)

    def share_service_endpoint(self, endpoint_project_references, endpoint_id):
        """ShareServiceEndpoint.
        Share service endpoint across projects
        :param [ServiceEndpointProjectReference] endpoint_project_references: Project reference details of the target project
        :param str endpoint_id: Endpoint Id of the endpoint to share
        """
        route_values = {}
        if endpoint_id is not None:
            route_values['endpointId'] = self._serialize.url('endpoint_id', endpoint_id, 'str')
        content = self._serialize.body(endpoint_project_references, '[ServiceEndpointProjectReference]')
        self._send(http_method='PATCH',
                   location_id='14e48fdc-2c8b-41ce-a0c3-e26f6cc55bd0',
                   version='7.0',
                   route_values=route_values,
                   content=content)

    def update_service_endpoint(self, endpoint, endpoint_id, operation=None):
        """UpdateServiceEndpoint.
        Update the service endpoint
        :param :class:`<ServiceEndpoint> <azure.devops.v7_0.service_endpoint.models.ServiceEndpoint>` endpoint: Updated data for the endpoint
        :param str endpoint_id: Endpoint Id of the endpoint to update
        :param str operation: operation type
        :rtype: :class:`<ServiceEndpoint> <azure.devops.v7_0.service_endpoint.models.ServiceEndpoint>`
        """
        route_values = {}
        if endpoint_id is not None:
            route_values['endpointId'] = self._serialize.url('endpoint_id', endpoint_id, 'str')
        query_parameters = {}
        if operation is not None:
            query_parameters['operation'] = self._serialize.query('operation', operation, 'str')
        content = self._serialize.body(endpoint, 'ServiceEndpoint')
        response = self._send(http_method='PUT',
                              location_id='14e48fdc-2c8b-41ce-a0c3-e26f6cc55bd0',
                              version='7.0',
                              route_values=route_values,
                              query_parameters=query_parameters,
                              content=content)
        return self._deserialize('ServiceEndpoint', response)

    def update_service_endpoints(self, endpoints):
        """UpdateServiceEndpoints.
        Update the service endpoints.
        :param [ServiceEndpoint] endpoints: Names of the service endpoints to update.
        :rtype: [ServiceEndpoint]
        """
        content = self._serialize.body(endpoints, '[ServiceEndpoint]')
        response = self._send(http_method='PUT',
                              location_id='14e48fdc-2c8b-41ce-a0c3-e26f6cc55bd0',
                              version='7.0',
                              content=content)
        return self._deserialize('[ServiceEndpoint]', self._unwrap_collection(response))

    def get_service_endpoint_details(self, project, endpoint_id, action_filter=None):
        """GetServiceEndpointDetails.
        Get the service endpoint details.
        :param str project: Project ID or project name
        :param str endpoint_id: Id of the service endpoint.
        :param str action_filter: Action filter for the service connection. It specifies the action which can be performed on the service connection.
        :rtype: :class:`<ServiceEndpoint> <azure.devops.v7_0.service_endpoint.models.ServiceEndpoint>`
        """
        route_values = {}
        if project is not None:
            route_values['project'] = self._serialize.url('project', project, 'str')
        if endpoint_id is not None:
            route_values['endpointId'] = self._serialize.url('endpoint_id', endpoint_id, 'str')
        query_parameters = {}
        if action_filter is not None:
            query_parameters['actionFilter'] = self._serialize.query('action_filter', action_filter, 'str')
        response = self._send(http_method='GET',
                              location_id='e85f1c62-adfc-4b74-b618-11a150fb195e',
                              version='7.0',
                              route_values=route_values,
                              query_parameters=query_parameters)
        return self._deserialize('ServiceEndpoint', response)

    def get_service_endpoints(self, project, type=None, auth_schemes=None, endpoint_ids=None, owner=None, include_failed=None, include_details=None):
        """GetServiceEndpoints.
        Get the service endpoints.
        :param str project: Project ID or project name
        :param str type: Type of the service endpoints.
        :param [str] auth_schemes: Authorization schemes used for service endpoints.
        :param [str] endpoint_ids: Ids of the service endpoints.
        :param str owner: Owner for service endpoints.
        :param bool include_failed: Failed flag for service endpoints.
        :param bool include_details: Flag to include more details for service endpoints. This is for internal use only and the flag will be treated as false for all other requests
        :rtype: [ServiceEndpoint]
        """
        route_values = {}
        if project is not None:
            route_values['project'] = self._serialize.url('project', project, 'str')
        query_parameters = {}
        if type is not None:
            query_parameters['type'] = self._serialize.query('type', type, 'str')
        if auth_schemes is not None:
            auth_schemes = ",".join(auth_schemes)
            query_parameters['authSchemes'] = self._serialize.query('auth_schemes', auth_schemes, 'str')
        if endpoint_ids is not None:
            endpoint_ids = ",".join(endpoint_ids)
            query_parameters['endpointIds'] = self._serialize.query('endpoint_ids', endpoint_ids, 'str')
        if owner is not None:
            query_parameters['owner'] = self._serialize.query('owner', owner, 'str')
        if include_failed is not None:
            query_parameters['includeFailed'] = self._serialize.query('include_failed', include_failed, 'bool')
        if include_details is not None:
            query_parameters['includeDetails'] = self._serialize.query('include_details', include_details, 'bool')
        response = self._send(http_method='GET',
                              location_id='e85f1c62-adfc-4b74-b618-11a150fb195e',
                              version='7.0',
                              route_values=route_values,
                              query_parameters=query_parameters)
        return self._deserialize('[ServiceEndpoint]', self._unwrap_collection(response))

    def get_service_endpoints_by_names(self, project, endpoint_names, type=None, auth_schemes=None, owner=None, include_failed=None, include_details=None):
        """GetServiceEndpointsByNames.
        Get the service endpoints by name.
        :param str project: Project ID or project name
        :param [str] endpoint_names: Names of the service endpoints.
        :param str type: Type of the service endpoints.
        :param [str] auth_schemes: Authorization schemes used for service endpoints.
        :param str owner: Owner for service endpoints.
        :param bool include_failed: Failed flag for service endpoints.
        :param bool include_details: Flag to include more details for service endpoints. This is for internal use only and the flag will be treated as false for all other requests
        :rtype: [ServiceEndpoint]
        """
        route_values = {}
        if project is not None:
            route_values['project'] = self._serialize.url('project', project, 'str')
        query_parameters = {}
        if endpoint_names is not None:
            endpoint_names = ",".join(endpoint_names)
            query_parameters['endpointNames'] = self._serialize.query('endpoint_names', endpoint_names, 'str')
        if type is not None:
            query_parameters['type'] = self._serialize.query('type', type, 'str')
        if auth_schemes is not None:
            auth_schemes = ",".join(auth_schemes)
            query_parameters['authSchemes'] = self._serialize.query('auth_schemes', auth_schemes, 'str')
        if owner is not None:
            query_parameters['owner'] = self._serialize.query('owner', owner, 'str')
        if include_failed is not None:
            query_parameters['includeFailed'] = self._serialize.query('include_failed', include_failed, 'bool')
        if include_details is not None:
            query_parameters['includeDetails'] = self._serialize.query('include_details', include_details, 'bool')
        response = self._send(http_method='GET',
                              location_id='e85f1c62-adfc-4b74-b618-11a150fb195e',
                              version='7.0',
                              route_values=route_values,
                              query_parameters=query_parameters)
        return self._deserialize('[ServiceEndpoint]', self._unwrap_collection(response))

    def get_service_endpoints_with_refreshed_authentication(self, refresh_authentication_parameters, project, endpoint_ids):
        """GetServiceEndpointsWithRefreshedAuthentication.
        Gets the service endpoints and patch new authorization parameters
        :param [RefreshAuthenticationParameters] refresh_authentication_parameters: Scope, Validity of Token requested.
        :param str project: Project ID or project name
        :param [str] endpoint_ids: Ids of the service endpoints.
        :rtype: [ServiceEndpoint]
        """
        route_values = {}
        if project is not None:
            route_values['project'] = self._serialize.url('project', project, 'str')
        query_parameters = {}
        if endpoint_ids is not None:
            endpoint_ids = ",".join(endpoint_ids)
            query_parameters['endpointIds'] = self._serialize.query('endpoint_ids', endpoint_ids, 'str')
        content = self._serialize.body(refresh_authentication_parameters, '[RefreshAuthenticationParameters]')
        response = self._send(http_method='POST',
                              location_id='e85f1c62-adfc-4b74-b618-11a150fb195e',
                              version='7.0',
                              route_values=route_values,
                              query_parameters=query_parameters,
                              content=content)
        return self._deserialize('[ServiceEndpoint]', self._unwrap_collection(response))

    def get_service_endpoint_execution_records(self, project, endpoint_id, top=None, continuation_token=None):
        """GetServiceEndpointExecutionRecords.
        Get service endpoint execution records.
        :param str project: Project ID or project name
        :param str endpoint_id: Id of the service endpoint.
        :param int top: Number of service endpoint execution records to get.
        :param long continuation_token: A continuation token, returned by a previous call to this method, that can be used to return the next set of records
        :rtype: :class:`<[ServiceEndpointExecutionRecord]> <azure.devops.v7_0.service_endpoint.models.[ServiceEndpointExecutionRecord]>`
        """
        route_values = {}
        if project is not None:
            route_values['project'] = self._serialize.url('project', project, 'str')
        if endpoint_id is not None:
            route_values['endpointId'] = self._serialize.url('endpoint_id', endpoint_id, 'str')
        query_parameters = {}
        if top is not None:
            query_parameters['top'] = self._serialize.query('top', top, 'int')
        if continuation_token is not None:
            query_parameters['continuationToken'] = self._serialize.query('continuation_token', continuation_token, 'long')
        response = self._send(http_method='GET',
                              location_id='10a16738-9299-4cd1-9a81-fd23ad6200d0',
                              version='7.0',
                              route_values=route_values,
                              query_parameters=query_parameters)
        return self._deserialize('[ServiceEndpointExecutionRecord]', self._unwrap_collection(response))

    def get_service_endpoint_types(self, type=None, scheme=None):
        """GetServiceEndpointTypes.
        Get service endpoint types.
        :param str type: Type of service endpoint.
        :param str scheme: Scheme of service endpoint.
        :rtype: [ServiceEndpointType]
        """
        query_parameters = {}
        if type is not None:
            query_parameters['type'] = self._serialize.query('type', type, 'str')
        if scheme is not None:
            query_parameters['scheme'] = self._serialize.query('scheme', scheme, 'str')
        response = self._send(http_method='GET',
                              location_id='5a7938a4-655e-486c-b562-b78c54a7e87b',
                              version='7.0',
                              query_parameters=query_parameters)
        return self._deserialize('[ServiceEndpointType]', self._unwrap_collection(response))