Firestore CRUD & Deep Clone 매니저 서버의 모든 것

이 섹션에서는 FastAPI를 기반으로 Firestore의 모든 관리 기능을 통합한 전체 소스 코드를 정리한다. 각 API는 독립적인 역할을 수행하며, 모든 상태 변화는 WebSocket을 통해 클라이언트에 즉시 반영된다.

🏗️ 전체 소스 코드 구조 (main.py)

Python

import time
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
from google.cloud import firestore

app = FastAPI()

# CORS 설정: 프론트엔드 연동용
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])

def get_db(project_id: str):
    return firestore.Client(project=project_id)

# ----------------------------------------------------------------
# 1. READ: 실시간 리스트 조회 (WebSocket)
# ----------------------------------------------------------------
@app.websocket("/{project}/{collection}/list")
async def websocket_list(websocket: WebSocket, project: str, collection: str):
    await websocket.accept()
    db = get_db(project)
    
    # 데이터 변경 시 클라이언트에 전송할 콜백
    def on_snapshot(col_snapshot, changes, read_time):
        import asyncio
        docs = []
        for doc in col_snapshot:
            d = doc.to_dict()
            d["id"] = doc.id
            docs.append(d)
        # 스레드 안전하게 소켓 전송
        asyncio.run_coroutine_threadsafe(websocket.send_json(docs), asyncio.get_event_loop())

    query = db.collection(collection).order_by("updated", direction=firestore.Query.DESCENDING)
    watcher = query.on_snapshot(on_snapshot)

    try:
        while True: await websocket.receive_text()
    except WebSocketDisconnect:
        watcher.unsubscribe()

# ----------------------------------------------------------------
# 2. CREATE: 문서 추가
# ----------------------------------------------------------------
@app.post("/{project}/{collection}/add")
async def add_document(project: str, collection: str):
    db = get_db(project)
    new_data = {
        "title": "New Document",
        "updated": firestore.SERVER_TIMESTAMP,
        "slug": str(int(time.time()))
    }
    db.collection(collection).add(new_data)
    return {"status": "success"}

# ----------------------------------------------------------------
# 3. UPDATE: 문서 수정 (PATCH)
# ----------------------------------------------------------------
@app.patch("/{project}/{collection}/{docid}/update")
async def update_document(project: str, collection: str, docid: str, data: dict):
    db = get_db(project)
    # 데이터 가공 및 타임스탬프 갱신
    data["updated"] = firestore.SERVER_TIMESTAMP
    db.collection(collection).document(docid).update(data)
    return {"status": "success"}

# ----------------------------------------------------------------
# 4. DELETE: 문서 삭제
# ----------------------------------------------------------------
@app.delete("/{project}/{collection}/{docid}/delete")
async def delete_document(project: str, collection: str, docid: str):
    db = get_db(project)
    db.collection(collection).document(docid).delete()
    return {"status": "success"}

# ----------------------------------------------------------------
# 5. CLONE: Deep Clone (핵심 기능)
# ----------------------------------------------------------------
@app.post("/{project}/{collection}/{docid}/clone")
async def clone_document(project: str, collection: str, docid: str):
    db = get_db(project)
    source_ref = db.collection(collection).document(docid)
    source_doc = source_ref.get()
    
    if not source_doc.exists: return {"status": "error"}

    # 부모 문서 복제
    data = dict(source_doc.to_dict())
    data["title"] += " (Clone)"
    data["updated"] = firestore.SERVER_TIMESTAMP
    _, new_doc_ref = db.collection(collection).add(data)

    # 서브 컬렉션까지 몽땅 복사
    for sub_col in source_ref.collections():
        for doc in sub_col.stream():
            new_doc_ref.collection(sub_col.id).document(doc.id).set(doc.to_dict())

    return {"status": "success", "new_id": new_doc_ref.id}

💡 정리하며: 프로젝트의 의의

이 코드는 단순한 API 서버를 넘어, Firestore의 가장 가려운 곳인 **'계층 구조 복제'**를 해결했다는 데 의의가 있다.

• 확장성: get_db 함수 덕분에 수십 개의 프로젝트를 하나의 서버에서 관리할 수 있다.

• 사용성: WebSocket 연동을 통해 관리자가 작업 결과를 0.1초 만에 시각적으로 확인할 수 있다.

• 견고함: 동기 처리 방식을 택해 대량의 문서 복제 시 발생할 수 있는 비동기 충돌을 원천 차단했다.