[LinkedIn Ads] Advertising API로 캠페인 그룹 호출하기 (version 수정)

 

LinkedIn Ads의 Advertising API를 호출하던 중

header 필수 값인 version 때문에 오류가 자주 발생하였다.

 

맨 처음 코드는 version을 호출하는 시점에서 한 달을 빼주고 지난 달로 호출을 해주었다.

하지만 LinkedIn Ads의 API 버전이 규칙적으로 반영되지는 않아서

어느 달에는 두 달전 혹은 세 달전 버전으로 호출하거나 해야하는 일이 빈번했다.

 

· 기존 코드

# version format :  '202502'

version = (datetime.now() - relativedelta(months=1)).strftime('%Y%m')

headers = {
    "Authorization": access_token, 
    "LinkedIn-Version": version,
    "X-Restli-Protocol-Version": '2.0.0',
    "Accept-Encoding" : 'gzip, deflate, br'
}

campaigngroup_params = {
    'q': 'search', 
    'search': '(status:(values:List(ACTIVE)))'
}

campaigngroup_params = urllib.parse.urlencode(campaigngroup_params, safe='#\':()+=%,')

r = requests.get(url = campaigngroup_url, headers = headers, params= campaigngroup_params )

if r.status_code == 200:
    content = r.text
    content_json = json.loads(content)

    cg_df = pd.DataFrame(content_json['elements'])

else:
    raise requests.HTTPError(f"something went wrong:  {r.status_code}, {r.text}")

 

위의 방식으로 호출하면 아래처럼 존재하지 않는 버전이라는 오류 메세지와 함께

426 status code가 떨어진다.

HTTPError: something went wrong:  426, {"status":426,"code":"NONEXISTENT_VERSION","message":"Requested version 20250401 is not active"}

 

그래서!!! while을 사용해서 status code가 200으로 정상호출이 될 때까지 수정해주는 로직을 추가하였다.

우선 426 코드를 예외처리 해주었다.

 

· 1차 수정 코드

mm = 0
max_attempts = 12  # 최대 시도 횟수 

while mm < max_attempts:
    version = (datetime.now() - relativedelta(months=mm)).strftime('%Y%m')

    headers = {
        "Authorization": access_token, 
        "LinkedIn-Version": version,
        "X-Restli-Protocol-Version": '2.0.0',
        "Accept-Encoding" : 'gzip, deflate, br'
    }

    # active되어 있는 campaigngroup 불러오기
    campaigngroup_params = {
        'q': 'search', 
        'search': '(status:(values:List(ACTIVE)))'
    }

    campaigngroup_params = urllib.parse.urlencode(campaigngroup_params, safe='#\':()+=%,')

    r = requests.get(url = campaigngroup_url, headers = headers, params = campaigngroup_params)

    if r.status_code == 200:
        content = r.text
        content_json = json.loads(content)

        cg_df = pd.DataFrame(content_json['elements'])

        return cg_df

    elif r.status_code == 426:
        print(f"something went wrong version: {version}, {r.status_code}, {r.text}")
        mm += 1 #버전 오류가 발생하면 month를 더 과거로 조회

    else: #그 외는 raise로 오류 발생
        raise requests.HTTPError(f"API Call Failed! Status Code: {r.status_code}, Response: {r.text}")

raise RuntimeError("Max retries!!") #최대 시도 횟수가 넘어가면 raise로 오류 발생

 

이렇게 수정하였더니 오류가 발생하지 않았다 !

 

하지만 오늘 .... (이 오류 때문에 짜증나서 바로 블로그 쓰는 중 ㅎ) 버전 오류가 또 발생하였다.

일단 LinkedIn Ads의 adversiting API의 202502 버전은 현재 존재하지 않는 상태이다.

그러면 당연히 202501로 버전을 낮춰서 API 호출을 해야하지만 안되었다.

왜냐면..... status code가 426이 아닌 아래처럼 404로 떨어졌다... ㅜ

왜인지는 진짜 모르겠다.... 

{"message":"No root resource defined for path '/partnerApiAdAccountsV20250201'","status":404}

 

그래서 예외처리 부분에 404 코드도 추가해주었다.. ㅜ

만약 버전 때문에 발생하는 404 코드가 아니라 정말 Not Found여도

12번 넘게 실패하면 오류를 발생시켜서 

디버깅 해보면 되니까 큰 문제는 없을 것 같다.

 

아무튼 그래서 오늘 추가된 코드이다.

ETL 작업을 Airflow에서 하고 있어서 Dag 중에서 캠페인 그룹 호출하는 task 코드를 첨부하였다.

 

 

· 최종 코드(Dag)

def campaigngroup_call(**kwargs):

    ti = kwargs['ti']
    # 이전 task에서 xcom에 저장한 access token 호출
    access_token = ti.xcom_pull(key='access_token', task_ids='token_headers')
    
    mm = 0
    max_attempts = 12  # 최대 시도 횟수 

    while mm < max_attempts:
        version = (datetime.now() - relativedelta(months=mm)).strftime('%Y%m')

        headers = {
            "Authorization": access_token, 
            "LinkedIn-Version": version,
            "X-Restli-Protocol-Version": '2.0.0',
            "Accept-Encoding" : 'gzip, deflate, br',
        }

        # active되어 있는 campaigngroup 불러오기
        campaigngroup_params = {
            'q': 'search', 
            'search': '(status:(values:List(ACTIVE)))'
        }

        campaigngroup_params = urllib.parse.urlencode(campaigngroup_params, safe='#\':()+=%,')

        r = requests.get(url = campaigngroup_url, headers = headers, params = campaigngroup_params)

        if r.status_code == 200:
            content = r.text
            content_json = json.loads(content)

            cg_df = pd.DataFrame(content_json['elements'])

        # 뒤의 task에서도 사용해야 되어서 Xcom으로 push
            ti.xcom_push(key='cg_df', value=cg_df.to_json())
            ti.xcom_push(key='headers', value=headers)

        # token이나 version의 경우 airflow 내 variable에 저장하여 사용
            Variable.set('access_token', access_token)
            Variable.set('version', version)

            return

        elif (r.status_code == 426) | (r.status_code == 404):
            print(f"something went wrong version: {version}, {r.status_code}, {r.text}")
            mm += 1
            
        else:
            raise requests.HTTPError(f"API Call Failed! Status Code: {r.status_code}, Response: {r.text}")

    raise RuntimeError("Max retries!!")
    

campaigngroup_call_task = PythonOperator(
    task_id='campaigngroup_call',
    python_callable=campaigngroup_call,
)

 

※ 함수의 첫 줄에서 사용되는 변수 ti는 Task Instance의 약자이다.
당연히 다른 변수를 사용해도 되긴 하지만

Airflow에서 XCom (Cross Communication)을 사용해 task 간 데이터를 주고받을 때 보통 ti 변수를 활용한다!!!