파이썬 셀레니움 함수 비동기(async)로 실행 방법

Python의 Selenium을 이용하여 크롤링을 진행할 때,

드라이버가 포함된 함수를 async로 실행해야 하는 경우가 있을 수 있습니다.

 

이 경우, 셀레니움의 함수를 비동기 함수처럼 사용이 가능하도록

코드를 작성할 수 있는 방법이 있는데요.

이번 글에서 해당 방법을 간략하게 정리해보도록 하겠습니다.

 

 

Sync 함수 예시

먼저, 이해를 돕기 위하여 다음과 같이 간단한 driver를 호출하는 함수 get_driver과

원하는 url에서 title 정보를 추출하는 get_title 함수가 있다고 가정해보도록 하겠습니다.

from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

# 드라이버를 불러오는 함수 가정
def get_driver():
    options = webdriver.ChromeOptions()
    options.add_argument('--headless')
    service = Service(executable_path="./chromedriver.exe")
    driver = webdriver.Chrome(service=service, options=options)
    return driver

# 드라이버를 사용하는 실제 호출 함수 가정(이 부분을 비동기로 변환 예정)
def get_title(url):
    driver = get_driver()
    driver.get(url)
    try:
        WebDriverWait(driver, 10).until(
            EC.presence_of_element_located((By.TAG_NAME, 'h1'))
        )
        title = driver.title
    finally:
        driver.quit()
    return title

위의 예시처럼 실제로 호출에 필요한 함수를 하나로 묶어주시면 이해가 쉽습니다.

이제 get_title 함수를 async 함수로 만들어보도록 하겠습니다.

 

 

Async 함수 변환 및 호출 예시

위처럼 드라이버를 사용하는 get_title 함수가 있다고 가정하면, 

이 부분을 다음과 같은 형태로 감싸서 비동기 함수로 변환이 가능합니다.

import asyncio
from concurrent.futures import ThreadPoolExecutor

# get_title의 async 버전
async def async_get_title(url):
    loop = asyncio.get_running_loop()
    with ThreadPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, get_title, url)
    return result

만약, get_title에서 필요한 변수의 종류가 여러개라면

loop.run_in_executor(pool, get_title, url1, url2)

와 같은 형태로 url1, url2 등의 변수 목록을 이어서 나열해주시면 됩니다.

 

 

이제, 이렇게 비동기로 변환한 함수를 실제로 호출하는 방법도 같이 살펴보겠습니다.

다음과 같은 main 부분으로 변환된 비동기 함수의 실행이 가능합니다.

async def main():
    urls = [
        "사이트 주소 1",
        "사이트 주소 2",
        "사이트 주소 3"
    ]
    
    # async 함수 호출 부분 예시
    tasks = [async_get_title(url) for url in urls]
    results = await asyncio.gather(*tasks)
    
    # 결과 출력 예시
    for url, result in zip(urls, results):
        print(f"Title of {url} is {result}")

if __name__ == "__main__":
    asyncio.run(main())