FastAPI, Javascript: EventSource Example

EventSource

HTTP를 통해서 server-sent-events (SSEs)를 받기 위한 인터페이스

페이지 새로고침 없이 real-time(실시간)으로 서버에서 데이터를 받을 때 유용하게 쓸 수 있다.

 

아래는 오픈소스 g6의 데이터 베이스가 설치되는 과정에서 Python 코루틴을 활용하여

백엔드 FastAPI에서 프론트엔드 Javascript로 페이지 리로딩 없이 설치 정보를 전달하는 예제

 

 

@router.get("/process")
async def install_process(request: Request):
    
    async def install_event():
        db_connect = DBConnect()
        engine = db_connect.engine
        SessionLocal = db_connect.sessionLocal
        yield "데이터베이스 연결 완료"

        try:
            form: InstallFrom = form_cache.get("form")

            if form.reinstall:
                models.Base.metadata.drop_all(bind=engine)
                metadata = MetaData()
                metadata.reflect(bind=engine)
                table_names = metadata.tables.keys()
                for name in table_names:
                    if name.startswith(f"{form.db_table_prefix}write_"):
                        Table(name, metadata, autoload=True).drop(bind=engine)

                yield "기존 데이터베이스 테이블 삭제 완료"

            models.Base.metadata.create_all(bind=engine)
            yield "데이터베이스 테이블 생성 완료"

            with SessionLocal() as db:
                config_setup(db, form.admin_id, form.admin_email)
                admin_member_setup(db, form.admin_id, form.admin_name,
                                   form.admin_password, form.admin_email)
                content_setup(db)
                qa_setup(db)
                faq_master_setup(db)
                board_group_setup(db)
                board_setup(db)
                db.commit()
                yield "기본설정 정보 입력 완료"

            for board in default_boards:
                dynamic_create_write_table(board['bo_table'], create_table=True)
            yield "게시판 테이블 생성 완료"

            setup_data_directory()
            yield "데이터 경로 생성 완료"

            yield f"[success] 축하합니다. {default_version} 설치가 완료되었습니다."
        
        except Exception as e:
            os.remove(ENV_PATH)
            yield f"[error] 설치가 실패했습니다. {e}"

    return EventSourceResponse(install_event())

 

document.getElementById("install_success").style.display = "none";
document.getElementById("install_fail").style.display = "none";

setTimeout(() => {
    token = generate_token();
    const evtSource = new EventSource("{{ url_for('install_process') }}?token=" + token);
    evtSource.onmessage = function(event) {
        const data = event.data.trim();  // 공백 제거

        if (data.includes("[success]")) {
            document.getElementById("install_result").innerHTML = data;
            document.getElementById("install_success").style.display = "block";
            evtSource.close();
        } else if (data.includes("[error]")) {
            document.getElementById("install_result").innerHTML = data;
            document.getElementById("install_fail").style.display = "block";
            evtSource.close();
        } else {
            document.getElementById("install_status").innerHTML += "<li>" + data + "</li>"; // 메시지 출력
        }
    }
}, 1000);

 

Javascript에서 EventSource를 활용해서 FastAPI 서버와 연결

(FastAPI에서도 EventSourceResponse를 통해 해당 방식으로 연결 받을 준비 되어 있음)

 

yield로 구분되어져 있는 부분들들을 FastAPI에서 처리하여 전송

 

yield1

...

yield2

...

yield3

 

위와 같이 짜여진 코드에서,

EventSource를 통해 HTTP 연결이 살아있고,

코드가 순차적으로 실행되며 yield1, yield2, yield3을 순차적으로 전송 (Streaming Response)

 

 

** Javascript -> EventSource를 통해 연결을 하면, 서버로부터 데이터를 받기 위해 연결을 한다.

설명처럼 데이터를 '받기' 위해서

그렇다면, 데이터를 받은 후에 다시 request를 하는게 아니다.

다시 request를 하는게 아니라서 FastAPI의 제너레이터가 다시 호출되는게 아님.

그런데 어떻게 순차적으로 다음 yield를 만날때까지 로직이 실행될까?

 

-> FastAPI 에서 비동기 이벤트 루프를 관리하며 코루틴의 실행을 스케줄링

FastAPI는 내부적으로 비동기 이벤트 루프가

yield 이후에 해당 코루틴 함수를 자동적으로 '다시 호출'

 

'async def'로 정의된 비동기 함수를 실행할 때, 'await' 표현식이나  'yield'를 만나면 해당 지점에서 실행을 일시 중지하고, 이벤트 루프는 다른 비동기 작업을 실행할 수 있다.

비동기 제너레이터에서 'yield'된 다음 값이 준비 되면, 이벤트 루프는 해당 코루틴의 실행을 재개하여

다음 'yield'까지 실행하거나 함수가 완료될 때까지 계속 진행

 

말이 복잡하지만,

FastAPI는 비동기 이벤트 루프를 통해서 여러 yield로 짜여져 있는 코루틴을 자체 관리하여 끝까지 실행시켜 준다.

 

 

** yield를 사용한 코루틴의 좀 더 복잡한 예를 한번 살펴보자

def print_test1():
    print("111")

def print_test2():
    print("222")

async def my_generator():
    for i in range(3):
        yield f"Yielded {i}"
        print_test1()
        await asyncio.sleep(1)  # 비동기 작업을 시뮬레이션

@app.get("/manual")
async def manual_generator():
    gen = my_generator()  # 제너레이터 인스턴스 생성
    results = []
    async for item in gen:
        results.append(item)
        print_test2()
    return {"results": results}

 

/manual로 접근시, 

222

111

위와 같이 print가 되는 것이 3번 반복되는 결과를 볼 수 있다.

그리고 마지막에 

{"results":["Yielded 0","Yielded 1","Yielded 2"]}

 

위의 결과값을 뱉는 페이지 결과를 얻는다.

 

my_generator()로 인스턴스화 된 gen을 forloop를 돌 때,

 

forloop의 첫 번째 item에서 yield

my_generator 제너레이터를 빠져나오고,

yield된 값은 result에 append,

그리고 print_test2() 실행 (222 출력)  

 

 

forloop의 첫 번째 item 중 yield 이후 부분부터 다시 실행 print_test1() 실행 및 1초 기다림 (await asyncio.sleep(1))

forloop의 두 번째 item에서 yield

다시 my_generator 제너레이터를 빠져나오고,

yield된 값은 result에 append,

그리고 print_test2() 실행

 

yield를 통해서 이벤트 루프 내의 다른 이벤트에 순서를 양보할 수 있는 기능 구현 가능