[Front/Web + Back]streamlit과 FastAPI로 시스템 구축

kk21·2025년 8월 27일
0

개발채널 K

목록 보기
1/4

FastAPI + DB 실습을 위해 환경을 미리 구축해두려고 한다

물론 이전에 쓰던 환경이 있지만..

어차피 배운 거 많이 복습해야 하고
많이 쓸수록 손에 익으니 실습을 할 때 마다 새로 환경을 구축하는 편

사실 크게 하는 게 많은 거 아니다

  1. 도커로 컨테이너 구축 (환경 분리 및 도커 복습)
  2. 컨테이너 외부 접속 포트 포워딩 (웹 접속을 위해..)
  3. 내부 DB 구축 (그냥 DB 새로 만드는 거 밖에 없음)
  4. 필요 라이브러리 설치

보통 이정도?
실습을 하면서 할 때도 있고 미리할 때도 있는데
내가 DB가 약한 편이라 항상 따라가기 조금 벅찰 때가 있어서

DB 때는 미리미리 준비해두는 편...

원래라면 걍 혼자 구축하고 끝이었겠지만 이번 실습에는 streamlit + FastAPI + PostgreSQL이 사용되기 때문에
좀 할게 많아서 천천히 정리해보려고 한다

사실 이렇게 프론트 + 백 + 인프라까지 다 구축하는 건 처음이라서 그러는 것도 있다

나중에 프로젝트 할 떄 도움이 되겠지 🙂‍↕️

쨋든 주절 주절이 끝났으니 이제 천천히 해보도록 하자

1. 도커 컨테이너 생성

1.1 작업 디렉토리 생성

도커 컨테이너를 만들면 서버 어딘가 가상 디렉토리로 만들어진다
그러면 나중에 파일을 보기 힘들기 때문에 특정 디렉토리를 만들고 그 디렉토리에 연결해서 작업 디렉터리로 쓰려고 한다

참고로 컨테이너 만드는 환경은 이미 우분투이다 !!

그럼 먼저 디렉터리 부터 만들자 

mkdir fast_ex

fast_ex라는 이름으로 디렉터리를 하나 만든 것이다

1.2 디렉터리 이동

cd fast_ex

작업 디렉터리로 이동하자

1.3 Dockerfile 작성

컨테이너를 만들기 위해 Dockerfile을 작성해보자 ..

도커 파일 내용은 아래와 같다

파일 다운로드는 아래 깃허브의 streamlit + FastAPI + PostgreSQL 버전
https://github.com/bkk21/Dokerfile

1.4 이미지 빌드

docker build -t fast_ex_img .

fast_ex_img 라는 이름으로 도커 이미지를 만든다

docker images

만든 이미지를 확인해보자

1.5 도커 실행

docker run -it --name fast_ex -p 9436:8501 -v ~/fast_ex:/app fast_ex_img

컨테이너 이름은 fast_ex, 포트는 외부포트 9436을 내부 8501에 연결 그리고 작업 디렉토리인 fast_ex 디렉터리를 내부 디렉터리인 app 디렉터리에 연결 사용할 이미지는 fast_ex_img

다른 터미널에서 컨테이너 접속

docker exec -it fast_ex bash

1.6 파이썬 버전 확인

python3 --version

1.7 streamlit 버전 확인

streamlit --version

2. DB - PostgreSQL

2.1 qsql 설치 확인

psql --version

2.2 클러스터 상태 확인

pg_ctlcluster 16 main status || true

2.3 클러스터 시작

pg_ctlcluster 16 main start

2.4. 클러스터 확인

pg_isready -h 127.0.0.1 -p 5432

2.5 psql 관리자 계정 접속

su - postgres -c "psql"

2.6 계정 생성

CREATE USER [유저명] WITH PASSWORD '[비번]' LOGIN;

\du

2.7 계정에 createdb권한 주기

su - postgres -c "psql -c \"ALTER ROLE sk_user CREATEDB;\""

2.8. 계정 접속

psql -U sk_user -d postgres -h localhost -p 5432

2.9 DB 생성

CREATE DATABASE sk_db OWNER sk_user ENCODING 'UTF8';

\list

\c

2.10 특정 계정 특정 DB 접속

psql -U sk_user -d sk_db -h localhost -p 5432

2.11 pgvertor 확장 설치 (슈퍼유저)

su - postgres -c "psql -d sk_db -c 'CREATE EXTENSION IF NOT EXISTS vector;'"

2.12 pgvertor 설치 확인 (특정 계정)

psql -U sk_user -d sk_db -h localhost -p 5432 -c "\dx"

3. FastAPI

먼저 FastAPI가 잘 되는지 간단하게 확인해보자

3.1 테스트 파일 작성

먼저 api.py라는 파일을 만들고 아래 내용을 작성한다

from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def ping():
    return {"status": "ok"}

3.2 테스트 위한 jq 설치

apt-get update

apt-update 진행

apt-get install -y --no-install-recommends jq

3.3 파이썬 실행기 선책

if command -v python3.11 >/dev/null; then PY=python3.11; else PY=python3; fi

3.4 필요한 패키지 설치

$PY -m pip install --upgrade pip
$PY -m pip install "uvicorn[standard]" fastapi

3.5 API 서버 실행

$PY -m uvicorn api:app --host 0.0.0.0 --port 8000 --reload &

3.6 테스트

curl -s http://127.0.0.1:8000/ | $PY -m json.tool

200이 뜨면 정상인데 잘 뜨고 있다
굿굿

3.7 .env 패키지 설치

파이썬에서 .env 사용을 위해 패키지를 설치하자

python3 -m pip install python-dotenv

3.8. FastAPI 설치

python3 -m pip install FastAPI

3.9 DB랑 통신 코드 작성

파일 이름은 api_vec.py 작성

from fastapi import FastAPI
from pydantic import BaseModel
from typing import List
import os, psycopg2
app = FastAPI(title="FastAPI + pgvector demo")
# 환경변수(없으면 기본값)
PGHOST = os.getenv("PGHOST", "localhost")
PGPORT = int(os.getenv("PGPORT", "5432"))
PGDATABASE = os.getenv("PGDATABASE", "sk_db")
PGUSER = os.getenv("PGUSER", "sk_user")
PGPASSWORD = os.getenv("PGPASSWORD", "sk_pw_123!")
def get_conn():
    # DSN 문자열로 접속 (URL 인코딩 이슈 회피)
    dsn = f"host={PGHOST} port={PGPORT} dbname={PGDATABASE} user={PGUSER} password={PGPASSWORD}"
    return psycopg2.connect(dsn)
def vec_literal(v):
    return "[" + ",".join(str(float(x)) for x in v) + "]"
class UpsertIn(BaseModel):
    text: str
    embedding: List[float]  # 예: 길이 3
@app.post("/upsert")
def upsert(inp: UpsertIn):
    emb = vec_literal(inp.embedding)
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                "INSERT INTO items (text, embedding) VALUES (%s, %s::vector) RETURNING id;",
                (inp.text, emb),
            )
            new_id = cur.fetchone()[0]
    return {"id": new_id, "text": inp.text}
class SearchIn(BaseModel):
    embedding: List[float]
    top_k: int = 5
@app.post("/search")
def search(inp: SearchIn):
    emb = vec_literal(inp.embedding)
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT id, text FROM items ORDER BY embedding <-> %s::vector LIMIT %s;",
                (emb, inp.top_k),
            )
            rows = cur.fetchall()
    return [{"id": r[0], "text": r[1]} for r in rows]
@app.get("/")
def ping():
    return {"status": "ok"}

3.10 실행

python3.11 -m uvicorn api_vec:app --host 0.0.0.0 --port 8000 --reload &

3.11 테스트

curl -s http://127.0.0.1:8000/ | python3 -m json.tool

4. Streamlit

4.1 버전 확인

streamlit --version

4.2 streamlit으로 파일 작성

app.py라는 이름으로 파일을 작성하면 된다

import os, json, requests, streamlit as st
API_URL = os.getenv("API_URL", "http://localhost:8000")
st.set_page_config(page_title="Streamlit + FastAPI + pgvector (TEST_v1)", layout="centered")
st.title("🔗 TEST_v1 — 저장 & 검색")
# 서버 메타 가져와서 차원/테이블 확인
try:
    meta = requests.get(f"{API_URL}/meta", timeout=10).json()
    DIM = int(meta.get("dim", 3))
    TABLE = meta.get("table", "TEST_v1")
    st.caption(f"DB: {meta.get('db')} | Table: {TABLE} | Dim: {DIM}")
except Exception as e:
    DIM = int(os.getenv("VEC_DIM", "3"))
    TABLE = "TEST_v1"
    st.warning(f"메타 조회 실패. Dim={DIM}로 가정. ({e})")
# --- 저장 ---
st.subheader("1) TEST_v1에 저장 (Upsert)")
text = st.text_input("텍스트", value="hello world")
emb_str = st.text_input(f"임베딩(쉼표로 {DIM}개)", value="0.1, 0.2, 0.3" if DIM==3 else "")
if st.button("DB에 저장"):
    try:
        emb = [float(x.strip()) for x in emb_str.split(",") if x.strip() != ""]
        if len(emb) != DIM:
            st.error(f"임베딩 길이는 {DIM} 이어야 합니다. (현재 {len(emb)})")
        else:
            r = requests.post(f"{API_URL}/upsert", json={"text": text, "embedding": emb}, timeout=30)
            r.raise_for_status()
            st.success(f"저장 성공: {r.json()}")
    except Exception as e:
        st.error(f"실패: {e}")
# --- 검색 ---
st.subheader("2) 유사도 검색 (Cosine)")
q_str = st.text_input(f"쿼리 임베딩(쉼표로 {DIM}개)", value="0.12, 0.21, 0.31" if DIM==3 else "")
top_k = st.number_input("top_k", min_value=1, max_value=100, value=3, step=1)
if st.button("검색"):
    try:
        q = [float(x.strip()) for x in q_str.split(",") if x.strip() != ""]
        if len(q) != DIM:
            st.error(f"임베딩 길이는 {DIM} 이어야 합니다. (현재 {len(q)})")
        else:
            r = requests.post(f"{API_URL}/search", json={"embedding": q, "top_k": int(top_k)}, timeout=30)
            r.raise_for_status()
            st.write(r.json())
    except Exception as e:
        st.error(f"실패: {e}")
# --- 최근 N건 보기 ---
st.subheader("3) 최근 레코드")
try:
    lst = requests.get(f"{API_URL}/list", timeout=10, params={"limit": 10}).json()
    st.table(lst)
except Exception as e:
    st.info(f"최근 레코드 조회 실패: {e}")

5. 실행

5.1 파일 구조 확인

/app
├─ .env
├─ api/
│  └─ main.py
└─ web/
   └─ app.py

5.2 백엔드 실행

터미널을 하나 띄우고 백엔드 실행 명령어 입력

pkill -f "uvicorn.*8000" || true
python3.11 -m uvicorn api.main:app \
  --host 0.0.0.0 --port 8000 --reload --app-dir /app

5.3 프론트 실행

다른 터미널 하나 더 띄워서 실행 명령어 입력 

python3.11 -m streamlit run /app/web/app.py \
  --server.address=0.0.0.0 \
  --server.port=8501


6. 동작 테스트

6.1 웹에서 테스트 해보자

웹에서 저장 버튼을 누르면 통신하고 결과값을 받아와서 화면에 출력한다

이렇게 유사도 검색도 가능!

이렇게 최근 레코드도 볼 수 있다 ~

그럼 기록 끝 !!!

profile
kk21
LLM Engineer의 성장 일기 ing. . . ✨
다음 포스트

Slack 채널 알림봇 만들기

0개의 댓글