[Python] 크롤링 봇과 FastAPI로 간단한 백엔드 구축하기

JaehyoJJAng2024년 06월 07일

개요

requests와 bs4를 사용하여 특정 사이트로부터 데이터를 수집하고,

수집한 데이터를 API로 제공해주는 간단한 백엔드 시스템을 구축해보려고 한다.

해당 프로젝트의 요구 사항은 다음과 같다.

1. 주기적으로 특정 사이터 정보 크롤링
2. 크롤링 데이터를 데이터베이스에 저장
3. 필요할 때 API로 데이터를 제공

수집 데이터

수집되는 데이터는 다음과 같다.

[{'title': '에이서 스위프트 GO 16 OLED, 스틸 그레이, 코어i7, 512GB, 16GB, WIN11 Home, SFG16-71-77FT', 'link': '#product1_detail.html'}, {'title': '삼성전자 노트북 플러스2 15.6, 퓨어 화이트, NT550XDA-K24AT, 펜티엄, 256GB, 8GB, WIN11 Pro', 'link': '#product2_detail.html'}, {'title': '레노버 아이디어패드 슬림 1 15AMN7 15.6, 256GB, Free DOS, 82VG002EKR, 라이젠3, Cloud Grey (82VG), 8GB', 'link': '#product3_detail.html'}, {'title': '레노버 V15 G4 AMN 15.6, Arctic Grey, 라이젠3, 256GB, 8GB, WIN11 Home, 82YU0009KR', 'link': '#product4_detail.html'}, {'title': 'LG 울트라PC 엣지 16, 차콜 그레이, 라이젠5, 256GB, 16GB, WIN11 Home, 16U70R-GA56K', 'link': '#product5_detail.html'}, {'title': '베이직스 베이직북 14 3세대, BB1422SS, 256GB, White, WIN11 Pro, 셀러론, 8GB', 'link': '#product6_detail.html'}, {'title': '레노버 아이디어패드 슬림 5i 14IRL 14, Cloud Grey, 코어i5, 512GB, 16GB, Free DOS, 82XD002XKR', 'link': '#product7_detail.html'}, {'title': '레노버 아이디어패드 슬림 5 16IRL 16, Cloud Grey, 512GB, 16GB, Free DOS, 82XF001RKR', 'link': '#product8_detail.html'}, {'title': '에이서 스위프트 GO 16 OLED, 스틸 그레이, 코어i5, 512GB, 16GB, Free DOS, SFG16-71-51BY', 'link': '#product9_detail.html'}, {'title': '삼성전자 갤럭시북 2 15.6, 500GB, 실버, NT550XED-K78AS, 코어i7, 16GB, WIN11 Home', 'link': '#product10_detail.html'}]

디렉토리 구조

디렉토리 구조는 다음과 같다.

project/
├── app.py               # FastAPI 서버 실행
├── crawling.py          # 크롤링 관련 코드
├── database.py          # 데이터베이스 설정 및 함수
└── models.py            # 데이터베이스 모델 정의

스크립트 작성

1. DB 모델 정의 (models.py)

from sqlalchemy import Column, String, Integer
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class DataItem(Base):
    __tablename__ = 'data_items'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String)
    link = Column(String)

2. 데이터베이스 설정 (database.py)

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Base, DataItem

# SQLite 데이터베이스 연결 설정
DATABASE_URL = 'mysql+pymysql://username:password@localhost/mydatabase'
engine = create_engine(DATABASE_URL)
Base.metadata.create_all(engine)  # 테이블 생성
SessionLocal = sessionmaker(bind=engine)

def get_db_session():
    """
    새로운 데이터베이스 세션을 생성하여 반환합니다.
    """
    session = SessionLocal()
    try:
        yield session
    finally:
        session.close()

def save_to_db(data, session):
    """
    크롤링한 데이터를 DB에 저장.
    """
    for item in data:
        new_item = DataItem(title=item['title'], link=item['link'])
        session.add(new_item)
    session.commit()

def get_data_from_db(session):
    """
    DB에서 모든 데이터를 조회하여 리스트로 반환.
    """
    db_data = session.query(DataItem).all()
    return [{"title": item.title, "link": item.link} for item in db_data]

3. 크롤링 (crawling.py)

requests와 bs4을 사용하여 데이터를 크롤링 해보자.

여기서는 특정 사이트에서 title과 link를 추출하는 예시를 들었다.

import requests as rq
from bs4 import BeautifulSoup as bs

class Crawl():
    def __init__(self):
        self.url :str = 'https://startcoding.pythonanywhere.com/basic'

    @staticmethod
    def get_soup_object(resp: rq.Response) -> bs:
        return bs(resp.text, 'html.parser')
        
    def get_data(self) -> None:
        resp = rq.get(url=self.url)
        soup = self.get_soup_object(resp=resp)
        data = []        
        for item in soup.select('#product-container > div > div > div.product-body > h3 > a'):
            data.append({
                'title': item.text.strip(),
                'link': item.attrs['href']
            })
        return data

이 코드는 지정된 URL에서 데이터를 가져와서 제목과 링크 정보를 추출한다. 추출된 데이터는 리스트에 저장되도록 하였다.

4. FastAPI 서버 (app.py)

데이터를 API로 제공하기 위해 FastAPI를 사용해 API 서버를 구성한다.

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from crawling import Crawl
from database import get_db_session, save_to_db, get_data_from_db

app = FastAPI()
crawl = Crawl()

@app.get("/data")
def get_data(db: Session = Depends(get_db_session)):
    """
    DB에서 데이터를 조회하여 반환하는 API 엔드포인트.
    """
    data = get_data_from_db(db)
    return data

@app.post("/crawl")
def crawl_and_save(db: Session = Depends(get_db_session)):
    """
    크롤링한 데이터를 DB에 저장하는 API 엔드포인트.
    """
    data :list[dict] = crawl.get_data()
    save_to_db(data, db)
    return {"message": "Data crawled and saved"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

/data: 해당 엔드포인트로 요청이 오면, DB에서 데이터를 가져와 반환.
/crawl: 해당 엔드포인트로 요청이 오면, 크롤링 데이터를 DB에 저장함.

실행

python3 app.py

먼저 테이블에 아무 데이터가 들어가 있지 않은 상태이므로 /crawl 엔드포인트로 요청을 날려보자.

curl -X POST http://192.168.219.114:8000/crawl

그런 다음 /data 엔드포인트로 요청을 날려보자.

DB에 있는 데이터를 정상적으로 가져오는 것을 볼 수 있다.

python