DBMS 및 SQL 활용 #1 실습 #
#2025-08-27
1. 실습1 #
실습 시나리오
- 사용자가 설계안 텍스트(예: description)를 입력
- 해당 텍스트에 대해 Python에서 AI 임베딩을 수행
- 임베딩 결과가 유효할 경우 design 테이블에 등록 (COMMIT)
- 실패하면 아무 데이터도 등록하지 않음 (ROLLBACK)
- PostgreSQL + pgvector 확장 사용
- Python에서 psycopg2 + 임베딩 처리
#
코드
#1 SQL
CREATE EXTENSION IF NOT EXISTS vector;
CREATE TABLE IF NOT EXISTS design (
id SERIAL PRIMARY KEY,
description TEXT,
embedding VECTOR(1536) -- OpenAI 임베딩 차원
);
#
#2 python
import psycopg2
import pandas as pd
import os
from dotenv import load_dotenv
from openai import OpenAI
os.chdir("/Users/yshmbid/Documents/home/github/SQL")
# 1. .env 파일 로드
load_dotenv()
# 2. OpenAI Client 생성
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) # .env에서 OPENAI_API_KEY 가져옴
# 3. PostgreSQL DB에 연결
conn = psycopg2.connect(
host="localhost",
port=5432,
database="postgres",
user="postgres",
password=os.getenv("PG_PASSWORD"), # .env에서 PG_PASSWORD 가져옴
)
cursor = conn.cursor() # SQL문 실행
# 4. 임베딩 함수
def get_embedding(text: str):
response = client.embeddings.create(
input=text,
model="text-embedding-3-small" # “text-embedding-3-small” 모델로 임베딩 생성
)
return response.data[0].embedding
# 5. DB 삽입 함수
def insert_design(description: str):
try:
cursor.execute("BEGIN;") # 트랜잭션 시작
# 임베딩 생성
embedding = get_embedding(description)
# design 테이블에 삽입
cursor.execute(
"INSERT INTO design (description, embedding) VALUES (%s, %s)",
(description, embedding)
)
conn.commit()
print(f"[COMMIT] 등록 성공 → {description[:40]}...")
except Exception as e:
conn.rollback()
print(f"[ROLLBACK] 실패 → {description[:40]}... 에러: {e}")
# 6. CSV 파일 로드 & 처리
df = pd.read_csv("sample_designs_500.csv")
for idx, row in df.iterrows():
desc = row.get("description")
if pd.notna(desc): # description이 비어있지 않을 때만 실행
insert_design(desc)
# 7. 연결 종료
cursor.close()
conn.close()
#
#3 시나리오 구현
- 사용자가 설계안 텍스트(예: description)를 입력
- insert_design(desc) -> description(desc): str
- 해당 텍스트에 대해 Python에서 AI 임베딩을 수행
- get_embedding(description) -> client.embeddings.create
- 임베딩 결과가 유효할 경우 design 테이블에 등록 (COMMIT)
- insert_design -> conn.commit()
- 실패하면 아무 데이터도 등록하지 않음 (ROLLBACK)
- insert_design -> except Exception as e -> conn.rollback()
#
#4 개념
트랜젝션?
- DB에서 여러 SQL 실행을 하나의 작업 단위로 묶는것
- 여러 SQL 실행?
- BEGIN; (시작) / INSERT … (데이터 넣기) / UPDATE … (데이터 수정하기) / COMMIT; (끝내기 → 확정 반영) 등 commit?
- commit 전에는 cursor.execute를 실행해도 DB 내부 버퍼/임시 상태에만 반영됨.
- commit을 하면 변경사항을 실제 DB 파일(디스크)에 확정 저장되고 다른 클라이언트(psql, pgAdmin 등)에서도 데이터를 조회 가능.
#
2. 실습2 #
실습 시나리오
- FastAPI 기반 /register_design API를 구현해보세요(Python)
- Streamlit 를 통해 입력 UI를 만들고 위에 만든 FastAPI를 호출하는 방식으로 해보세요.
아래의 순서대로 진행해보세요.
- PostgreSQL의
design
테이블 (생성됨) - FastAPI 서버 실행:
uvicorn app:app --reload
- Streamlit 클라이언트 실행:
streamlit run streamlit_client.py
- 입력 → POST → 등록 확인
#
코드
#1 FastAPI 서버
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import psycopg2
from dotenv import load_dotenv
from openai import OpenAI
import os
# 경로 설정
os.chdir("/Users/yshmbid/Documents/home/github/SQL")
# .env 로드
load_dotenv()
# OpenAI 클라이언트
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
# DB 연결
conn = psycopg2.connect(
host="localhost",
port=5432,
database="postgres",
user="postgres",
password=os.getenv("PG_PASSWORD"),
)
cursor = conn.cursor()
# FastAPI 앱 객체 생성
app = FastAPI()
# 요청 데이터 모델 정의
class DesignRequest(BaseModel):
description: str
# 임베딩 함수
def get_embedding(text: str):
response = client.embeddings.create(
model="text-embedding-3-small",
input=text
)
return response.data[0].embedding
@app.post("/register_design")
def register_design(req: DesignRequest):
try:
cursor.execute("BEGIN;")
embedding = get_embedding(req.description)
cursor.execute(
"INSERT INTO design (description, embedding) VALUES (%s, %s)",
(req.description, embedding)
)
conn.commit()
return {"status": "success", "message": "등록 성공"}
except Exception as e:
conn.rollback()
raise HTTPException(status_code=500, detail=f"등록 실패: {e}")
#2 Streamlit 클라이언트
import streamlit as st
import requests
import os
# 경로 설정
os.chdir("/Users/yshmbid/Documents/home/github/SQL")
st.title("Design 등록 클라이언트")
# 입력 박스
description = st.text_area("설계안 입력", "")
if st.button("등록하기"):
if description.strip() == "":
st.warning("설계안을 입력해주세요.")
else:
try:
response = requests.post(
"http://127.0.0.1:8000/register_design",
json={"description": description}
)
if response.status_code == 200:
st.success(response.json())
else:
st.error(response.json())
except Exception as e:
st.error(f"서버 연결 실패: {e}")
#
#3 시나리오 구현
- FastAPI 서버- class DesignRequest
- Input: 사용자가 Streamlit 화면에서 입력한 설계안 텍스트를 json {“description”: “텍스트”} 로 변환
- Pydantic이 json을 검증후 python 객체(req.description)로 변환
- Output: req.description (문자열)
- FastAPI 서버- register_design()
- Input: req.description (문자열)
- OpenAI API 호출해서 임베딩 벡터 생성 → PostgreSQL design 테이블에 (description, embedding) 저장
- Output: 성공/실패 메시지 JSON 응답 ({“status”: “success”, “message”: “등록 성공”})
- Streamlit
- Input: 사용자가 입력 설계안 description 텍스트
- FastAPI에 전송하면 json {“description”: “텍스트”} 로 감싸서 fastapi에 POST 요청 → description 데이터 등록
- Output: 성공 실패 메시지 표시
#
#4 개념
class DesignRequest와 register_design()와의 호환?
- JSON을 파싱해서 Python 객체로 바꾸고 description이 문자열인지 검증한 뒤 통과하면 register_design()에서 DesignRequest 객체를 만들어 req에 넣는다.
롤백?
- rollback을 안 하면 “INSERT는 됐는데 commit 전에 에러 발생” 같은 상태가 DB에 남을 수 있음.
#
3. 실습2 - 레퍼런스 코드 #
# AI 설계안 등록 통합 실습 (FastAPI + Streamlit + PostgreSQL)
## FastAPI 서버 (app.py)
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from sentence_transformers import SentenceTransformer
import psycopg2
app = FastAPI()
model = SentenceTransformer('all-MiniLM-L6-v2')
def get_db_conn():
return psycopg2.connect(
dbname="yourdb", user="youruser", password="yourpass", host="localhost"
)
class DesignInput(BaseModel):
title: str
description: str
@app.post("/register_design")
def register_design(data: DesignInput):
conn = get_db_conn()
cur = conn.cursor()
try:
embedding = model.encode(data.description).tolist()
cur.execute("BEGIN;")
cur.execute("""
INSERT INTO design (title, description, embedding)
VALUES (%s, %s, %s);
""", (data.title, data.description, embedding))
conn.commit()
return {"status": "success", "message": "등록 완료"}
except Exception as e:
conn.rollback()
raise HTTPException(status_code=500, detail=f"등록 실패: {str(e)}")
finally:
cur.close()
conn.close()
## Streamlit 클라이언트 (streamlit_client.py)
import streamlit as st
import requests
st.title("AI 설계안 등록")
title = st.text_input("설계안 제목")
description = st.text_area("설계안 설명")
if st.button("등록 요청"):
response = requests.post("http://localhost:8000/register_design", json={
"title": title,
"description": description
})
if response.status_code == 200:
st.success(response.json()["message"])
else:
st.error(response.json()["detail"])
## 실행 순서 요약
1. PostgreSQL에 `design` 테이블 생성
2. FastAPI 서버 실행: `uvicorn app:app --reload`
3. Streamlit 클라이언트 실행: `streamlit run streamlit_client.py`
4. 입력 → POST → 등록 확인
## 테스트용 CSV & SQL 삽입 예제
- `sample_designs_500.csv`: 샘플 설계안 + 임베딩 포함
- `insert_designs.sql`: INSERT 문 자동 생성
**Tip**: Python에서 `psycopg2`로 임베딩 포함 대량 INSERT 가능