in c prac (리드미) 업데이트
DB 저장 및 파싱 추가
리스트 조회 추가
# in-C Backend 반복 구현 연습 가이드 (v2)
이 가이드는 in-C 프로젝트의 백엔드 핵심 기능(스크래퍼 추상화, 데이터 파싱, DB 연동)을 처음부터 구현하며 숙달하기 위한 연습 문서입니다.
## 1. 프로젝트 초기 설정
```bash
# 1. 폴더 생성 및 이동
mkdir backend && cd backend
# 2. 가상환경 설정
python -m venv venv
source venv/bin/activate # Mac/Linux (Windows는 venv\Scripts\activate)
# 3. 필수 라이브러리 설치
pip install fastapi uvicorn requests sqlalchemy
SQLite를 사용하여 로컬 데이터를 관리합니다.
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# SQLite 데이터베이스 URL
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
# DB 세션 의존성 주입을 위한 함수
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
수집한 공연 정보를 저장할 테이블 구조를 정의합니다.
from sqlalchemy import Column, String, Date, DateTime
from app.core.database import Base
import uuid
from datetime import datetime, timezone
class Concert(Base):
__tablename__ = "concerts"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
title = Column(String, nullable=False)
venue_detail = Column(String)
start_date = Column(Date)
end_date = Column(Date)
created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
단순 수집을 넘어, 데이터를 가져오는(Fetch) 로직과 정제하는(Parse) 로직을 분리하여 확장성을 확보합니다.
from abc import ABC, abstractmethod
from typing import List, Dict, Any
class BaseScraper(ABC):
@abstractmethod
def fetch_raw_data(self) -> Any:
"""외부 원천 데이터를 가져옵니다."""
pass
@abstractmethod
def parse(self, raw_data: Any) -> List[Dict[str, Any]]:
"""원천 데이터를 정형화된 리스트로 변환합니다."""
pass
def collect(self) -> List[Dict[str, Any]]:
"""수집 프로세스 실행 (Fetch -> Parse)"""
raw = self.fetch_raw_data()
return self.parse(raw)
app/scrapers/sac_scraper.py
import requests
from .base import BaseScraper
class SACScraper(BaseScraper):
def __init__(self, hall_id: str, base_url: str):
self.hall_id = hall_id
self.base_url = base_url
def fetch_raw_data(self):
params = {"CATEGORY_SECONDARY": "000005", "pageSize": "10"}
headers = {"X-Requested-With": "XMLHttpRequest"}
response = requests.get(self.base_url, params=params, headers=headers)
response.raise_for_status()
return response.json()
def parse(self, raw_data):
results = raw_data.get("paging", {}).get("result", [])
concerts = []
for item in results:
concerts.append({
"title": item.get("PROGRAM_SUBJECT"),
"venue_detail": item.get("CATEGORY_PRIMARY_NAME"),
"start_date": item.get("BEGIN_DATE", "").replace(".", "-"),
"end_date": item.get("END_DATE", "").replace(".", "-")
})
return concerts
어떤 공연장이 들어와도 유연하게 처리할 수 있게 해줍니다.
from .sac_scraper import SACScraper
class ScraperFactory:
@staticmethod
def get_scraper(scraper_type: str, hall_id: str, base_url: str):
if scraper_type == "SEOUL_ARTS_CENTER":
return SACScraper(hall_id, base_url)
raise ValueError(f"Unknown scraper type: {scraper_type}")
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from datetime import datetime
from app.core.database import engine, Base, get_db
from app.models.models import Concert
from app.scrapers.factory import ScraperFactory
Base.metadata.create_all(bind=engine)
app = FastAPI()
@app.get("/collect")
def collect_concerts(hall_type: str = "SEOUL_ARTS_CENTER", db: Session = Depends(get_db)):
# 1. 팩토리를 통해 스크래퍼 생성
scraper = ScraperFactory.get_scraper(
hall_type,
hall_id="sac-123",
base_url="https://www.sac.or.kr/site/main/show/dataList"
)
# 2. 데이터 수집 및 DB 저장
data_list = scraper.collect()
for data in data_list:
try:
# 중복 체크 로직 생략(연습용), 날짜 변환 안전 처리
start_date = datetime.strptime(data["start_date"], "%Y-%m-%d").date() if data["start_date"] else None
end_date = datetime.strptime(data["end_date"], "%Y-%m-%d").date() if data["end_date"] else None
db.add(Concert(
title=data["title"],
venue_detail=data["venue_detail"],
start_date=start_date,
end_date=end_date
))
except Exception as e:
print(f"Error parsing data: {e}")
continue
db.commit()
return {"status": "success", "collected": len(data_list)}
@app.get("/concerts")
def list_concerts(db: Session = Depends(get_db)):
return db.query(Concert).all()
uvicorn app.main:app --reload
http://127.0.0.1:8000/collect?hall_type=SEOUL_ARTS_CENTER 호출로 수집 및 저장 확인.


레벨 업~