[FastAPI] websocket을 사용한 통신

WebSocket란?

 

웹소켓(WebSocket)은 "프로토콜(Protocol, 통신 규약)"입니다. 웹 브라우저와 서버가 단일 TCP 연결을 통해 실시간으로 양방향 통신을 할 수 있게 해주는 프로토콜입니다. 즉, SSH나 HTTP와 같은 항렬에 있는 컴퓨터끼리 대화하는 규칙입니다.

 

TCP(Transmission Control Protocol)는 인터넷상에서 데이터를 정보를 안정적으로, 순서대로, 에러 없이 교환하기 위해 사용하는 전송 규약(프로토콜)입니다.

 

하지만 HTTP와 결정적인 차이가 하나 있습니다.

• HTTP: "나(클라이언트)만 말 걸 수 있어. 너(서버)는 대답만 해." (단방향)
• WebSocket: "우리 전화선 끊지 말고 계속 들고 있자. 할 말 있으면 언제든 먼저 해." (양방향)

 

 

핸드셰이크(Handshake)는 컴퓨터 네트워크에서 두 장치가 데이터를 본격적으로 주고받기 전에, "서로 연결할 준비가 되었는지 확인하고 규칙을 맞추는 과정"을 말합니다.

 

웹소켓(WebSocket)을 쓰는 이유는 딱 하나, "서버가 먼저 말을 걸어야 할 때(실시간성)" 때문입니다. 앱과 서버가 한번 연결(Handshake)하면, 선을 끊지 않고 계속 연결해 둡니다.

 

만약 웹소켓을 사용하지 않는다면 웹페이지에서 HTTP를 통해 필요한 정보가 도착했는지 계속 물어봐야 합니다. 하지만 웹소켓을 사용하면 서버에 정보가 드렁온 순간 바로 데이터를 push할 수 있습니다. 

 

---

FastAPI WebSocket 활용 

 

본격적으로 웹소켓 코드를 짜기 이전에 폴더 구조를 프로젝트에 맞게끔 변경해보겠습니다.  

 

이런 구조를 소프트웨어 공학에서는 "계층형 아키텍처 (Layered Architecture)" 라고 부릅니다. 유지보수, 재사용성, 협업의 효율성, 가독성 등을 고려해서 코드를 작성합니다. 

siren-backend/
├── .env                  # 🔐 [보안] Azure/Naver API 키, Oracle DB 비밀번호 등 민감 정보 저장소
├── requirements.txt      # 📦 [라이브러리] 프로젝트 실행에 필요한 패키지 목록 (pip install -r ...)
├── main.py               # 🚀 [진입점] 앱(App) 실행, CORS(보안 문지기) 설정, 라우터들을 하나로 합치는 곳
├── database.py           # 🗄️ [DB 설정] Oracle 데이터베이스와 연결을 맺고 세션을 관리하는 공장
├── routers/              # 🚦 [컨트롤러] "어떤 주소(URL)로 요청이 오면 누가 처리할지" 교통 정리하는 곳
│   ├── notification.py   # 📡 [웹소켓] 병원 프론트엔드와 24시간 연결된 전용 회선 (ws:// 담당)
│   ├── analysis.py       # 🎤 [핵심 기능] 앱에서 음성 파일을 받아 분석 명령을 내리고, 병원에 알림을 쏘는 곳
└── services/             # 👨‍🍳 [요리사] 실제 복잡한 기능(AI 분석, DB 저장, 지도 계산)을 수행하는 작업장
    ├── ws_manager.py     # 📒 [명부 관리] 현재 접속 중인 병원 리스트 관리 & "전체 방송(Broadcast)" 기능
    ├── audio_service.py  # 🧠 [AI 두뇌] Azure Whisper(STT)로 듣고, GPT(LLM)로 환자 상태를 요약하는 로직
    └── db_service.py     # 💾 [기록 담당] 분석된 환자 정보와 로그를 Oracle DB에 INSERT(저장)하는 로직

 

지난 포스팅에서 작성한 코드에 이어서 곧바로 웹소켓을 활용하는 코드를 작성해보겠습니다. 

---

ws_manager.py

 

ws_manager.py 코드는 FastAPI에서 웹소켓 연결을 관리하는 관리자(Manager) 역할을 하는 클래스입니다. 

from fastapi import WebSocket
from typing import List

 

가장 먼저 FastAPI가 제공하는 WebSocket 도구를 가져옵니다. 

class ConnectionManager:
    def __init__(self):
        # 접속한 병원(클라이언트) 리스트
        self.active_connections: List[WebSocket] = []

 

ConnectionManager라는 클래스를 만들고 def __init__(self) 코드는 클래스가 처음 생성될 때 실행되는 초기화 함수입니다.

async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)
        print(f"🏥 [WebSocket] 병원 접속! 현재 접속자: {len(self.active_connections)}")

 

self.active_connections: List[WebSocket] = []

→ 현재 서버에 접속해 있는 웹소켓 연결 객체들을 저장할 빈 리스트를 만듭니다. 여기에 접속한 병원들의 연결 정보가 차곡차곡 쌓입니다.

 

async def connect(...)

→  비동기 함수입니다. 웹소켓 연결은 네트워크 대기가 필요하므로 async를 사용합니다.

 

await websocket.accept()

→ 클라이언트(병원)가 연결을 요청했을 때, 서버가 "연결을 허가한다"는 응답을 보냅니다. 이 과정이 끝나야 통신이 가능해집니다.

def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)
        print(f"🔌 [WebSocket] 병원 접속 해제. 남은 접속자: {len(self.active_connections)}")

 

def disconnect(...)

→ 병원이 브라우저를 닫거나 네트워크가 끊겼을 때 호출되는 함수입니다.

async def broadcast(self, data: dict):
        # 접속한 모든 병원에게 JSON 데이터 전송
        print(f"📢 [WebSocket] 데이터 브로드캐스트 시작 (대상: {len(self.active_connections)}명)")
        for connection in self.active_connections:
            await connection.send_json(data)

 

async def broadcast(self, data: dict)

→ 딕셔너리(Dict) 형태의 데이터를 입력받아 모든 접속자에게 보냅니다.

 

await connection.send_json(data)

→ 파이썬의 딕셔너리 데이터를 웹 통신 표준 형식인 JSON으로 자동 변환하여 해당 연결로 전송합니다.

# 이 인스턴스를 다른 파일에서 import해서 사용합니다.
manager = ConnectionManager()

 

manager = ConnectionManager()

→ 위에서 정의한 클래스(설계도)를 바탕으로 실체(인스턴스)를 하나 만듭니다.

 

함수대신 클래스를 만든 이유는 데이터를 기억하며 작동하기 때문입니다. 

웹소켓 관리자 클래스 ConnectionManager는 누가 접속해 있는지를 계속 기억하고 있어야(저장해야) 합니다.

---

notification.py

 

notification.py 코드는 병원이 서버에 접속했을 때, 그 연결을 어떻게 유지할 것인가?를 정의하는 웹소켓 전용 라우터(Router)입니다.

from fastapi import APIRouter, WebSocket, WebSocketDisconnect
from services.ws_manager import manager

• APIRouter: main.py가 너무 복잡해지지 않게, 기능별로 파일(라우터)을 나눌 때 쓰는 도구입니다.
• WebSocket: 웹소켓 연결 객체의 타입(형태)을 알려주는 도구입니다.
• WebSocketDisconnect: "상대방이 연결을 끊었다!"라는 특수한 에러 상황을 감지하는 도구입니다.
• from services.ws_manager import manager: 방금 전 파일(ws_manager.py)에서 만든 manager (클래스)를 가져옵니다.

router = APIRouter(tags=["Notification"])

 

 

router = APIRouter(...): 이 파일(notification.py)을 하나의 "작은 앱"처럼 만들어줍니다. 나중에 main.py가 이 라우터를 가져다 씁니다.

# 병원 대시보드가 접속할 주소: ws://localhost:8000/ws/hospital
@router.websocket("/ws/hospital")

 

 

@router.websocket(...): 일반적인 인터넷 주소(http://)가 아니라, 웹소켓 주소(ws://)를 만드는 명령어입니다.

async def websocket_endpoint(websocket: WebSocket):

 

 

웹소켓은 연결을 유지하느라 대기하는 시간이 길어서 반드시 비동기(async) 함수여야 합니다. websocket에 접속한 정보가 담겨서 들어옵니다.

await manager.connect(websocket)

 

ws_manager.py에서 만든 클래스인 manager에서 선언한 함수 connect을 사용해 websocket을 인자로 보내고 이때 websocket.accept()(전화 받기)도 내부에서 실행됩니다.

try:
        while True:
            # 병원과의 연결을 유지합니다. (병원이 보내는 메시지는 딱히 처리 안 해도 됨)
            await websocket.receive_text()

 

await websocket.receive_text()

→ 서버가 일방적으로 프론트에게 push하기 때문에 그냥 연결을 끊지 말고 대기하라는 일시정지 역할을 하는 코드입니다. 

except WebSocketDisconnect:
        manager.disconnect(websocket)

 

브라우저 창을 닫거나 인터넷이 끊어지면 연결이 끊어집니다. 

---

analysis.py

지난번에 작성한 코드에서 웹소켓에 데이터를 보내는 코드를 추가합니다.

from services.ws_manager import manager

 

ws_manager.py에서 만든 인스턴스를 import 해줍니다.

await manager.broadcast({ ... })

 

ws_manager.py 안에서 정의했던 broadcast 함수를 여기서 호출합니다. 

이 줄이 실행되는 순간, ws_manager.py에 있는 for 반복문이 돌아갑니다. manager는 자기 리스트(self.active_connections)를 쫙 펼칩니다. 거기에 적혀있는 모든 병원(WebSocket 연결)에게 방금 만든 JSON 데이터(analyzed_json)를 send_json으로 쏘아 보냅니다.

 

 

routers/notification.py에서 await websocket.receive_text()를 하며 무한히 대기하고 있던 병원들이 analysis.py가 broadcast를 실행하면서 데이터가 웹소켓 관(ws://)을 타고 전송돼 대기하고 있던 병원의 브라우저(프론트엔드)에 도착합니다.

 

---

main.py

아래와 같이 main.py를 수정합니다.

# main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from dotenv import load_dotenv

# 만든 라우터 가져오기
from routers import analysis, navigation, notification

load_dotenv()

app = FastAPI(title="Siren", version="1.0.0")

# CORS 설정
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 라우터 등록 
app.include_router(analysis.router)
app.include_router(navigation.router)
app.include_router(notification.router)

@app.get("/")
def health_check():
    return {"status": "ok", "message": "Siren Whisper Server Running 🚑"}

 

 

 

CORS (Cross-Origin Resource Sharing)는 "웹 브라우저의 보안 문지기"입니다. 브라우저는 기본적으로 포트 번호가 다른 곳끼리의 대화를 차단합니다.

 

아래의 코드는 어떤 주소에서 오는 요청이든 다 허락한다는 의미입니다. 

 

실무(배포) 환경에서는 보안을 위해 allow_origins=["https://naver.com"] 처럼 믿을 수 있는 사이트 주소만 딱 적어주는 것이 정석입니다. 

app.add_middleware(
    CORSMiddleware,
    # ["*"]은 "모든 곳"을 의미합니다.
    # 즉, "어떤 주소(Origin)에서 오는 요청이든 다 허락해라" 라는 뜻입니다.
    allow_origins=["*"],
    
    # 쿠키나 인증 정보가 포함된 요청도 허락해라
    allow_credentials=True,
    
    # GET, POST, DELETE 등 모든 방식의 요청을 허락해라
    allow_methods=["*"],
    
    # 어떤 헤더 정보가 붙어있든 다 허락해라
    allow_headers=["*"],
)

 

그리고 앞서 만든 라우터 notification을 추가합니다.

app.include_router(notification.router)