fastapi + airflow

mkdir fastapi_pr && cd fastapi_pr하고
fastapi 컨테이너랑 airflow 컨테이너 두개 띄워서 웹사이트 띄울 예정

fastapi는 main.py가 기본인데

# main.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

이런 식으로 하면 나중에 화면에 "Hello": "World" 가 뜰 것임.

그래서 Dockerfile_airflow랑 Dockerfile_fastapi를 각각 만들고
docker-compose.yml을 만들었다. requirements.txt도 만들었다.

그리고 airflow에 대한 dags 폴더를 만들고 그 안에
lotto_crawling.py를 넣어뒀다.

구조를 보면 먼저 airflow로 크롤링을 해오고 그걸 txt형식으로 data폴더에 저장한 뒤 main.py에서 data폴더의 txt파일을 읽어와서 그걸 화면에 내보내는 식이다.

---

먼저 컨테이너를 띄우면서 문제가 좀 있었는데

Dockerfile_airflow

FROM apache/airflow:2.6.1
USER root
RUN apt-get update \
  && apt-get install -y --no-install-recommends \
        vim \
  && apt-get autoremove -yqq --purge \
  && apt-get clean \
  && rm -rf /var/lib/apt/lists/*
  
USER "${AIRFLOW_UID:-50000}:0"

Dockerfile_fastapi 처음 실패 버젼

FROM tiangolo/uvicorn-gunicorn-fastapi:python3.9
# 작업 디렉토리를 /app으로 설정
WORKDIR /app
# 애플리케이션에 필요한 파일들을 복사
COPY requirements.txt /app/
COPY main.py /app/
COPY templates /app/templates
COPY static /app/static
# 필요한 패키지들을 설치
COPY requirements.txt /app/
RUN pip install --no-cache-dir -r requirements.txt
# FastAPI 서버를 실행
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "80"]

Dockerfile_fastapi 맞는 버젼

# FastAPI 애플리케이션을 위한 베이스 이미지 선택
FROM tiangolo/uvicorn-gunicorn-fastapi:python3.9
# 작업 디렉토리를 /app으로 설정
WORKDIR /app
# 필요한 패키지들을 설치
COPY requirements.txt /app/
RUN pip install --no-cache-dir -r requirements.txt
# FastAPI 서버를 실행
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "80"]

docker-compose.yml

version: '3'
services:
  web:
    build:
      context: ./backend/
      dockerfile: Dockerfile_fastapi
    ports:
      - "80:80"
    volumes:
      - ./backend:/app/
  airflow:
    build:
      context: ./batch/
      dockerfile: Dockerfile_airflow
    command: >
      bash -c '
      airflow db init 
      &&
      airflow users create --username admin --password admin --firstname Anonymous --lastname Admin --role Admin --email test@test.com 
      &&
      airflow webserver 
      && 
      airflow scheduler'
    ports:
      - "8080:8080"
    volumes:
      - ./batch/dags:/opt/airflow/dags/
      - ./batch/data:/home/airflow/data

Dockerfile_airflow는 docker image를 살짝 바꿔서 빌드한 경량화 된
에어플로우 컨테이너인데 뒤에
docker-compose.yml을 보면 command에 airflow db init부터 해서 airflow scheduler까지 있다. 처음에 command를 안 넣고
docker-compose.yml을 구성해서 컨테이너가 켜지다가 꺼져버리는 사태가 벌어졌었다. init으로 확실하게 켜줘야한다. 그리고 localhost:8080을
들어가보면 dags에 crawl_lotto_numbers가 안 뜨는데 이게

airflow webserver 
&& 
airflow scheduler

airflow scheduler가 나오기 전에 timesleep 같은 것이 있어야해서 그렇다. 최근 오류라고 함.

그래서 dags에 띄우기 위해서

docker exec -it fastapi_pr-airflow-1 /bin/bash로 들어가서
airflow scheduler -D를 쳐서 실행을 해주면 dags에 잘 뜬다.

그리고 Dockerfile_fastapi에서도 문제가 있었는데, 계속 127.0.0.1:80으로 들어갔더니 "hello": "world"만 뜨는 것이다.

처음에 실패 버젼을 보면

COPY requirements.txt /app/
COPY main.py /app/
COPY templates /app/templates
COPY static /app/static

이런 부분들이 있는데 컨테이너에 정적 파일이나 소스 코드에 대한 의존성을 주입해서 그렇다.
이미지에서 저 녀석들을 구워 넣어버려서 그 뒤에 main.py를 수정해도 main.py가 맨 처음

# main.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

이 상태였던 것이다.

이것도 docker exec -it fastapi_pr-web-1 /bin/bash로 들어가서 main.py 파일을 읽어봤더니 알 수 있었다.

이럴땐 컨테이너랑 이미지 다 지워버리고 다시 Dockerfile_fastapi를 고친 다음 다시 컨테이너를 빌드해야한다.

COPY부분을 지워버리고 docker-compose.yml에서 볼륨 공유를 하는 것을 통해 문제를 해결하였다.

docker-compose.yml에서

build:
      context: ./batch/
      dockerfile: Dockerfile_airflow
command: >
      bash -c '
      airflow db init 
      &&
      airflow users create --username admin --password admin --firstname Anonymous --lastname Admin --role Admin --email test@test.com 
      &&
      airflow webserver 
      && 
      airflow scheduler'
ports:
      - "8080:8080"
volumes:
      - ./batch/dags:/opt/airflow/dags/
      - ./batch/data:/home/airflow/data

크게 build, command, ports, volumes가 있는데
build에서 context는 경로, dockerfile은 도커파일명을 말한다.
volumes에서 :를 기준으로 앞부분은 로컬에서의 경로고 뒷부분은 컨테이너 내부에서의 경로이다. 컨테이너 내부 경로는 공식문서 참조하는게 맞다.

그래서 보통은 chatgpt한테 물어볼 때, 볼륨공유를 어떻게 할 것인지 미리 얘기를 해주는 등의 과정을 통해 도커파일을 작성하는게 맞다.

---

이제 dags/lotto_crawling.py를 보면

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.operators.dummy_operator import DummyOperator
import requests
from bs4 import BeautifulSoup as BS
import os
from datetime import datetime

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 1, 1),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

def crawl_lotto_numbers():
    url = "https://dhlottery.co.kr/common.do?method=main"
    payload = {"method" : "getMainBannerList"}
    r = requests.post(url, data=payload)
    bs = BS(r.text, 'html.parser')
    lotto_numbers = []
    for x in bs.find("a", {"id": "numView"}).findAll("span"):
        lotto_numbers.append(x.text)
        
     # 파일 이름을 날짜별로 다르게 만들기
    current_date = datetime.now().strftime("%Y-%m-%d")
    file_name = f"./data/numbers_{current_date}.txt"

    # 크롤링한 로또 번호 리스트를 새로운 파일에 저장
    with open(file_name, "w") as file:
        for number in lotto_numbers:
            file.write(number + "\n")

    # lotto_numbers를 return하지 않고 None을 리턴
    # 왜냐하면 DAG 내에서 PythonOperator의 리턴 값은 무시되기 때문에
    return None

# Airflow DAG 설정
with DAG(
    'lotto_crawling_dag',
    default_args=default_args,
    schedule_interval=timedelta(days=7),  # 매주 일요일 0시 0분에 실행
) as dag:
    start = DummyOperator(task_id='start')
    crawl_task = PythonOperator(
        task_id='crawl_lotto_numbers',
        python_callable=crawl_lotto_numbers,
    )
    end = DummyOperator(task_id='end')

# DAG에 추가적인 Task를 만들 필요 없이 크롤링 Task만 작성하여 실행할 수 있습니다.
start >> crawl_task >> end

이런데 맨 밑 as dag 뒤에 start, crawl_task, end 이렇게 세 개 순서를 정해주는게 중요함. start랑 end는 DummyOperator를 꼭 import 해야 한다.

중간에 크롤링 해온걸 data폴더에 날짜에 대한 이름으로 파일로 저장한다.

main.py를 보면

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from datetime import datetime

app = FastAPI()

# 정적 파일 (CSS, JS 등)을 제공하기 위해 static 디렉토리를 설정합니다.
app.mount("/static", StaticFiles(directory="static"), name="static")

# FastAPI에서 템플릿을 렌더링하기 위해 Jinja2 템플릿 객체를 생성합니다.
templates = Jinja2Templates(directory="templates")

def get_current_week_numbers_file():
    # 현재 날짜를 구합니다.
    current_date = datetime.now().strftime("%Y-%m-%d")
    # 현재 날짜에 해당하는 파일 이름을 만듭니다.
    file_name = f"./data/numbers_{current_date}.txt"
    return file_name

@app.get("/", response_class=HTMLResponse)
async def read_root(request: Request):
    try:
        # 현재 날짜에 해당하는 파일을 읽어옵니다.
        file_name = get_current_week_numbers_file()
        with open(file_name, "r") as file:
            lotto_numbers = file.read().splitlines()
    except FileNotFoundError:
        # 파일이 없을 경우 빈 리스트로 초기화
        lotto_numbers = []

    # 가져온 로또 번호 리스트를 템플릿에 전달하여 렌더링합니다.
    return templates.TemplateResponse("index.html", {"request": request, "lotto_numbers": lotto_numbers})

from fastapi.responses import HTMLResponse 이거 import 받는거..
HTMLResponse 이거 chatgpt가 잘못 알려줘서 고생 좀 했다.
링크텍스트

보면 app, templates 객체 만들고 data에서 파일 읽어와서 화면에 나오게 하는 코드이다.

항상 경로 써주는거 주의하자.

나중에 fastapi 관련된 것들은 모두 backend 폴더에 집어넣고,
airflow 관련된 것들은 모두 batch 폴더에 집어넣었다.

venv로 로컬에서 한번씩 띄워보는 실험하면 좋고, 마지막에 깔끔하게 폴더에 몰아넣으면 좋은데 그럴때 상대경로가 유용하다는 것 기억하자.