aioresponses에 대한 본격적인 내용은 서론 아래에 있습니다

 

목차

- @aioresponses() 데코레이터 사용

- Context Manager를 활용한 aioresponses()

- 동일 URL 반복 호출

- 특정 URL만 실제 api로의 요청을 허용하는 방법

 


 

 

 

파이썬 서버 테스트 과정에서 공부한 외부 api mock out 방법에 대해 기록하고자 한다

 

 

현 회사에서 최근 서버 개발을 위한 unit test를 작성하는 방법에 대해 새롭게 익히고 있다.

기존에는 postman을 통해 e2e test를 중심으로 개발하였는데, e2e test는 사용자 입장에서의 테스트이기 때문에 개발 디버깅용으로는 비효율적인 방식이었다. 

 

일반적으로 서버가 MVC 방식으로 모듈화 되어있다고 했을 때, 각 모듈이 어떻게 동작하는지를 먼저 테스트 해야한다.

이때 사용하는 것이 unit test(단위 테스트)다. 

파이썬 unit test 툴로는 가장 유명한 것이 Pytest와 Unitttest 두 가지가 있는데, 필자는 pytest를 중심으로 공부하고 있다.

 

 

최근 크롤링 등의 이유로 외부 api를 많이 호출하면서 단위 테스트 과정에서 문제가 되었다.

 

예를 들어 크롤링해야하는 웹페이지는 로그인을 해야지만 접근할 수 있다면 테스트를 할 때 authorization 헤더에 매번 올바른 토큰값을 넣어줘야 한다.

이를 위해 주기적으로 로그인을 한 후 토큰값을 가져와 헤더에 넣어주는 절차를 거쳐야하는데, 비용 측면에서 큰 낭비가 될 것이다.

 

이 때 사용할 수 있는 방법은 외부 페이지를 mocking 하는 것이다.

mock은 서로 의존성이 강한 객체가 있을 때 가짜 객체를 만들어 사용하는 것을 말한다

 

 

그렇다면 외부 api를 호출할 때 어떻게 mocking한 페이지를 가져올까? 이 때 사용하는 것이 aioresponses 라이브러리다

 

 

 

 


 

 

 

 

aioresponses는  aiohttp request를 테스트할 때 사용할 수 있는 helper다.

 

requests 모듈을 테스트할 때 사용하는 툴은 httpretty, responses, requests-mock 등 많이 제공되고 있지만 aiohttp와 같이 비동기 방식이 적용되면 테스트 난이도가 올라가게 된다. 이때 사용할 수 있는것이 aioresponses다

 

aioresponses는 aiohttp session을 통해 외부 api로 request를 보낼때, mocking한 가상의 response를 대신 받을 수 있도록 한다.

 

aioresponses를 사용하는 방법에 대해 알아보자

 

 

 


 

@aioresponses() 데코레이터 사용

 

다음 코드를 보자

import aiohttp
import asyncio
from aioresponses import aioresponses

TEST_URL = "https://test.co.kr"

@aioresponses()
async def test(mocked):
    # 가상의 응답 설정
    mocked.get(TEST_URL, status=200, payload=dict(test="Success"))
    
    # 외부 api 호출
    session = aiohttp.ClientSession()
    async with session.get(TEST_URL) as response:
        print(await response.json())
    await session.close()


if __name__ == '__main__':
    asyncio.run(test())

 

결과

{'test': 'Success'}

 

 

@aioresponses() 데코레이터를 단 함수는 첫번째 인자로 aioresponses 객체를 받는다.

이 객체를 사용하면 어떤 api 호출에 어떤 응답을 줄 것인지 지정할 수 있다.

 

mocked 변수에 받은 aioreponses 객체를 사용하는 코드를 살펴보자

 

mocked.get(TEST_URL, status=200, payload=dict(test="Success"))

 

aioresponses에는 get, post, put, delete 등 http method를 지원한다.

인자로는 호출하고자 하는 api url를 전달해야하며, status, payload, body 등 다양한 응답을 지정할 수 있다.

 

payload의 인자로는 dict, str, list 등을 사용할 수 있지만 body의 경우 str만 전달이 가능하다

만약 json을 응답으로 전달한다면 payload가 좋을 것이고, html을 전달한다면 body가 나은 선택일 수 있다.

 

이외에 headers, exception, status 등도 직접 설정이 가능하다. 

 

 

 


 

Context Manager를 활용한 aioresponses

 

위 코드에서 데코레이터를 활용한 방식을 소개했다면 이번에는 context manager를 사용한 방식을 소개하려 한다

 

여기서 context manager란 with 구문을 사용하여 특정 자원을 정확한 시간동안 사용하고 종료할 수 있도록 하는 것을 말한다.

즉, with 구문 아래에 들여쓰기가 되어있는 코드에만 할당한 자원을 사용할 수 있도록 하는 것을 말한다.

위 코드에서 이미 aiohttp session을 사용하는데 활용했기 때문에 여기서는 간단하게만 설명하였다

 

 

다음 코드를 살펴보자

 

TEST_URL = "https://test.co.kr"

async def test():
    # context manager를 사용한 aioresponses
    with aioresponses() as mocked:
        mocked.get(TEST_URL, status=200, payload=dict(test="Success"))

        session = aiohttp.ClientSession()
        async with session.get(TEST_URL) as response:
            print(await response.json())
        await session.close()

 

결과

{'test': 'Success'}

 

위 코드를 보면 aioresponses 메서드에서 해당 객체를 mocked 변수에 받아 사용하고 있다.

아래에 코드는 데코레이터를 사용한 코드와 동일하다.

 

이 방법은 외부 api를 사용하는 부분이 전체 코드의 일부분이고, 복잡한 코드로 인해 에러가 우려되거나 메모리 누수(resource leak) 우려가 있는 경우 사용할 수 있다.

데코레이터는 함수 전체가 끝날 때까지 aioresponses 객체를 계속 유지한다.

때문에 외부 api 호출이 주요 기능인 경우가 아니라면 위와 같이 context manager를 사용해서 일정 시간동안만 리소스를 사용하고 안전하게 닫는 것이 더 좋은 방법일 수 있다.

 

 

 


 

동일 URL 반복 호출

 

이렇게 설정한 가상의 응답은 기본적으로 1회만 사용이 가능하다. 

아래의 예시 코드를 보자

 

TEST_URL = "https://test.co.kr"

@aioresponses()
async def test(mocked):
    mocked.get(TEST_URL, status=404)	# 첫번째 응답
    mocked.get(TEST_URL, status=200)	# 두번째 응답

    session = aiohttp.ClientSession()
    async with session.get(TEST_URL) as response:
        print(response.status)
    async with session.get(TEST_URL) as response:
        print(response.status)
    await session.close()

 

결과

404
200

 

위 코드에서는 같은 url에 대해 mocking된 응답을 2개 만든 뒤 같은 url을 2번 호출하였다.

결과는 보다시피 첫번째 요청은 첫번째 응답을, 두번째 요청은 두번째 응답을 받았다.

 

만약 같은 url에 대해 다양한 응답을 받고 싶다면 위와 같은 방법을 사용하면 된다.

 

 

 

그렇다면 만약 같은 url을 여러 번 호출하고 계속 같은 응답을 받고 싶다면 어떻게 하면 될까?

이때 사용할 수 있는 파라미터가 repeat 이다. 

 

가령 url은 같지만 body를 바꿔가며 여러개의 응답을 전부 받아야하는 경우가 있을 수 있는데  repeat=False (default)로 되어있는 경우 Connection refused 에러가 발생하게 된다. 

내가 mocking 한 응답을 재사용한다는 의미로 repeat=True를 해주어야만 반복적인 요청이 가능하다.

 

예시 코드를 보자

 

TEST_URL = "https://test.co.kr"

@aioresponses()
async def test(mocked):
    mocked.get(TEST_URL, status=200, repeat=True)	# repeat 인자 설정

    session = aiohttp.ClientSession()
    async with session.get(TEST_URL) as response:
        print(response.status)
    async with session.get(TEST_URL) as response:
        print(response.status)
    await session.close()

 

결과

200
200

 

 

 


 

특정 URL만 실제 api로의 요청을 허용하는 방법

 

aioresponses는 외부 api를 호출하는 모든 요청을 통제하고 응답이 mocking된 URL 요청만 허용한다.

만약 mocking되지 않은 URL을 임의로 호출하면 aioresponses는 해당 요청을 차단하고 에러를 보낸다.

 

아래의 코드를 보자

 

REAL_URL = "https://real.url.com"

@aioresponses()
async def test(mocked):
    session = aiohttp.ClientSession()
    # mocking하지 않은 외부 api 호출
    async with session.get(REAL_URL) as response:	
        print(await response.json())
    await session.close()

 

결과

aiohttp.client_exceptions.ClientConnectionError: Connection refused: GET https://real.url.com

 

위와 같이 mocking되지 않은 api를 호출하면 해당 요청은 Connection refused 되었다는 에러메세지를 볼 수 있다.

 

 

경우에 따라, 일부 api는 mocking된 응답을 받고 일부는 외부에서 실제 데이터를 가져와야 할 수 있다.

그런 경우에 사용할 수 있는 것이 passthrough 파라미터다.

 

aioresponses 함수를 호출할 때, passthrought 파라미터에 특정 url 리스트를 인자로 전달하면 해당 url은 외부로 요청할 수 있도록 허용한다. 

 

아래의 코드를 보자

 

TEST_URL = "https://test.co.kr"
REAL_URL = "https://real.url.com"

# https://real.url.com은 예시를 위한 임의의 URL입니다.
@aioresponses(passthrough=[REAL_URL])
async def test(mocked):
    mocked.get(TEST_URL, status=200, payload=dict(test="Success"))

    session = aiohttp.ClientSession()
    # mocking된 요청
    async with session.get(TEST_URL) as response:
        print(await response.json())
        
    # 실제 외부 api로의 요청
    async with session.get(REAL_URL) as response:
        print(await response.json())
        
    await session.close()

 

결과

{'test': 'Success'}
{'real_url': 'Connected'}

 

위 코드는 mocking 된 TEST_URL과 외부로 실제 요청을 보낼 REAL_URL 두개에 각각 요청을 보내고 있다.

TEST_URL은 가상의 응답을 받아왔고, REAL_URL은 외부 서버에서 응답을 받아온 것을 확인할 수 있다.

 

 


 

 

본 포스트는 제가 공부하는 과정에서 작성하였습니다. 정정사항 댓글은 언제나 환영합니다!

반응형

locust를 실행할 때 user 클래스에게 특정 값을 초기에 전달해야하는 경우가 있다.

예를 들면, 특정 도메인으로 요청을 보내고 싶을 때가 있고 아닐 때가 있는데

이 분기를 실행할 때 user 클래스로 True/False를 전달해서 결정할 수 있도록 하는 것이다.

 

 

이번 포스팅은 locust 실행 시 커스텀 인수를 전달하는 2가지 방법에 대해 정리하고자 한다.

 

  1. command line 명령어를 커스터마이징 하기
  2. 환경 변수 설정하기

 


 

 

1. command line 명령어 커스터마이징 하기

 

 

locust에는 디폴트로 설정된 커맨드라인 명령어가 있다. 여기에 내가 커스터마이징 한 명령어를 추가하는 방법을 사용한다

 

이 방법은 locust의 events 데코레이션을 활용하게 된다.

 

 

from locust import events

@events.init_command_line_parser.add_listener	#데코레이터 추가
def set_command_line(parser):
    #명령어 추가
    parser.add_argument("--test", type=bool, default=False, help="let's test customizing")

 

 

코드를 한 줄씩 살펴보면 먼저 events 데코레이션에서 커맨드라인 parser의 add_listener를 사용한다.

 

그리고 parser를 파라미터로 전달받아 argument를 추가한다.

 

add_argument 내부 파라미터는 여러가지가 있는데, 차례로 명령어, 입력타입, 디폴트값, 그리고 설명을 위 코드에 넣었다.

이외의 유용한 파라미터들은 공식문서를 통해 파악하면 된다.

 

위 함수는 user클래스 밖에 선언하면 된다.

 

재밌는 점은, 위와 같이 커맨드라인 명령어를 추가하고 locust -h 를 누르면 디폴트 명령어들과 함께 내가 추가한 명령어까지 콘솔에 나온다는 것이다. help는 이때 출력되는 명령어 설명에 들어가는 부분이다.

 

 

 

위와 같이 명령어를 추가했다면 user 클래스에서 사용해보자.

 

 

class TestUser(HttpUser):
    if self.environment.parsed_options.test == True:	#명령어 입력값 가져오기
    	self.client.get("/test")

 

 

사용은 간단하게 environment에 포함된 parsed_options에서 내가 추가한 명령어의 값을 가져오면 된다.

필자는 --test 라고 작성했으니 사용도 parsed_options.test로 사용하면 되는 것이다.

 

 

 

이제 이 명령어를 콘솔에서 사용해보자

 

 

locust --test True

 

 

 

 

2.  환경 변수 설정하기

 

따로 명령어를 추가하고 사용하는 것이 번거롭다면 단순히 환경변수를 사용하는 방법도 있다.

 

환경변수의 장점은 어떤 코드에서든 입력값에 접근할 수 있다는 점이다.

필자는 TaskSet에서 입력값을 가져와야 했는데 추가한 명령어로 접근하는 경우 user class에서 값을 가져와 TaskSet으로 전달하는 번거로운 과정을 거쳐야 했다. TaskSet은 environment에 접근할 수 없기 때문이다.

 

그러나 환경변수를 사용하면 TaskSet이든 일반 함수든 자유롭게 입력값에 접근할 수 있다.

 

 

환경변수는 따로 추가하거나 설정해줄 필요 없이 콘솔에서 다음과 같이 실행하면 된다.

 

 

<linux/mac>
TEST=True locust ...

<Window>
SET TEST=True
locust ...

 

 

운영체제에 따라 달라지긴 하지만 위처럼 간단하게 환경변수를 추가할 수 있다.

 

이렇게 추가한 변수를 사용해보자

 

 

import os
from locust import TaskSet, task

class TestTaskSet(TaskSet):
	
    @task
    def test_task(self):
    	if os.environ['TEST'] == "True":		# 환경변수 값 가져오기
        	self.client.get("/test")

 

 

사용은 os에서 환경변수를 가져오는 방식과 동일하다.

단, 환경변수는 타입이 str이므로 필요하면 타입을 변경하는 방식으로 사용하면 된다.

반응형

Locust로 테스트 시나리오를 작성하다보면 복잡한 코드를 구조적으로 구성해야하는 순간이 온다.

TaskSet은 테스트 코드를 계층적으로 구조화 할 때 유용하게 사용할 수 있다.

 

 


 

1. TaskSet

 

 

TaskSet은 User 클래스와 유사하게 사용할 수 있다.

단, User 클래스가 아니기 때문에 User 클래스 대신 사용하면 안된다.

 

 

from locust import HttpUser, task, TaskSet

class TaskSetTest(TaskSet):
    @task
    def test1(self):
    	self.client.get("/hello")
       
    @task(2)
    def test2(self):
    	self.client.get("/world")

 

 

위 코드를 보면 부모클래스로 HttpUser 대신 TaskSet을 상속받고 있다.

그 외에 나머지는 HttpUser를 사용하는 것과 동일하게 작성되어 있는걸 볼 수 있다.

 

이렇게 작성한 taskSet 클래스는 직접적으로 locust에서 사용할 수는 없다.

이 taskSet은 아래 User 클래스를 상속받은 클래스에서 사용할 것이다.

 

 

class UserClassTest(HttpUser):
    host = "https://localhost:8000"
    tasks = [TaskSetTest]

 

 

locust에서는 HttpUser를 상속받은 UserClassTest를 사용할 것이다.

 

미리 작성해둔 taskSet은 이 클래스에서 tasks = [] 형식으로 주입해 사용한다.

리스트 내부에 taskSet 클래스를 넣으면 클래스 내에 선언한 여러개의 task들이 리스트 안으로 들어간다.

그리고 locust를 실행하면 이 task들을 weight에 맞게 무작위로 선택하여 수행한다.

 

tasks 리스트 안에는 여러개의 TaskSet을 넣을 수 있다.

그리고 task를 추가하는 개념이기 때문에 아래에 UserClassTest만의 task를 만들 수 있다.

 

 

 

 

2. SequentialTaskSet

 

 

그렇다면 SequentialTaskSet은 무엇일까?

 

 SequentialTaskSet은 TaskSet에서 특징이 하나 추가되었다.

SequentialTaskSet을 부모클래스로 상속받고 task를 여러개 만들었다면, 위에서부터 순서대로 수행한다.

 

 

class SequentialTest(SequentialTaskSet):
    @task
    def test1(self):
    	self.client.get("/num1")
    
    @task
    def test2(self):
    	self.client.get("/num2")

    @task
    def test3(self):
    	self.client.get("/num3")

 

 

위와 같은 클래스가 있다고 했을 때, 이 클래스를 User 클래스에 넣어 수행하면

/num1 -> /num2 -> /num3 -> /num1 -> /num2 ...

이렇게 순서대로 수행하게 될 것이다. 

반응형

1.  Locust란?

 

" Python 스크립트 언어로 구현하는 서버 부하테스트 툴 "

 

즉, 내가 만든 서버에 수백, 수천의 동시사용자가 들어왔을 때 어떤 일이 벌어지는지를 테스트할 수 있는 툴이다.

파이썬 언어로 테스트 시나리오를 간편하게 작성할 수 있고, 결과를 Locust 웹으로 확인할 수 있다.

 

 

 

이 튜토리얼은 '서버 테스트'이기 때문에 서버가 이미 만들어져 있고 정상적으로 동작한다는 것을 가정합니다.

 

url로 접속할 것이기 때문에 서버 언어는 python이 아니어도 관계 없다.

필자의 서버는 localhost로 실행하였다

 

 


 

2.  설치

 

우선 파이썬 프로젝트를 하나 준비하자.

서버가 fastapi처럼 python3 언어를 사용한다면 서버 안에 설치해도 관계없다.

 

해당 파이썬 프로젝트 루트 폴더에서 다음 명령을 입력한다.

 

pip install locust

 

잘 설치가 되었는지 확인하자.

 

locust -V

locust 2.8.6

 

locust 버전 확인을 했을 때 위와 같이 잘 뜨면 설치가 완료된 것이다.

 

 

 

 


 

3.  locustfile.py 작성하기

 

 

locustfile.py는 locust를 실행했을 때 별다른 옵션을 주지 않으면 자동으로 실행시켜주는 파일이다.

우선, 다음과 같이 작성해보자

 

from locust import HttpUser, task

class ServerTest(HttpUser):
	@task
    def api_test(self):
    	self.client.get("/hello")
        self.client.get("/world")

 

이제 코드를 한 줄씩 살펴보자.

 

locust에서 알아야 할 가장 중요한 개념은 2개가 있다.

User class Task다.

 

 

class ServerTest(HttpUser):

 

User class는 한 명의 유저를 나타낸다.

Locust는 각각의 유저를 생성하기 위해 해당 클래스의 인스턴스를 생성한다.

즉, 유저가 만약 3명이라면 작성된 User class의 인스턴스를 3개 생성하게 되는 것이다.

이 클래스 안에는  request를 보내고, response를 받아 print를 하는 등의 테스트 스크립트가 작성된다.

 

HttpUser는  Http 프로토콜을 사용하는 User class로 가장 보편적으로 사용되는 클래스이다.

HttpUser를 내가 만든 클래스에서 상속받으면 클래스가 완성된다.

 

 

@task
def api_test(self):

 

Task는 유저가 실행하는 함수이다.

User class는 green thread(실제 스레드를 모방한 가짜 스레드)를 통해 실행된다.

task가 여러개 있을 때, 유저는 무작위로 task를 선택하여 실행하고 다음 task를 다시 선택해 실행하는 방식으로 테스트가 진행된다.

 

@task 는 task를 추가하는 가장 간단한 방법이다. 함수 위에 선언해주면 된다.

 

 

self.client.get("/hello")
self.client.get("/world")

 

client는 파이썬 라이브러리인 requests.Sessions의 서브클래스인 HttpSession의 인스턴스이다.

 HttpSession이기 때문에 Http의 모든 메서드를 사용할 수 있다.(get, post, delete...)

requests.Session과 동일하게 활용하면 된다.

url은 호스트 부분을 제외한 나머지 부분을 적어주면 된다. (http://localhost:8000/hello 라고 했을 때 /hello 만 작성한다.)

 

 


 

4.  실행하기

 

실행은 터미널에 locust라고만 작성해주면 된다.

(서버는 이미 실행중이어야 한다)

 

locust

 

locust를 실행하면 링크가 하나 뜨게된다.

"http://localhost:8089"로 접속하면 locust에서 제공하는 테스트 웹 인터페이스에 접속할 수 있다.

(localhost는 0.0.0.0과 같은 의미이다)

 

 

접속해보면 위와 같은 화면이 뜬다. 하나씩 살펴보자

 

Number of users는 총 동시사용자 수를 의미한다.

즉, 만약 내가 1000이라고 작성하면 테스트에는 총 1000명의 동시사용자가 서버에 접속하게 된다.

 

Spawn rate는 1초 당 추가로 접속할 사용자의 수이다.

만약 총 동시사용자가 1000명이고 초당 100명으로 설정했다면 1초에 100명, 2초에 200명... 이렇게 총 동시사용자수에 도달하기까지 10초가 걸릴 것이다.

 

Host는 내 서버의 host url이다. http://localhost:8000 과 같이 작성해주면 된다.

호스트를 잘못 작성하면 에러가 뜨니 제대로 작성했는지 확인하자.

 

 

 

접속한 후 화면을 살펴보자

 

 

아까 get으로  task에 작성한 url이 뜨는 것을 볼 수 있다.

Median, Average, Min 등은 각 요청에 대한 응답 중간값, 평균, 최솟값 등을 의미한다.

 

위에 툴바에 edit 버튼을 누르면 실행 중간에라도 총 동시사용자 수나 초당 사용자 수를 변경할 수 있다.

 

 

차트에서는 이렇게 테스트 현황을 실시간으로 확인할 수 있다.

이때, 사용자는 아까 설정한 초당 추가사용자수만큼 증가하고 있는 것이다. 

 

첫번째 차트는 RPS(current Requests Per Seconds)와 초당 Failures를 보여준다. 

두번째 차트는 Median Response Time과 95% percentile를 보여준다.

캡쳐하지는 않았지만 스크롤 아래에 세번째 차트도 있는데, 증가하는 유저수 추이를 보여준다.

 

탭에는 발생한 이슈를 보여주는 Failures, 가중치를 둔 task들간의 current rate 등도 있다.

 

 


 

 

본 포스트는 제가 공부하는 과정에서 작성하였습니다. 정정사항 댓글은 언제나 환영합니다!

반응형

이번 게시물에서는 response에서 원하는 데이터를 추출하는 방식을 소개한다.

 

Scrapy 튜토리얼1 - Scrapy를 선택한 이유

Scrapy 튜토리얼2 - 네이버 검색 결과 크롤링(크롤러 설정)


 

1. 데이터 추출

 

위 조건을 다 맞춰 spider를 만들었다면 parse함수에서 받아온 response로 원하는 값을 가져오자.

 

 

크롤링을 하기 위해서는 기본적으로 내가 원하는 데이터가 들어있는 태그를 파악해야한다.

F12키를 눌러 개발자 도구를 열고 왼쪽 위에 커서모양 버튼을 누르면 내가 원하는 데이터의 태그를 쉽게 찾을 수 있다.

 

 

scrapy는 class명과 id명, xpath 등을 활용할 수 있는데 간단히 class명으로 데이터를 받아온다.

 

response.css() 함수를 사용하면 class명과 id명을 이용해 데이터를 받아올 수 있다.

클래스는 .클래스명 으로 앞에 점을 하나 붙여야 하고 id는 #아이디명 으로 #을 붙여줘야 동작한다.

 

그 뒤에는 get()함수나 getall() 혹은 extract() 등을 사용해야 비로소 데이터를 추출해온다.

get() 함수는 해당하는 여러 태그 중 1개만 추출하고, getall()이나 extract()는 해당하는 태그가 여러개일 경우 리스트로 반환한다.

getall()과 extract()는 큰 차이는 없지만 getall()은 무조건 리스트로 반환하는 반면 extract() 결과 개수에 따라 리스트 혹은 str로 반환된다. 

 

selector 함수에 대한 더 자세한 사항은 공식문서를 참고하자.

 

 

 

위 코드를 실행한 결과는 다음과 같다.

<a href="https://blog.naver.com/kizaki56?Redirect=Log&amp;logNo=222665532164" class="api_txt_lines total_tit" target="_blank" onclick="return goOtherCR(this, 'a=blg*a.iblg&amp;r=1&amp;i=90000003_0000000000000033D7E66304&amp;u='+urlencode(this.href))"><mark>강남</mark> 한우 <mark>맛집</mark> 장위동유성집 특선 메뉴 추천</a>

 

 

 

2. 텍스트 추출

 

scrapy도 beautifulSoup의 get_text()함수처럼 텍스트만 추출해오는 함수가 존재한다.

이번에는 결과에서 텍스트만 추출해보자

 

    def parse(self, response):
        result = response.css('.api_txt_lines::text').get()
        print(result)

css 선택자에서 텍스트만 추출하도록 하려면 class명 뒤에 ::text 를 붙이면 된다.

 

 

추출된 결과를 보자

 

한우

 

 

분명 제목 전체를 가져오기를 원했는데 결과가 조금 이상하다. 그 이유는 html파일 구조를 보면 알 수 있다.

 

 

::text는 하위태그 내의 text까지는 접근할 수 없고, 텍스트 중 첫번째 텍스트만 가져오는 단점이 있다.

하위 태그까지 포괄한 모든 텍스트를 가져오고자 한다면 xpath를 사용해야한다.

 

 

    def parse(self, response):
        result = response.css('.api_txt_lines').xpath('string(.)').get()
        print(result)

 

xpath는 함수 xpath()를 사용하여 접근하면 된다.

일반적인 사용 예시는 'response.xpath('//*[@id="sp_blog_1"]/div/div/a').get()' 처럼 활용하면 된다.

 

여기서 주목해야할 것은 string()이다.

xpath 링크를 string()으로 감싸면 현재 태그가 감싸고 있는 모든 텍스트를 전부 반환하는 상당히 편리한 함수이다. 

 

한가지 더 설명하자면, string() 함수 안에 있는 .은 현재 위치를 의미한다. css로 접근한 뒤 xpath로 재접근을 하는 것이기 때문에 css로 접근한 현재위치인 a태그에서 모든 텍스트를 가져오는 것을 뜻한다. 만약 그 안쪽으로 더 접근하고자 한다면 .xpath('./div') 등으로 활용할 수 있다. 

 

 

결과를 살펴보자

 

강남 한우 맛집 장위동유성집 특선 메뉴 추천

 

원하는 결과를 잘 추출한 것을 볼 수 있다.

반응형

이번 게시글에서는 Scrapy를 파이썬 스크립트에서 어떻게 호출하고 사용하는지에 대해 소개하려 한다.

 

Scrapy는 프레임워크이기 때문에 쉘에서 사용하는 것이 일반적이지만, 필자는 서버에서 사용해야 하기 때문에 파이썬 스크립트에서 활용하였다.

 

튜토리얼은 네이버 블로그 검색 결과를 가져오는 방법을 소개한다.

 

 

Scrapy 튜토리얼1 - Scrapy를 선택한 이유

Scrapy 튜토리얼3 - 네이버 검색 결과 크롤링(데이터 추출)


 

1. 설치

 

우선 Scrapy를 사용하기 위해 파이썬 프로젝트에 설치해보자

pip install Scrapy

 

 

 

2. Spider 클래스 생성

 

설치가 완료되었다면 우선 Spider라는 것을 만들어야 한다. 용어가 조금 생소한데, 거미줄을 쳐서 html에서 원하는 정보만 걸러 가져온다.. 라는 발상에서 나온 용어이지 않나 싶다.

 

Scrapy는 이 Spider라는 클래스를 호출해 크롤링을 실행한다.

필자는 크롤링을 실제로 진행하는 함수들을 모아둔 클래스 정도로 이해했다.

 

import scrapy

class NaverSpider(scrapy.Spider):
    name = "naver"

    # 처음 크롤링을 시작하는 함수
    def start_requests(self):
    	# '강남 맛집'으로 검색한 네이버 블로그 결과 링크
        url = "https://search.naver.com/search.naver?query=강남 맛집&nso=&where=blog&sm=tab_opt"

        yield scrapy.Request(url=url, callback=self.parse)

    # 크롤링 결과를 받는 callback함수
    def parse(self, response):
    	# 해당 코드는 다음 게시물에서 설명합니다
        result = response.css('.api_txt_lines').get()
        print(result)

 

spider 클래스를 만드는 것에는 몇 가지 제약 조건이 있다.

 

 

 

1. 클래스는 scrapy.Spider를 상속받아야 한다.

 

클래스명은 어떤 것이든 관계 없다.

 

 

2. 클래스 멤버변수 name을 생성한다.

 

Scrapy가 spider를 사용하여 크롤링을 시작할 때, spider를 구분하기 위해 name 멤버변수를 사용한다. 

name은 말 그대로 spider를 식별할 수 있는 이름으로 어떤 이름을 사용하든 상관없다.

 

 

3. start_requests(self)를 생성한다.

 

Scrapy는 이 start_requests 함수를 가장 먼저 호출한다. 함수명은 변경하면 안된다.

함수 안에는 반드시 'yield scrapy.Request(url=링크, callback=콜백함수)'가 선언되어 있어야한다.

return을 쓰면 TypeError: 'Request' object is not iterable  라는 에러가 발생한다.

(yield은 return과 달리 여러번 결과를 반환할 수 있다.)

 

yield로 인해 scrapy는 해당 링크로 요청을 보내고 콜백함수를 호출해 응답을 사용할 수 있도록 한다.

for문을 통해 여러개의 링크로 request를 보낼수도 있다.

 

#예시코드

class NaverSpider(scrapy.Spider):
    name = "naver"

    def start_requests(self):
        urls = [링크1, 링크2, 링크3]

        for url in urls :
        	yield scrapy.Request(url=url, callback=self.parse)

 

 

4. scrapy.Request의 콜백함수를 만든다.

 

콜백함수의 함수명은 어떤것이든 관계 없다. 그러나 매개변수로 response는 반드시 들어가있어야 한다.

scrapy가 콜백함수로 response를 보내주기 때문이다.

 

 

 

 

3. CrawlerProcess 호출

 

spider를 만들었다면 이제 scrapy에서 spider를 호출해야한다.

 

파이썬 스크립트에서 scrapy를 사용하는 방법은 2가지이다.

 

  1.  CrawlerRunner()
  2.  CrawlerProcess()

 

두 함수의 차이는 호출 방식도 있지만 메인 스레드가 아닌 다른 스레드에서 호출이 가능하냐 아니냐의 차이이다.

 

CrawlerRunner는 메인 스레드가 아닌 서브스레드, 서브 프로세스에서도 동작이 가능하다.

CrawlerProcess는 오직 메인 스레드에서만 동작이 가능하다.

CrawlerRunner여러개의 spider를 동시에 실행할 수 있다는 장점도 있다.

 

두 함수의 호출 방식을 살펴보자.

 

from scrapy.crawler import CrawlerRunner

if __name__ == '__main__':

    configure_logging({'LOG_FORMAT': '%(levelname)s: %(message)s'})
    runner = CrawlerRunner()
    runner.crawl(NaverSpider)  # 이 부분에 자신이 정의한 spider 클래스를 넣는다
    # runner.crawl(DaumSpider)   # 여러개의 spider를 동시에 실행할 수도 있다.
    crawler = runner.join()
    crawler.addBoth(lambda _: reactor.stop())
    reactor.run()  # the script will block here until the crawling is finished

 

위 코드는 공식문서 튜토리얼 코드 그대로이다. 필자도 특별히 이 부분에 대해 건드린 것은 없다.

reactor.run()이 실행되면 크롤링이 모두 끝날때까지 다음 코드로 이동하지 않고 기다려준다.

 

단 주의해야할 사항은 reactor는 한번 실행되면 해당 프로세스가 끝날때까지 재실행이 불가능하다.

 

 

from scrapy.crawler import CrawlerProcess

if __name__ == '__main__' :
    process = CrawlerProcess()
    process.crawl(NaverSpider)  # 이 부분에 자신이 정의한 spider 클래스를 넣는다
    process.start()

 

CrawlerProcess는 호출 방식이 더 심플하다.

reactor와 달리 여러번 호출이 가능하고, process.start() 역시 크롤링이 다 끝날때까지 기다려준다. 

 

 

다음 포스트에서는 크롤링하여 받아온 response에서 데이터를 추출하는 작업을 진행하도록 하겠다

반응형

이번 프로젝트에서는 서버에서 블로그 본문 데이터를 크롤링해 온 다음 이를 분석하여 클라이언트에 반환하는 기능을 구현했다. 이 과정에서 Scrapy를 공부하고 사용한 내용을 기록하고자 한다.

 

Scrapy 튜토리얼2 - 네이버 검색결과 크롤링(크롤러 설정)

Scrapy 튜토리얼3 - 네이버 검색결과 크롤링(데이터 추출)


 

웹 크롤링은 대표적으로 BeautifulSoup와 Selenium을 사용한다. 배우기 쉽고, 한국어 자료도 방대한 장점이 있다. 

 

다만 내가 프로젝트에서 두 라이브러리 대신 Scrapy를 사용한 이유는 3가지가 있다.

 

 

 

Webdriver를 호출하는 Selenium의 속도저하 문제

 

Selenium은 실제 웹페이지를 띄우고, 렌더링되는 시간을 기다려야 크롤링이 가능하다. 지도나 무한 스크롤같은 동적인 페이지를 크롤링해야 한다면 Selenium을 사용하는 것이 맞다.

그러나 나는 정적인 페이지를 크롤링하고, BeautifulSoup이나 Scrapy와 비교했을때 훨씬 많은 시간을 소요해 선택하지 않았다.

 

 

 

동기적으로 동작하는 BeautifulSoup

 

BeautifulSoup와 Scrapy는 둘 다 request를 보냈을 때 response를 그대로 받아온다는 점에서 동일하다.

이 둘의 차이점은 BeautifulSoup은 동기적으로, Scrapy는 비동기적으로 동작한다는 것이다.

쉽게 말해 BeautifulSoup은 페이지 크롤링하는 동안 어떤 코드도 수행할 수 없어 대기해야 한다. 그러나 Scrapy는 페이지를 크롤링하는 동안 다른 동작을 수행할 수 있다. 

 

따라서 시간적으로 봤을때 BeautifulSoup는 Scrapy보다 더 많은 시간을 소요한다. BeautifulSoup와 multiprocessing을 함께 사용하면 상당히 빨라지지만 프로젝트 특성상 이후에도 많은 멀티스레딩을 동원하기 때문에 추가적으로 멀티프로세스를 호출하는 것이 오버헤드 측면에서 부담스러웠다.

 

 

 

Scrapy의 확장성 및 편의

 

BeautifulSoup가 단순히 html에서 정보를 가져오는데 그친다면 Scrapy는 확장성 측면에서 상당히 유연하다. middleware를 사용자가 직접 손볼 수 있고 데이터를 다운로드 하거나 하는 등의 여러 확장이 자유롭다.

 

더불어, 필자는 블로그에서 본문데이터를 가져오는 작업을 진행했는데 BeautifulSoup보다는 Scrapy가 하위 태그에서 텍스트만 골라 가져오는 것이 더 편리했다.(이것은 개인적인 의견일 수도 있다.)

 

 

 

한국에서는 Scrapy를 잘 사용하지 않는지 한국어 자료보다는 영어자료가 훨씬 많았다. 다행스러운 점은 스택오버플로우 등 영어권에는 Scrapy 자료가 훨씬 방대하다.

 

 

이러저러한 이유로 결론은 Scrapy를 선택했다는 것이다.

Scrapy가 BeautifulSoup와 Selenium에 비해서 배우기 어렵고 한국어 자료도 많이 없는 단점은 있다.

그러나 공식문서 튜토리얼을 따라가며 배우다보면 대략적으로 어떻게 동작하는지 이해하는 데에는 오래걸리지 않는다.

 

 

다음 포스트에서 본격적으로 Scrapy를 어떻게 사용하는지에 대해 소개하도록 하겠다.

반응형

프로젝트 중 파이썬의 subprocess.check_output의 리턴값인 string을 다시 dict(json.load)로 변환하는 과정에서 문제가 발생했다.

 

 

 

subprocess모듈의 check_output은 다른 프로세스에서 수행된 print 값을 모두 모아 하나의 string으로 반환해준다.

그런데 dict를 print 해서 string으로 나온 것을 다시 json.load 함수를 써서 dict로 바꾸려니 온갖 DecodeError가 떴다.

 

작은 따옴표를 큰 따옴표로 바꾸고 '\'을 '\\'로 바꾸라는 둥 replace를 해결책으로 권하는 사람들이 많았는데

여간 귀찮은게 아닌데다 나는 딕셔너리의 value 내에 큰따옴표로 바꾸면 안되는 작은 따옴표들도 들어있어서 난감했다.

 

 


 

 

여기서 상당히 간단한 해결책이 있다.

ast 모듈eval()함수를 활용하면 된다.

 

import ast
import subprocess

result = subprocess.check_output([쉘 명령어], encoding='utf-8')
#result : "{'key': 'value'}"

dict = ast.literal_eval(result)
#dict : {'key': 'value'}

 

literal_eval은 문자열 내의 파이썬 표현식을 인식하는 함수인데 

예를 들어 '10+10'문자열이 있다고 하면 eval을 거쳐서 20의 결과를 반환해주는 상당히 편리한 함수이다.

그래서 위 코드를 수행하면 문자열이 딕셔너리로 변환된다.

 

 

 

다만 파이썬에서는 eval함수 사용을 보안상 문제로 인해 사용을 자제하도록 권고하고 있다.

 

이 함수는 어떤 문자열이든 인증절차나 확인 없이 바로 코드로 인식해 수행해버리기 때문에

만약 외부에서 들어온 악의적인 해킹 코드가 문자열로 들어오고, 그 값이 eval 함수에 들어가면 코드로 변환해 수행해버린다. 서버를 부수거나 기밀정보를 빼돌리는 것이 얼마든지 가능해지는 것이다.

 

따라서 사용에 상당히 주의해야함은 맞다.

반응형

 

 

프로젝트 중 네이버 검색 결과 60개의 각 링크를 크롤링해야하는 기능이 있어 화면에서 스크롤을 내리며 크롤링을 진행했다.

 

네이버 View의 경우 한번에 30개씩 검색결과를 보여준다.

그리고 스크롤을 내리면 계속해서 30개씩 불러와 결과를 보여주는 방식이다.

 

셀레니움에는 자바스크립트를 수행할 수 있는 코드가 있는데 이것을 활용하고자 한다.

 


 

1. 검색 화면 띄우기

 

 

네이버 View 중 블로그 카테고리의 검색결과만을 가져오도록 url을 설정한다.

그리고 검색 결과가 나올 때까지 5초까지 대기한다.

 

해당 코드의 일부

 

from selenium import webdriver
from selenium.webdriver.support.ui import WebDriverWait

URL = 'https://search.naver.com/search.naver?query=' + query + '&nso=&where=blog&sm=tab_opt'
    
try :
	#검색 결과가 나올 때까지 대기
        WebDriverWait(driver, 5).until(
            EC.presence_of_element_located((By.CLASS_NAME, 'thumb_single'))
        )
        
except Exception as e :
	print(e)

 

Explicitly wait

말 그대로 명시적 대기이다. 조건을 작성하고 해당 조건을 충족할때까지 기다려주는 방식이다.

 

웹페이지가 완전히 불러와질때까지 5초까지 대기하는데 이때 explicitly wait 방식을 사용하였다.

 

 

 

WebDriverWait(driver, 5).until()

 

WebDriverWait의 파라미터로 첫번째는 webDriver, 두번째는 기다릴 시간(sec 단위)를 작성한다.

until 내부에 찾고자 하는 element를 탐색하는 함수를 전달한다. 만일 element를 찾는다면 5초까지 기다리지 않고 다음 코드로 넘어간다.

element를 찾지 못하고 5초가 넘어가면 TimeoutException이 발생한다.

 

EC.presence_of_element_located((By.CLASS_NAME, 'thumb_single'))

 

WebDriverWait의 조건으로 위 코드를 작성하였다.

어떤 element가 존재하는지 여부를 확인하는 함수를 until 함수에 전달하게 된다.

위 코드에 따르면, 찾고자하는 요소는 'thumb_single'이라는 클래스 이름을 가지고 있다.

 

 

 

 

 


 

2. 화면 스크롤

 

화면이 출력되었다면 크롤링을 할 준비가 되었다.

네이버는 한 번에 30개씩 출력하기 때문에 만약 60개의 결과를 크롤링하고 싶다면 스크롤을 내려 30개를 추가로 받아와야 한다.

 

스크롤을 화면 끝까지 내리는 코드

 

driver.execute_script("window.scrollTo(0, document.body.scrollHeight)")
WebDriverWait(driver, 2).until(
	lambda driver: len(driver.find_elements(By.CLASS_NAME, 'thumb_single')) == 원하는 개수
)

 

excute_script는 자바스크립트 코드를 수행하는 함수이다.

도큐먼트의 높이만큼 스크롤을 내리면 화면의 가장 하단에 도달한다.

하단에 도달한 후 검색결과를 가져오는 시간을 다시 explicitly wait 방식으로 기다려준다.

 

 

 

lambda driver: len(driver.find_elements(By.CLASS_NAME, 'thumb_single')) == 원하는 개수

 

until 함수에는 메소드를 파라미터로 전달해야 한다.

전달된 메소드는 until 함수 내에서 반복적으로 수행되며, 메소드 반환값이 존재하면 WebDriverWait을 중지한다.

따라서 element가 특정 개수만큼 존재하는지를 확인하기 위해서 위와 같은 람다함수를 작성한다.

find_elements 함수를 통해 element 전부를 받아온 후 이 개수가 원하는 개수인지를 확인하는 방식이다.

 

참고 : https://stackoverflow.com/questions/64746509/how-to-wait-for-number-of-elements-to-be-loaded-using-selenium-and-python

 

 

 

 

 


 

3. 원하는 요소 크롤링

 

스크롤을 내려 원하는 만큼 검색결과가 나왔다면 이제 원하는 요소를 크롤링한다.

프로젝트에서는 각 결과의 링크들을 크롤링하기 때문에 간단히 다음 코드로 크롤링을 진행했다.

 

result = driver.find_elements(By.CLASS_NAME, 'thumb_single')

for idx in range(len(result)):
	element = result[idx]
     	try:
		links.append(element.get_attribute("href"))
        
    	except Exception as e:
		print("exception During link crawling")
		print(e)

 

반응형

데이터분석 프로젝트 중 약 250만개의 데이터를 분류하는 파트에서의 문제.

 

Dataframe의 append 함수를 사용해서 같은 dataframe 변수에 리스트처럼 계속 분류된 값들을 추가하는 작업이었다.

그런데 대략 80만정도를 넘어가니 주피터 노트북에서 속도가 상당히 느려지면서

1시간동안 100만개 밖에 분류를 못하는 상황이 벌어졌다. 

속도 문제로 130만개쯤에서 중단.

 

 

 

문제의 코드 예시

#대량의 데이터를 담은 원본 데이터프레임
origin

#분류한 데이터를 나눠담을 데이터프레임들
d1 = pd.Dataframe(column=['c1', 'c2', 'c3'])
d2 = pd.Dataframe(column=['c1', 'c2', 'c3'])
d3 = pd.Dataframe(column=['c1', 'c2', 'c3'])

#반복문
for idx, row in origin.iterrows():
	if(row.c1 == 1): d1 = d1.append(row, ignore_index=True)
	elif(row.c1 == 2): d2 = d2.append(row, ignore_index=True)
	elif(row.c1 == 3): d3 = d3.append(row, ignore_index=True)

 

 

 

원인을 생각해보니 append 함수는 기존 프레임 뒤에 한 데이터를 추가해 새로운 프레임을 만드는 방식이었다.

즉, 데이터가 하나 들어갈 때마다 하나만 넣는게 아닌 처음부터 새로 넣는 방식이었던 것 같다.

(정정 사항이 있다면 댓글 부탁드립니다)

 

 

 

따라서 검색 결과 찾아낸 딕셔너리를 이용한 방식

#대량의 데이터를 담은 원본 데이터프레임
origin

#분류한 데이터를 나눠담을 딕셔너리들
d1 = {}
d2 = {}
d3 = {}
idx_d1 = 0
idx_d2 = 0
idx_d3 = 0


#각 딕셔너리에 데이터를 추가할 함수
def addRow(col_idx, group, data):
    if(group == 1):
        d1[col_idx] = data
    elif(group == 2):
        d2[col_idx] = data
    elif(group == 3):
        d3[col_idx] = data


#반복문
for idx, row in origin.iterrows():

	data = {"c1":row.c1, "c2":row.c2, "c3":row.c3}
    
	if(row.c1 == 1):
    	addRow(idx1, 1, data)
	elif(row.c1 == 2):
    	addRow(idx2, 2, data)
	elif(row.c1 == 3):
    	addRow(idx3, 3, data)

 

 

 

딕셔너리의 키를 인덱스로 활용하면 앞선 방식보다 몇 배는 빠른 성능을 낼 수 있었다.

이후 딕셔너리를 dataframe으로 만들거나, json화해서 파일시스템에 저장하는 등 변형이 자유롭다.

반응형

+ Recent posts