File: github_service_endpoint_manager.py

package info (click to toggle)
azure-functions-devops-build 0.0.22-6
  • links: PTS, VCS
  • area: main
  • in suites: bookworm, forky, sid
  • size: 556 kB
  • sloc: python: 2,422; sh: 12; makefile: 8
file content (45 lines) | stat: -rw-r--r-- 1,839 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
import vsts.service_endpoint.v4_1.models as models
from vsts.exceptions import VstsServiceError
from ..base.base_manager import BaseManager
from .service_endpoint_utils import sanitize_github_repository_fullname

class GithubServiceEndpointManager(BaseManager):
    def __init__(self, organization_name, project_name, creds):
        super(GithubServiceEndpointManager, self).__init__(
            creds, organization_name=organization_name, project_name=project_name
        )

    def get_github_service_endpoints(self, repository_fullname):
        service_endpoint_name = self._get_service_github_endpoint_name(repository_fullname)
        try:
            result = self._service_endpoint_client.get_service_endpoints_by_names(
                self._project_name,
                [service_endpoint_name],
                type="github"
            )
        except VstsServiceError:
            return []
        return result

    def create_github_service_endpoint(self, repository_fullname, github_pat):
        data = {}
        auth = models.endpoint_authorization.EndpointAuthorization(
            parameters={
                "accessToken": github_pat
            },
            scheme="PersonalAccessToken"
        )
        service_endpoint_name = self._get_service_github_endpoint_name(repository_fullname)
        service_endpoint = models.service_endpoint.ServiceEndpoint(
            administrators_group=None,
            authorization=auth,
            data=data,
            name=service_endpoint_name,
            type="github",
            url="http://github.com"
        )

        return self._service_endpoint_client.create_service_endpoint(service_endpoint, self._project_name)

    def _get_service_github_endpoint_name(self, repository_name):
        return sanitize_github_repository_fullname(repository_name)