어렵다...

학습시간 09:00~21:00(당일12H/누적2024H)

---

◆ 학습내용

실습!!

---

7. Base Response API

문제

response 모델의 기본구조를 만들어주세요
    - 모든 response 에 공통적으로 포함될 필드를 정의
    success : bool
    timestamp : str (datetime.now().isoformat())
    version : str ("1.0")

class BaseResponse(

코드

from fastapi import FastAPI, Response, status
from pydantic import BaseModel, Field
from datetime import datetime
from typing import Generic, TypeVar, Optional, List

app = FastAPI()

# --- 1. BaseResponse 모델 정의 ---

T = TypeVar('T') # 모든 타입을 받을 수 있는 제네릭 타입 변수 선언

class BaseResponse(BaseModel, Generic[T]):
    success: bool = True
    timestamp: str = Field(default_factory=lambda: datetime.now().isoformat())
    version: str = "1.0"
    data: Optional[T] = None # 실제 데이터를 담을 필드, 제네릭 타입으로 선언

# --- 2. BaseResponse 모델 활용 예시 ---

class User(BaseModel):
    id: int
    name: str
    email: str

# 샘플 데이터베이스
users_db = {
    1: User(id=1, name="홍길동", email="hong@example.com"),
    2: User(id=2, name="이순신", email="lee@example.com"),
}

@app.get("/users/{user_id}", response_model=BaseResponse[User])
def get_user(user_id: int):
    user = users_db.get(user_id)
    if not user:
        # 실패 시 BaseResponse 형태를 직접 만들어 반환
        return Response(
            content=BaseResponse[User](success=False, data=None).model_dump_json(),
            status_code=status.HTTP_404_NOT_FOUND,
            media_type="application/json"
        )
    # 성공 시 데이터를 BaseResponse로 감싸서 반환
    return BaseResponse[User](data=user)

@app.get("/users", response_model=BaseResponse[List[User]])
def get_all_users():
    all_users = list(users_db.values())
    # 성공 시 리스트 데이터를 BaseResponse로 감싸서 반환
    return BaseResponse[List[User]](data=all_users)

BaseResponse 모델 상세

• Generic[T]
  • BaseResponse 모델이 다양한 타입의 데이터를 담을 수 있도록 제네릭(Generic)을 사용
  • T는 '타입 변수'로, User, List[User], str 등 어떤 타입이든 올 수 있음
  • 이를 통해 매번 새로운 응답 모델을 만들 필요 없이 BaseResponse를 재사용 가능
• success: bool = True
  • API 요청의 성공 여부를 나타내는 필드
  • 기본값을 True로 설정하여 성공적인 경우를 표준으로 함
• timestamp: str = Field(default_factory=...)
  • 응답이 생성된 시간을 ISO 8601 형식의 문자열로 나타내는 필드
  • default_factory를 사용하여 BaseResponse 모델 객체가 생성될 때마다 datetime.now().isoformat() 함수가 동적으로 호출되도록 함
• version: str = "1.0"
  • API의 버전을 나타내는 필드
• data: Optional[T] = None
  • 실제 응답 데이터를 담는 핵심 필드
  • 제네릭 타입 T를 사용하여 단일 객체(User), 객체 리스트(List[User]) 등 다양한 형태의 데이터를 담을 수 있음
  • Optional과 기본값 None을 사용하여 데이터가 없는 경우(실패 등)도 표현 가능

BaseResponse 활용

• response_model=BaseResponse[User]
  • API 엔드포인트의 응답 모델을 지정할 때, 제네릭 문법을 사용해 data 필드에 User 모델이 들어갈 것임을 명시
• response_model=BaseResponse[List[User]]
  • data 필드에 User 모델의 리스트가 들어갈 것임을 명시
• 성공 시 반환: return BaseResponse[User](data=user)
  • 조회한 user 객체를 BaseResponse로 감싸서 반환
  • FastAPI가 response_model 정의에 따라 자동으로 JSON 형식으로 변환해 줌
• 실패 시 반환: Response 객체 직접 생성
  • BaseResponse 모델의 success를 False로 설정하고 data를 None으로 하여 실패 응답 본문을 직접 생성
  • FastAPI의 Response 객체를 사용하여 상태 코드(404), 미디어 타입 등을 명시적으로 제어하여 반환

---

8. Product 도메인 정의 API

문제

product 도메인이랑 권한을 정의
1. 상품 상태 : active / inactive / out_of_stock
class ProductStatus

2. 상품의 생성/수정에 사용할 Product
: name : 1~100자
  description: str
  price : >0
  category : str
  status : active

코드

from enum import Enum
from typing import Optional
from pydantic import BaseModel, Field, ValidationError

# 1. 상품 상태(ProductStatus)를 Enum으로 정의
class ProductStatus(str, Enum):
    ACTIVE = "active"
    INACTIVE = "inactive"
    OUT_OF_STOCK = "out_of_stock"

# 2. 상품(Product) Pydantic 모델 정의
class Product(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    description: Optional[str] = None
    price: float = Field(gt=0)
    category: str
    status: ProductStatus = ProductStatus.ACTIVE

# --- 모델 동작 테스트 ---
if __name__ == "__main__":
    print("--- 1. 정상 데이터 테스트 (상태 기본값 사용) ---")
    try:
        product1 = Product(
            name="노트북",
            price=1500000.0,
            category="전자기기"
        )
        print("성공: 유효한 상품 데이터입니다.")
        print(product1.model_dump_json(indent=2))
    except ValidationError as e:
        print(f"실패: {e}")

    print("\n--- 2. 정상 데이터 테스트 (상태 명시적 지정) ---")
    try:
        product2 = Product(
            name="키보드",
            price=85000.0,
            category="컴퓨터 주변기기",
            status=ProductStatus.OUT_OF_STOCK # Enum 멤버로 상태 지정
        )
        print("성공: 유효한 상품 데이터입니다.")
        print(product2.model_dump_json(indent=2))
    except ValidationError as e:
        print(f"실패: {e}")

    print("\n--- 3. 유효성 검사 실패 테스트 (가격이 0 이하) ---")
    try:
        invalid_product = Product(
            name="마우스",
            price=0, # price는 0보다 커야 함
            category="컴퓨터 주변기기"
        )
    except ValidationError as e:
        print(f"성공 (에러 발생 예상):\n{e}")

    print("\n--- 4. 유효성 검사 실패 테스트 (잘못된 상태 값) ---")
    try:
        invalid_status_product = Product(
            name="모니터",
            price=300000,
            category="전자기기",
            status="sold_out" # 허용되지 않은 상태 값
        )
    except ValidationError as e:
        print(f"성공 (에러 발생 예상):\n{e}")

ProductStatus Enum 정의

• 상품 상태 값을 특정 값들(active, inactive, out_of_stock)로만 제한하기 위해 사용
• enum 라이브러리의 Enum을 상속받아 클래스 생성
• str을 함께 상속하여, Enum 멤버들이 JSON으로 변환될 때 "active"와 같은 일반 문자열로 처리되도록 함

Product 모델 정의

• name: str = Field(min_length=1, max_length=100)
  • name 필드는 문자열 타입이며, Field를 통해 1자 이상 100자 이하의 길이 제약조건을 부여
• description: Optional[str] = None
  • description 필드는 선택 사항으로, 값이 없을 수도 있음을 Optional로 명시
• price: float = Field(gt=0)
  • price 필드는 부동소수점 타입이며, Field의 gt(greater than) 옵션을 사용하여 값이 항상 0보다 커야 한다는 규칙을 적용
• status: ProductStatus = ProductStatus.ACTIVE
  • status 필드의 타입을 위에서 정의한 ProductStatus Enum으로 지정
  • 이를 통해 ProductStatus에 정의된 세 가지 값 외에는 받을 수 없도록 강제
  • 기본값을 ProductStatus.ACTIVE로 설정하여, status 값이 주어지지 않을 경우 자동으로 'active' 상태가 되도록 함

---

9. 권한 제어 상품 조회 API

문제

상품 조회 API를 만들어주세요
GET /products/{product_id}
1. product_id : >=1
2. user_role : UserRole.user
3. 존재하지 않으면 : PRODUCT_NOT~~
class ErrorResponse(BaseResponse):
    success : bool = False
    error : str
    error_code : int
    details : Optional[Dict] = None
4. 역할별로 product 정보 노출을 다르게 해주세요
5. 성공하면 SuccessResponse(message = "", data = product)

코드

from fastapi import FastAPI, HTTPException, Path, Header, Response, status
from pydantic import BaseModel, Field
from enum import Enum
from typing import Optional, Dict, Type, Any, Generic, TypeVar

app = FastAPI()

# --- Enum 및 기본 모델 정의 ---

class UserRole(str, Enum):
    USER = "user"
    ADMIN = "admin"

class ProductStatus(str, Enum):
    ACTIVE = "active"
    INACTIVE = "inactive"
    OUT_OF_STOCK = "out_of_stock"

# --- 공통 응답 모델 정의 ---
T = TypeVar('T')

class BaseResponse(BaseModel):
    success: bool
    
class ErrorResponse(BaseResponse):
    success: bool = False
    error: str
    error_code: int
    details: Optional[Dict[str, Any]] = None

class SuccessResponse(BaseResponse, Generic[T]):
    success: bool = True
    message: str = "성공"
    data: T

# --- 상품 도메인 모델 정의 ---

# DB에 저장되는 전체 정보 모델
class ProductDB(BaseModel):
    id: int
    name: str
    description: str
    price: float
    status: ProductStatus

# 일반 사용자에게 노출될 정보 모델
class ProductUserView(BaseModel):
    name: str
    price: float
    status: ProductStatus
    
# 관리자에게 노출될 정보 모델 (모든 정보)
class ProductAdminView(ProductDB):
    pass

# --- 인메모리 데이터베이스 ---

products_db: Dict[int, ProductDB] = {
    1: ProductDB(id=1, name="노트북", description="고성능 노트북입니다.", price=1500000, status=ProductStatus.ACTIVE),
    2: ProductDB(id=2, name="키보드", description="기계식 키보드입니다.", price=120000, status=ProductStatus.OUT_OF_STOCK),
    3: ProductDB(id=3, name="마우스", description="게이밍 마우스입니다.", price=50000, status=ProductStatus.INACTIVE),
}

# --- API 엔드포인트 구현 ---

@app.get("/products/{product_id}", response_model=SuccessResponse[Any])
def get_product_by_id(
    product_id: int = Path(..., ge=1, description="상품 ID는 1 이상이어야 합니다."),
    user_role: UserRole = Header(UserRole.USER, description="사용자 역할 (user 또는 admin)")
):
    product = products_db.get(product_id)

    # 3. 존재하지 않을 경우 ErrorResponse 반환
    if not product:
        error_content = ErrorResponse(
            error="PRODUCT_NOT_FOUND",
            error_code=404,
            details={"product_id": product_id}
        ).model_dump_json()
        return Response(content=error_content, status_code=status.HTTP_404_NOT_FOUND, media_type="application/json")

    # 4. 역할별로 다른 데이터 모델(View) 선택
    ProductView: Type[BaseModel]
    if user_role == UserRole.ADMIN:
        ProductView = ProductAdminView
    else: # user_role == UserRole.USER
        ProductView = ProductUserView

    # 선택된 View 모델로 데이터 변환
    product_data = ProductView.model_validate(product)

    # 5. 성공 시 SuccessResponse 반환
    return SuccessResponse(data=product_data)

공통 응답 모델 (BaseResponse, ErrorResponse, SuccessResponse)

• 목적: API의 모든 응답 형식을 일관되게 유지하기 위함
• ErrorResponse: 실패 응답을 위한 전용 모델. success는 항상 False이며, 에러 메시지, 코드 등 상세 정보를 포함
• SuccessResponse: 성공 응답을 위한 전용 모델. success는 항상 True이며, 제네릭(Generic[T])을 사용하여 data 필드에 다양한 타입의 데이터를 담을 수 있도록 설계

역할별 데이터 모델 (ProductUserView, ProductAdminView)

• 목적: 동일한 데이터라도 사용자의 역할에 따라 노출되는 정보의 범위를 다르게 제어하기 위함 (보안 및 정보 은닉)
• ProductUserView: 일반 사용자에게는 name, price, status 등 민감하지 않은 정보만 노출
• ProductAdminView: 관리자에게는 id, description을 포함한 모든 정보를 노출

API 엔드포인트 (GET /products/{product_id})

• product_id: int = Path(..., ge=1, ...)
  • Path를 사용하여 경로 매개변수 product_id가 1 이상(ge=1)이어야 한다는 제약조건을 추가
• user_role: UserRole = Header(UserRole.USER, ...)
  • HTTP 요청의 헤더(Header)에서 user-role 값을 읽어 user_role 변수에 할당
  • 헤더 값이 없으면 기본값으로 UserRole.USER를 사용
• 에러 처리 로직
  • products_db에서 상품을 찾지 못하면, ErrorResponse 모델을 사용하여 표준화된 에러 메시지를 생성하고 404 Not Found 상태 코드로 응답
• 역할 기반 분기 처리
  • user_role 값에 따라 ProductView 변수에 ProductAdminView 또는 ProductUserView 모델 클래스를 할당
  • ProductView.model_validate(product)를 통해 DB에서 가져온 전체 데이터를 해당 역할에 맞는 모델로 변환 (불필요한 필드는 자동으로 걸러짐)
• 성공 응답
  • 역할에 맞게 가공된 product_data를 SuccessResponse로 감싸서 반환

---

10. 파일 업로드 시스템 API

문제

streamlit 에서 파일을 업로드하면
fastapi를 통해서 서버에 파일이 저장되는 
어플리케이션을 만들어주세요

코드

FastAPI

import shutil
from pathlib import Path
from fastapi import FastAPI, UploadFile, File

app = FastAPI()

# 업로드된 파일을 저장할 디렉토리
UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True) # 서버 시작 시 디렉토리가 없으면 생성

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    # 저장될 파일의 전체 경로
    dest_path = UPLOAD_DIR / file.filename

    try:
        # 파일을 디스크에 저장
        with dest_path.open("wb") as buffer:
            shutil.copyfileobj(file.file, buffer)
    finally:
        # 파일 핸들러 닫기
        file.file.close()
        
    return {"filename": file.filename, "saved_path": str(dest_path)}

Streamlit

import streamlit as st
import requests

# FastAPI 백엔드 서버 주소
FASTAPI_URL = "http://127.0.0.1:8000/uploadfile/"

st.title("Streamlit에서 FastAPI로 파일 업로드")

# 파일 업로더 위젯
uploaded_file = st.file_uploader("업로드할 파일을 선택하세요.")

if uploaded_file is not None:
    # 업로드 버튼
    if st.button("서버로 파일 업로드"):
        with st.spinner("파일을 업로드 중입니다..."):
            try:
                # FastAPI 서버로 전송할 파일 데이터 준비
                # 'file'이라는 키는 FastAPI의 매개변수 이름(file: UploadFile)과 일치해야 함
                files = {"file": (uploaded_file.name, uploaded_file.getvalue(), uploaded_file.type)}
                
                # POST 요청 보내기
                response = requests.post(FASTAPI_URL, files=files)
                
                if response.status_code == 200:
                    st.success("파일이 성공적으로 업로드되었습니다.")
                    st.json(response.json())
                else:
                    st.error(f"파일 업로드 실패. 상태 코드: {response.status_code}")
                    st.text(response.text)

            except requests.exceptions.RequestException as e:
                st.error(f"서버에 연결할 수 없습니다: {e}")

FastAPI

• Path("uploads")
  • pathlib을 사용하여 파일 경로를 객체로 관리
  • UPLOAD_DIR.mkdir(exist_ok=True): 서버 실행 시 uploads 디렉토리가 없으면 자동으로 생성
• @app.post("/uploadfile/")
  • /uploadfile/ 경로로 오는 POST 요청을 처리하는 엔드포인트 정의
• file: UploadFile = File(...)
  • FastAPI의 내장 기능을 사용하여 HTTP 요청의 파일 부분을 UploadFile 객체로 받음
  • python-multipart 라이브러리가 이 기능을 위해 필요
• shutil.copyfileobj(file.file, buffer)
  • 업로드된 파일(file.file)의 내용을 서버의 디스크 파일(buffer)로 효율적으로 복사하여 저장

Streamlit

• st.file_uploader(...)
  • Streamlit에서 파일 선택 UI를 생성하는 위젯
  • 사용자가 파일을 선택하면 해당 파일 객체를 반환
• requests.post(FASTAPI_URL, files=files)
  • requests 라이브러리를 사용하여 FastAPI 서버에 POST 요청을 보냄
• files = {"file": ...}
  • 파일 데이터를 전송하기 위한 핵심 부분
  • 딕셔너리의 키("file")는 FastAPI 엔드포인트 함수의 매개변수 이름(file)과 반드시 일치해야 함
  • 값은 (파일명, 파일내용, 파일타입) 형식의 튜플로 구성
• st.spinner, st.success, st.error
  • 사용자에게 현재 진행 상황(업로드 중, 성공, 실패)을 시각적으로 보여주는 Streamlit의 UI 요소

---

12. 블로그 DB 관리 API

문제

FastAPI와 SQLModel을 사용하여 블로그 게시물을 관리하는 전체 CRUD API를 구현하세요.

1.  데이터베이스 모델 (Post):
      * id: int, 기본 키(PK), 자동 증가
      * title: str, 3자 이상 50자 이하
      * content: str
      * author: str, 2자 이상 20자 이하
      * created_at: datetime, DB에 추가될 때 시간 자동 생성

2.  입력/출력 모델 분리:
      * PostCreate: title, content, author 필드를 가짐 (게시물 생성용)
      * PostUpdate: title, content 필드를 가짐 (게시물 수정용, 모두 선택 사항)
      * PostRead: id, title, content, author, created_at 필드를 모두 가짐 (응답용)

3.  API 엔드포인트 구현:
      * POST /posts: 새 게시물 생성. (성공 시 201 Created)
      * GET /posts: 모든 게시물 목록 조회.
      * GET /posts/{post_id}: 특정 ID의 게시물 조회. (없으면 404 Not Found)
      * PUT /posts/{post_id}: 특정 ID의 게시물 수정. (title, content만 수정)
      * DELETE /posts/{post_id}: 특정 ID의 게시물 삭제. (성공 시 204 No Content)

4.  기타:
      * lifespan을 사용하여 앱 시작 시 DB와 테이블을 생성하세요.
      * Dependency Injection을 통해 DB 세션을 관리하세요.

코드

# 필요 라이브러리: pip install fastapi "uvicorn[standard]" sqlmodel
from contextlib import asynccontextmanager
from datetime import datetime
from typing import List, Optional
from fastapi import FastAPI, Depends, HTTPException, Response, status
from sqlmodel import Field, Session, SQLModel, create_engine, select

# --- DB 설정 ---
DATABASE_URL = "sqlite:///blog.db"
# 실제 서비스에서는 check_same_thread는 False로 두지 않는 것이 좋음 (예제용)
engine = create_engine(DATABASE_URL, echo=True, connect_args={"check_same_thread": False})

def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

# --- FastAPI Lifespan 설정 ---
@asynccontextmanager
async def lifespan(app: FastAPI):
    print("애플리케이션 시작: DB 테이블 생성")
    create_db_and_tables()
    yield
    print("애플리케이션 종료")

app = FastAPI(lifespan=lifespan)

def get_session():
    with Session(engine) as session:
        yield session

# --- 모델 정의 ---

# 2. 입력/출력 모델
class PostCreate(SQLModel):
    title: str = Field(min_length=3, max_length=50)
    content: str
    author: str = Field(min_length=2, max_length=20)

class PostUpdate(SQLModel):
    title: Optional[str] = Field(None, min_length=3, max_length=50)
    content: Optional[str] = None

# 1. 데이터베이스 모델
class Post(PostCreate, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    created_at: datetime = Field(default_factory=datetime.utcnow, nullable=False)

# 응답 모델 (ID와 생성 시간 포함)
class PostRead(PostCreate):
    id: int
    created_at: datetime


# --- API 엔드포인트 구현 ---

# 3-1. CREATE
@app.post("/posts", response_model=PostRead, status_code=status.HTTP_201_CREATED)
def create_post(post_create: PostCreate, session: Session = Depends(get_session)):
    db_post = Post.model_validate(post_create)
    session.add(db_post)
    session.commit()
    session.refresh(db_post)
    return db_post

# 3-2. READ (All)
@app.get("/posts", response_model=List[PostRead])
def read_posts(session: Session = Depends(get_session)):
    posts = session.exec(select(Post)).all()
    return posts

# 3-3. READ (One)
@app.get("/posts/{post_id}", response_model=PostRead)
def read_post(post_id: int, session: Session = Depends(get_session)):
    db_post = session.get(Post, post_id)
    if not db_post:
        raise HTTPException(status_code=404, detail="게시물을 찾을 수 없습니다.")
    return db_post

# 3-4. UPDATE
@app.put("/posts/{post_id}", response_model=PostRead)
def update_post(post_id: int, post_update: PostUpdate, session: Session = Depends(get_session)):
    db_post = session.get(Post, post_id)
    if not db_post:
        raise HTTPException(status_code=404, detail="게시물을 찾을 수 없습니다.")
    
    update_data = post_update.model_dump(exclude_unset=True)
    for key, value in update_data.items():
        setattr(db_post, key, value)
        
    session.add(db_post)
    session.commit()
    session.refresh(db_post)
    return db_post

# 3-5. DELETE
@app.delete("/posts/{post_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_post(post_id: int, session: Session = Depends(get_session)):
    db_post = session.get(Post, post_id)
    if not db_post:
        raise HTTPException(status_code=404, detail="게시물을 찾을 수 없습니다.")
    
    session.delete(db_post)
    session.commit()
    return Response(status_code=status.HTTP_204_NO_CONTENT)

모델링 (SQLModel 및 Pydantic)

• Post: DB 테이블과 직접 매핑되는 모델, id와 created_at 같이 서버에서 관리하는 필드를 포함
• PostCreate: 게시물 생성 시 클라이언트로부터 받을 데이터의 규칙을 정의 (id 등 불필요한 필드 제외)
• PostUpdate: 게시물 수정 시 받을 데이터 규칙 정의. 모든 필드를 Optional로 만들어 부분 수정이 가능하도록 함
• PostRead: 클라이언트에게 응답을 보낼 때 사용할 모델. DB에 저장된 모든 정보를 포함

DB 및 lifespan 설정

• lifespan: FastAPI 앱이 시작될 때 create_db_and_tables 함수를 실행하여 DB 테이블이 존재하지 않으면 자동으로 생성
• get_session: API 함수가 호출될 때마다 독립적인 DB 세션을 생성하고, 함수 실행이 끝나면 세션을 닫아주는 의존성 주입 함수

API 엔드포인트 설명 (CRUD)

• POST /posts (생성)
  • PostCreate 모델로 데이터를 받아 유효성 검사 후 DB에 저장
  • 성공 시 201 Created 상태 코드 반환
• GET /posts/{post_id} (조회)
  • session.get()으로 특정 ID의 게시물을 효율적으로 조회
  • 데이터가 없으면 404 Not Found 에러 발생
• PUT /posts/{post_id} (수정)
  • PostUpdate 모델로 수정할 데이터만 받음
  • model_dump(exclude_unset=True)를 사용하여 클라이언트가 보낸 필드만 업데이트
• DELETE /posts/{post_id} (삭제)
  • session.get()으로 삭제할 게시물을 먼저 조회
  • 게시물이 존재하면 session.delete()로 삭제 후 session.commit()으로 DB에 반영
  • 성공적으로 삭제되면, 클라이언트에게 별도의 본문 내용 없이 성공했음을 알리는 204 No Content 상태 코드를 반환

---

13. 가상 주식 DB 조회 API

문제

FastAPI, Streamlit, SQLModel을 사용하여 가상 주식 데이터를 저장하고 조회하는 웹 애플리케이션을 구현하세요.

1.  데이터베이스 모델 (StockData):
      * id: int, 기본 키(PK), 자동 증가
      * ticker: str, 주식 티커 (예: "하이닉스")
      * date: datetime, 날짜
      * close: float, 종가
      * volume: int, 거래량

2.  FastAPI 백엔드 (stock_api.py):
      * 앱 시작 시(lifespan) DB와 테이블을 생성하고, "하이닉스"와 "삼성전자"에 대한 가상의 30일치 주가 데이터를 자동으로 생성하여 DB에 삽입하세요.
      * GET /stocks/{ticker} 엔드포인트를 구현하세요.
          * 경로 매개변수로 받은 ticker에 해당하는 모든 주가 데이터를 DB에서 조회합니다.
          * 데이터는 날짜순으로 정렬하여 반환합니다.
          * 해당 ticker의 데이터가 없으면 빈 리스트를 반환합니다.

3.  Streamlit 프론트엔드 (dashboard.py):
      * 사용자가 주식 티커를 입력할 수 있는 텍스트 입력창을 제공하세요.
      * '차트 보기' 버튼을 누르면 FastAPI 백엔드에 데이터를 요청합니다.
      * API로부터 받은 데이터를 아래와 같이 시각화하세요.
          * 꺾은선 그래프: 시간의 흐름에 따른 종가(close) 변화를 표시합니다.
          * 데이터 표: 최근 데이터가 가장 위에 오도록 정렬하여 date, close, volume을 표시합니다.
      * 데이터가 없을 경우 "데이터를 찾을 수 없습니다." 메시지를 표시합니다.

코드

FastAPI

import random
from contextlib import asynccontextmanager
from datetime import datetime, timedelta
from typing import List
from fastapi import FastAPI, Depends
from sqlmodel import Field, Session, SQLModel, create_engine, select

# --- DB 설정 ---
DATABASE_URL = "sqlite:///stock.db"
engine = create_engine(DATABASE_URL, echo=False, connect_args={"check_same_thread": False})

def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

def get_session():
    with Session(engine) as session:
        yield session

# --- 모델 정의 ---
class StockData(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    ticker: str = Field(index=True)
    date: datetime
    close: float
    volume: int

# --- FastAPI Lifespan 및 앱 초기화 ---
@asynccontextmanager
async def lifespan(app: FastAPI):
    print("DB 테이블 생성 및 초기 데이터 삽입")
    create_db_and_tables()
    
    # DB에 데이터가 비어있을 때만 샘플 데이터 생성
    with Session(engine) as session:
        if not session.exec(select(StockData)).first():
            tickers = ["하이닉스", "삼성전자"]
            today = datetime.now()
            for ticker in tickers:
                price = random.uniform(80000, 150000)
                for i in range(30):
                    date = today - timedelta(days=i)
                    price *= (1 + random.uniform(-0.05, 0.05)) # 가격 변동 시뮬레이션
                    volume = random.randint(100, 1000)
                    sample_data = StockData(ticker=ticker, date=date, close=round(price, 2), volume=volume)
                    session.add(sample_data)
            session.commit()
    yield
    print("애플리케이션 종료")

app = FastAPI(lifespan=lifespan)

# --- API 엔드포인트 ---
@app.get("/stocks/{ticker}", response_model=List[StockData])
def get_stock_data(ticker: str, session: Session = Depends(get_session)):
    statement = select(StockData).where(StockData.ticker == ticker).order_by(StockData.date)
    results = session.exec(statement).all()
    return results

Streamlit

import streamlit as st
import requests
import pandas as pd

# FastAPI 백엔드 서버 주소
API_URL = "http://127.0.0.1:8000/stocks/"

st.title("가상 주식 차트")

# 사용자 입력
ticker = st.text_input("주식 티커를 입력하세요:", "하이닉스")

if st.button("차트 보기"):
    if not ticker:
        st.warning("티커를 입력해주세요.")
    else:
        with st.spinner(f"'{ticker}' 주가 데이터를 불러오는 중..."):
            try:
                # API 호출
                response = requests.get(API_URL + ticker)
                response.raise_for_status() # 200번대 상태 코드가 아니면 에러 발생
                
                data = response.json()

                if not data:
                    st.warning("해당 티커의 데이터를 찾을 수 없습니다.")
                else:
                    # Pandas DataFrame으로 변환
                    df = pd.DataFrame(data)
                    df['date'] = pd.to_datetime(df['date'])
                    df = df.set_index('date')
                    
                    # 차트 및 데이터 표시
                    st.subheader(f"{ticker} 주가")
                    st.line_chart(df['close'])
                    
                    st.subheader("최근 데이터")
                    st.dataframe(df[['close', 'volume']].sort_index(ascending=False))

            except requests.exceptions.RequestException as e:
                st.error(f"API 서버에 연결할 수 없습니다: {e}")

FastAPI

• lifespan
  • 앱 시작 시 DB 테이블을 생성하고, DB가 비어있을 경우에만 가상의 주가 데이터를 동적으로 생성하여 삽입
• GET /stocks/{ticker}
  • 경로 매개변수로 받은 ticker를 기준으로 DB에서 where 절을 이용해 데이터를 필터링
  • order_by로 날짜순 정렬 후, 조회된 모든 데이터를 리스트 형태로 반환

Streamlit

• requests.get
  • '차트 보기' 버튼 클릭 시 FastAPI 서버에 해당 티커의 데이터를 HTTP GET으로 요청
• pandas.DataFrame
  • API로부터 받은 JSON(리스트 형태) 데이터를 표(DataFrame) 형태로 변환
  • 데이터 처리(날짜 타입 변환, 인덱스 설정) 및 시각화를 용이하게 함
• st.line_chart
  • Pandas DataFrame의 특정 열(close)을 꺾은선 그래프로 시각화
• st.dataframe
  • DataFrame 데이터를 웹 페이지에 표 형태로 깔끔하게 출력. 최신 데이터가 위로 오도록 내림차순 정렬