옵저버 패턴과 FAST API를 활용한 도서 관리 시스템 예제

예제 내용: 도서 관리 API + Swagger UI로 문서 자동화 및 테스트 환경 구성

# GET / books: 조회
# POST / book: 등록
옵저버 패턴을 적용하여 새 도서가 추가되면 데이터 베이스 검색 모듈에 알림을 주도록 설계합니다.

---

초기 패키지 설정

pip install fastapi pydantic

---

Book 클래스 정의: pydantic

# pydantic 모델 정의
class Book(BaseModel):
    id:int
    title:str

class BookCreate(BaseModel):
    title : str

# 샘플 데이터
books = [
    Book(id=1, title="llm study"),
    Book(id=2, title="python study"),
]

---

옵저버 패턴 구현: Observer와 Subject

# 옵저버 패턴 구현
class Subject:
    def __init__(self) -> None:
        self._observer = []
    def add_observer(self, obs):
        self._observer.append(obs)
    def notify(self, message):
        for obs in self._observer:
            obs.update(message)

class Observer: # 추상화( 다양한 옵저버 클래스들을 커버 )
    def update(self, message):
        raise NotImplementedError
    
# 옵저버.. RAG 시스템의 검색 메소드
class SearchModule(Observer): # RAG
    def update(self, message):
        print(f"RAG 검색 모듈: {message}, 인덱스 업데이트")

# BookManager( Subject 상속 )
class BookManager(Subject):
    # 책 추가
    def add_book(self, book:BookCreate):
        new_id = max(b.id for b in books) + 1        
        new_book = Book(id=new_id, title=book.title)
        books.append(new_book)
        self.notify(f"새로운 도서 추가: {new_book.title}")
        return new_book

---

API의 엔드 포인트

book_manager = BookManager()
search_module = SearchModule()
book_manager.add_observer(search_module)

app = FastAPI()

# URL에 접속했을 때 바로 테스트 화면으로 전환
@app.get("/")
def intro():
    return RedirectResponse(url="/docs")

# 호출
@app.get("/books", response_model = List[Book] ,tags=["Books"]) # 결과 반환 값 설정
async def get_books():
    return books

@app.post("/book", response_model = Book, tags=["Books"]) # 결과 반환 값 설정
async def create_book(book:BookCreate):
    return book_manager.add_book(book)

---

실행 결과

create_book 실행 결과

 

get_books 실행 결과