1. 개요

Google Cloud API를 Python에서 사용하기 위해 인증 과정을 거쳐 클라이언트를 생성해야 한다. 클라이언트 생성을 통한 API 호출 과정에 대해 정리하였다.

2. API 호출 과정

먼저 Google Cloud에 인증하는 과정이 필요하다. Python 뿐 아니라 다른 언어를 통해 API를 사용할 때에도 동일한 과정이 필요하다.

2.1. Credential

클라이언트를 생성할 때 넘길 수 있는 credentials는 oauth2client.Credentials 또는 google.auth.credentials.Credentials 객체를 생성하여 파라미터로 넘길 수 있다. oauth2client의 경우 현재 deprecated이므로(링크) google.auth모듈을 사용한다.

google.auth 모듈의 경우 다음과 같은 인증 방법을 사용할 수 있다.

  • Google Application Default Credentials.
  • JWTs
  • Google ID Tokens.
  • ID Tokens.
  • Google Service Account credentials.
  • Google Impersonated Credentials.
  • Google Compute Engine credentials.
  • Google App Engine standard credentials.
  • Requests, urllib3, gRPC

이 문서에서는 GCP Service Account를 활용하여 Key 데이터를 통해 인증을 진행한다.

2.2. Service Account 생성

사용할 Service Account를 생성한다.

Console > IAM & Admin > Service Accounts > CREATE SERVICE ACCOUNT

Service Account를 활용해 호출할 API에 필요한 권한이 있는 Role을 추가한다.

Service Account 생성이 완료되면, Service Accounts 목록에서 생성한 Service Account를 선택 후 Create Key를 눌러 키 파일을 생성한다. 생성한 키파일은 다음과 같은 Json 형식으로 되어있다.

{
    "type": "service_account",
    "project_id": "PROJECT_ID",
    "private_key_id": "PRIVATE_KEY_ID",
    "private_key": "PRIVATE_KEY",
    "client_email": "SERVICE_ACCOUNT_ID@PROJECT_ID.iam.gserviceaccount.com",
    "client_id": "CLIENT_ID",
    "auth_uri": "https://accounts.google.com/o/oauth2/auth",
    "token_uri": "https://oauth2.googleapis.com/token",
    "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
    "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/SERVICE_ACCOUNT_ID%40cnl.iam.gserviceaccount.com"
}

2.3. service_account 객체 생성

위에서 생성한 Service Account 키 데이터를 활용하여 service_account 객체를 생성한다.

먼저 클라이언트를 생성하기 위해 googleapiclient.discovery 모듈과 google.oauth2 모듈이 필요하다.

import googleapiclient.discovery as discovery
from google.oauth2 import service_account

키 데이터로부터 Service Account 객체를 생성한다.

def get_cred(cred_gcp):
    cred = service_account.Credentials.from_service_account_info(JSON_FORMAT_KEY_DATA)
    return cred

Json 파일로부터 바로 객체를 생성하기 위해서 다음과 같은 방법으로도 생성이 가능하다.

def get_cred():
    cred = service_account.IDTokenCredentials.from_service_account_file('service-account.json')
    return cred

2.4. 클라이언트 객체 생성

googleapiclient.discovery에서 사용 가능한 파라미터는 다음과 같다. 코드는 검색하여 확인이 가능하다.

def build(
    serviceName,
    version,
    http=None,
    discoveryServiceUrl=DISCOVERY_URI,
    developerKey=None,
    model=None,
    requestBuilder=HttpRequest,
    credentials=None,
    cache_discovery=True,
    cache=None,
    client_options=None,
    adc_cert_path=None,
    adc_key_path=None,
    num_retries=1,
)

사용 예시는 다음과 같다. serviceName으로 해당 서비스에 대한 이름 값을 넘긴다. 서비스 별 이름 값은 검색하여 확인할 수 있다.

이름 다음 사용되는 파라미터인 version은 각 API에 대한 버전을 의미한다. 각 서비스 별 사용 가능한 api 버전 역시 검색하여 확인할 수 있다.

credentials로 위에서 생성한 service_account를 넘겨준다. 문서 상단의 Credentials에 나열된 인증 방식 중 선택하여 적용이 가능하다.

discovery로 클라이언트를 생성 시 캐싱되지 않도록 cache_discovery=False 옵션을 추가한다. 해당 옵션을 적용하지 않을 시 ImportError: file_cache is unavailable when using oauth2client >= 4.0.0 등의 오류가 발생할 수 있다.

def get_gce_client(cred_gcp):
    cred = get_cred(cred_gcp)
    compute = discovery.build('compute', 'v1', credentials=cred, cache_discovery=False)
    return compute

def get_gae_client(cred_gcp):
    cred = get_cred(cred_gcp)
    compute = discovery.build('appengine', 'v1', credentials=cred, cache_discovery=False)
    return compute

def get_function_client(cred_gcp):
    cred = get_cred(cred_gcp)
    compute = discovery.build('cloudfunctions', 'v1', credentials=cred, cache_discovery=False)
    return compute

클라이언트까지 생성이 완료되면, 각 클라이언트에서 제공되는 API를 사용이 가능하다. 서비스, API마다 호출 방식이 다르므로 링크에서 사용하고자 하는 API를 찾아 형식에 맞게 호출한다.

  • 프로젝트의 Compute Engine 인스턴스 리스트 예시

프로젝트 내부의 Compute Engine 리스트를 구하는 Python 함수를 다음과 같이 작성할 수 있다.

ZONES = ["asia-northeast3-a", "asia-northeast3-b", "asia-northeast3-c"]

def list_project_gces():
    result = []
    for cred in client.list_creds():
        compute = client.get_gce_client(cred)
        for zone in ZONES:
            res = compute.instances().list(project=cred.get('project_id'), zone=zone).execute()
            try:
                for instance in res['items']:
                    instance['project'] = cred.get('project_id')
                    result.append(instance)
            except KeyError as e:
                print(e, ' No Instance in current zone: {}'.format(zone))
            except Exception as e:
                print(e)
            finally:
                pass
    return result
  • 프로젝트의 App Engine 앱 리스트 예시

프로젝트 내부의 App Engine 앱 리스트를 구하는 Python 함수를 다음과 같이 작성할 수 있다.

Compute Engine 인스턴스 리스트를 호출하는것과 기본적으로 동일한 구조이지만, appsId 파라미터를 포함해야 한다.

appsId값은 프로젝트마다 생성되며 Service Account에서 가져올 수 없으므로 생성해준다. (PROJECT_ID/)

def list_project_gaes():
    result = []
    for cred in client.list_creds():
        apps = client.get_gae_client(cred)
        appsId = cred.get('project_id')+"/"
        res = apps.apps().get(appsId=appsId).execute()
        result.append(res)
    return result
  • 프로젝트의 Cloud Functions 리스트 예시

프로젝트 내부의 Cloud Functions 리스트를 구하는 Python 함수를 다음과 같이 작성할 수 있다.

클라이언트.projects().locations().functions().list().execute()의 'functions' 값을 가져오면 function 리스트를 확인할 수 있다.

list 호출 시 parent라는 인자를 포함해야 하며, parent는 projects/PROJECT_ID/locations/LOCATION 의 형태로 구성되어 있다. 모든 location의 function을 가져올 경우, LOCATION 위치에 "-"를 사용한다.

def refresh_functions():
    result=[]
    for cred in client.list_creds():
        functions = client.get_function_client(cred)
        project = cred.get('keyData').get('project_id')
        parent = "projects/"+project+"/locations/-"
        try:
            for function in functions.projects().locations().functions().list(parent=parent).execute()['functions']:
                function['project'] = cred.get('project_id')
                info = function.get('name').split('/')
                function['location'] = info[3]
                function['functionName'] = info[5]
                result.append(function)
        except KeyError as e:
            print(e, ' No Function in current project: {}'.format(project))
        except Exception as e:
            print(e)
        finally:
            pass
    return result