[파이썬] FastAPI - SQLAlchemy와 CRUD (Depends, db.query)

이전 내용

[파이썬] FastAPI - ORM 연동하기(SQLAlchemy)

---

CRUD

CRUD는 다양한 소프트웨어 애플리케이션에서 데이터 관리를 위해 필수적인 개념으로, Create, Read, Update, Delete의 약자이다. 데이터베이스 작업의 기본적인 네 가지 기능을 나타낸다

• Create (생성): 새로운 데이터를 데이터베이스에 추가.
• Read (읽기): 저장된 데이터 조회.
• Update (업데이트): 기존 데이터 수정.
• Delete (삭제): 더 이상 필요하지 않은 데이터 제거.

이 네 가지 작업은 데이터베이스와 상호작용하는 모든 애플리케이션에서 일반적으로 수행되는 작업으로, 신뢰성과 효율적인 데이터 관리를 보장한다.

---

세션

세션은 사용자의 여러 요청을 하나의 연결로 관리하는 메커니즘이다. 웹 애플리케이션에서 세션은 사용자가 특정 웹 사이트를 탐색하는 동안 지속적으로 정보를 저장하고 사용할 수 있게 도와준다.

[세션의 역할]

• 사용자의 상태를 유지하여 로그인, 장바구니, 사용자 설정 등 관리.
• 서버에 데이터를 저장함으로써 클라이언트의 요청 간 정보를 연속성 제공.

[세션 관리 방법]

• 쿠키: 클라이언트에 저장된 작은 데이터 조각으로 세션 ID 저장.
• 서버 측 저장: 세션 데이터를 서버에 저장하고, 클라이언트는 세션 ID를 사용해 접근.

[CRUD와 세션의 연관성]

CURD 작업은 데이터베이스에서 가장 기본적인 동작으로, 이를 위해 세션이 매우 중요하게 작용한다.

CRUD와 세션은 웹 애플리케이션의 효율적인 작동을 위해 서로 보완적인 관계에 있다. 예를 들어, 사용자가 로그인하면 세션이 생성되고, 이후 CRUD 작업을 처리할 때 해당 세션 정보를 활용하여 사용자 맞춤형 경험을 제공한다.

---

FastAPI 에서의 CRUD 구현

세션은 SQLAlchemy에서 데이터베이스와의 모든 상호작용을 관리하며, 세션을 통해 실제 데이터베이스에 변화를 주거나 정보를 가져오는 작업, 즉 CRUD를 구현하기 위해서는 세션이 필요하다.

 

FastAPI에서 SQLAlchemy와 함께 작업하기 위해서는 Session 객체를 가져와야 하며, 이 객체를 이용해 데이터베이스와의 연결을 관리한다.

 

먼저 MySQL Workbench에 접속하여 테이블을 하나 생성한다.

CREATE DATABASE test_db;

 

◆ 데이터 입력 기능 구현하기

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from pydantic import BaseModel

# 데이터베이스 설정을 위한 문자열 정의
# 사용자 이름, 비밀 번호, 서버 주소, 데이터베이스 이름을 포함
DATABASE_URL = "mysql+pymysql://root:비밀번호@localhost/test_db"  # MySQL 데이터베이스 URL
engine = create_engine(DATABASE_URL) # SQLAlchemy 엔진 생성

# SQLAlchemy의 베이스 클래스를 생성
Base = declarative_base()

class User(Base):
    # 테이블 이름 설정
    __tablename__ = 'users'
    # 각 칼럼 정의
    id = Column(Integer, primary_key=True, autoincrement=True, comment="기본키")
    username = Column(String(50), unique=True, nullable=False, index=True, comment="사용자 이름")
    email = Column(String(150), unique=True, nullable=False, index=True, comment="이메일")
    
# pydantic 모델 정의
class UserCreate(BaseModel):
    username: str
    email: str
    
# 데이터베이스 세션 생성하고 관리하는 의존성 함수 정의
def get_db():
    db = Session(bind=engine)
    try:
        yield db
    finally:
        db.close()
        
# 데이터베이스 엔진 사용하여 데이터베이스에 테이블 생성
Base.metadata.create_all(bind=engine)

# FastAPI 애플리케이션 초기화
app = FastAPI()


@app.get("/")
def read_root():
    # 루트 경로에 접근 시 메시지 반환환
    return {"message": "Hello World"}

# 사용자 생성 POST API 엔드포인트 정의
@app.post("/users/")
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    # pydntic 모델 사용하여 데이터 유효성 검증, 새 User 인스턴스 생성
    new_user = User(username=user.username, email=user.email)
    db.add(new_user) # 생성된 User 인스턴스를 데이터베이스 세션에 추가
    db.commit() # 데이터베이스에 변경 사항 커밋
    db.refresh(new_user) # 데이터베이스로부터 새 User 인스턴스의 최신 정보 가져옴
    # 새로 생성된 사용자 정보 반환
    return {"id": new_user.id, "username": new_user.username, "email": new_user.email}

 

[주요 코드 설명]

1. 의존성 함수 정의

def get_db():
    db = Session(bind=engine)
    try:
        yield db
    finally:
        db.close()

Session객체를 생성하고, 요청 처리가 끝난 후 자동으로 정리하기 위해 yield를 사용한 get_db() 함수 정의

 

2. Pydantic 모델 정의

class UserCreate(BaseModel):
    username: str
    email: str

클라이언트로부터 받은 데이터의 유효성을 검증하기 위해 UserCreate의 Pydantic 모델을 정의한다. 이는 요청 데이터를 POST 방식으로 받기 위해 작성한 모델이다.

 

3. 의존성 주입 설정

@app.post("/users/")
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    # pydntic 모델 사용하여 데이터 유효성 검증, 새 User 인스턴스 생성
    new_user = User(username=user.username, email=user.email)
    db.add(new_user) # 생성된 User 인스턴스를 데이터베이스 세션에 추가
    db.commit() # 데이터베이스에 변경 사항 커밋
    db.refresh(new_user) # 데이터베이스로부터 새 User 인스턴스의 최신 정보 가져옴
    # 새로 생성된 사용자 정보 반환
    return {"id": new_user.id, "username": new_user.username, "email": new_user.email}

 

이 함수는 일반적으로 POST 요청의 엔드포인트로 사용된다. 사용자가 "/users/"와 같은 URL에 POST 요청을 보내면 이 함수가 호출된다.

• 데코레이터 @app.post("/users/"): 이 함수가 /users/ 경로로 들어오는 POST 요청을 처리하도록 정의. 이 경로는 새로운 사용자를 생성하는 목적이다.
• user: UserCreate: 이 매개변수는 클라이언트가 보낸 JSON 요청의 본문을 Pydantic 모델인 UserCreate로 변환. 이 모델은 사용자 생성에 필요한 속성(예: username, email)을 포함한다.
• db: Session = Depends(get_db): FastAPI의 의존성 시스템을 이용해 데이터베이스 세션을 주입합니다. get_db 함수가 데이터베이스 세션을 생성하고 반환하는 역할을 합니다.
• 사용자 인스턴스 생성: new_user = User(username=user.username, email=user.email)는 사용자 정보를 바탕으로 SQLAlchemy의 User 모델 인스턴스를 생성.
• 데이터베이스에 추가: db.add(new_user)는 생성된 사용자 인스턴스를 현재 데이터베이스 세션에 추가. 이 단계에서는 데이터베이스에 기록되지는 않지만, 세션이 이 인스턴스를 추적하게 된다.
• 변경 사항 커밋: db.commit()는 세션에 있는 모든 변경 사항을 데이터베이스에 실제로 저장. 이 과정에서 새로운 사용자가 데이터베이스에 기록된다.
• 최신 정보 갱신: db.refresh(new_user)는 데이터베이스로부터 새로 생성된 사용자 인스턴스의 최신 정보를 가져와 해당 인스턴스를 업데이트. 이 과정에서 자동으로 생성된 ID와 같은 정보가 갱신된다.
• 결과 반환: return {"id": new_user.id, "username": new_user.username, "email": new_user.email}는 새로 생성된 사용자에 대한 중요 정보를 JSON 형식으로 클라이언트에게 반환. 이는 요청한 클라이언트가 새 사용자의 ID와 기본 정보를 확인할 수 있게 한다.

테스트는 POST MAN으로 해본다.

포스트맨을 실행하고 http://127.0.0.1:8000/users/ 에 접속한 뒤 HTTP 메소드는 POST로 바꿔 JSON 형식으로 새로운 데이터를 입력 후 [SEND]를 누르면

{
    "username": "Hong", "email": "user123@example.com"
}

데이터 삽입이 잘 된 것을 확인할 수 있다.

 

◆ 데이터 조회(검색) 기능 구현하기: 특정 데이터

기존 코드에 데이터 조회 기능을 담당하는 코드를 추가한다.

# 데이터 조회 기능 추가 (READ)
@app.get("/users/{user_id}")
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first() # 사용자 ID로 데이터베이스에서 사용자 조회
    if db_user is None:
        return {"error": "User not found"} # 사용자가 존재하지 않을 경우 에러 메시지 반환
    return {"id": db_user.id, "username": db_user.username, "email": db_user.email} # 사용자 정보 반환

• 사용자 ID 등록: URL 경로에서 전달된 user_id 매개변수를 이용하여 요청된 사용자의 ID를 가져온다.
• 데이터베이스 세션 주입: 의존성 주입을 통해 데이터베이스 세션이 제공되며, 이를 통해 데이터베이스와 상호작용할 수 있게 된다.
• 사용자 조회: SQLAlchemy를 사용하여 User 테이블에서 입력받은 ID와 일치하는 사용자를 검색한다. 만약 해당 ID의 사용자가 존재하지 않을 경우, None이 반환된다.
• 에러 처리: 조회 결과가 None일 경우, "User not found"라는 메시지를 포함한 JSON 응답을 반환하여 사용자가 존재하지 않음을 클라이언트에게 알린다.
• 정보 반환: 사용자가 존재할 경우, 해당 사용자의 ID, 사용자 이름, 이메일 정보를 포함하는 JSON 형식의 응답을 반환한다.

역시 POST MAN에서 테스트를 해보면

서버를 실행하고 http://127.0.0.1:8000/users/3 (특정 id 조회를 위해 몇 개의 데이터를 더 입력해주었다.) 에 접속한 뒤, GET 메소드에서 [SEND]를 누르면 데이터가 잘 조회되는 것을 확인할 수 있다.

 

◆ 데이터 조회(검색) 기능 구현하기: 전체 데이터

# 전체 사용자 데이터 조회 기능 추가 (READ ALL)
@app.get("/users/")
def read_all_users(db: Session = Depends(get_db)):
    # 데이터베이스에서 모든 사용자 조회
    users = db.query(User).all()
    
    # 사용자 정보를 리스트 형태로 반환
    return [{"id": user.id, "username": user.username, "email": user.email} for user in users]

read_all_users 함수는 사용자가 /users/ 경로로 GET 요청을 보낼 때 호출된다. 데이터베이스에서 모든 사용자를 조회한 후, 각 사용자의 정보를 매핑하여 JSON 형식으로 클라이언트에게 반환한다. 이를 통해 클라이언트는 시스템에 등록된 모든 사용자의 정보를 쉽게 확인할 수 있다.

 

◆ 데이터 수정 기능 구현하기

from typing import Optional

# 데이터 수정 기능 추가 (UPDATE)
class UserUpdate(BaseModel):
    username: Optional[str] = None
    email: Optional[str] = None
    
@app.put("/users/{user_id}")
def update_user(user_id: int, user: UserUpdate, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if db_user is None:
        return {"error": "User not found"} # 사용자가 존재하지 않을 경우 에러 메시지 반환
    
    # 사용자 정보 업데이트
    if user.username is not None:
        db_user.username = user.username
    if user.email is not None:
        db_user.email = user.email
    db.commit() # 데이터베이스에 변경 사항 커밋
    db.refresh(db_user) # 데이터베이스로부터 업데이트된 사용자 정보 가져옴
    # 업데이트된 사용자 정보 반환
    return {"id": db_user.id, "username": db_user.username, "email": db_user.email}

 

[주요 코드 설명]

from typing import Optional:

• Optional은 Python의 타입 힌팅 기능 중 하나로, 특정 변수 또는 매개변수가 None이 될 수 있음을 나타낸다.
• Optional[str]는 문자열(str) 타입의 값을 가질 수 있으며, 값이 없을 경우 None일 수 있음을 의미.
• 이 기능은 Pydantic 모델을 정의할 때 유용하며, 데이터 유효성 검사와 직렬화 과정에서 사용.

UserUpdate 클래스:

• UserUpdate 클래스는 BaseModel을 상속받아 수정할 수 있는 사용자 정보 정의. 이 클래스 내의 두 필드(username과 email)는 모두 선택적이다.

PUT 요청 처리:

• @app.put("/users/{user_id}") 데코레이터는 특정 사용자 ID에 대한 PUT 요청을 처리한다. 사용자 ID를 URL 경로에서 확인할 수 있다.
• PUT 데코레이터는 해당 함수가 HTTP PUT 메소드 요청을 처리하도록 지시한다.

update_user 함수:

• update_user 함수는 사용자가 제공한 사용자 ID를 사용하여 데이터베이스에서 해당 사용자를 조회하며, 만약 사용자가 존재하지 않으면, 에러 메시지를 반환한다.
• user_id: int : 경로에서 추출된 사용자 식별자로, 데이터베이스의 해당 사용자를 찾는 데 사용
• user: UserUpdate: 요청 바디에서 받은 UserUpdate 모델의 인스턴스로, 이 인스턴스는 수정하고자 하는 사용자의 새로운 데이터를 포함한다.
• db: Session = Depends(get_db): Depends를 통해 의존성 주입이 이뤄지며, get_db() 함수로부터 Session 객체를 가져온다. 이 세션은 데이터베이스와의 모든 상호작용을 처리한다.
• db.query(User).filter(User.id == user_id).first(): 데이터베이스 쿼리를 통해 주어진 user_id를 가진 User 객체 검색. first() 메소드는 조건에 맞는 첫 번째 객체를 반환하거나 없을 경우 None 반환
• 사용자의 username이나 email 값이 제공되었을 경우(None이 아닐 경우)에만 해당 값을 업데이트한다.
• 데이터베이스에 변경 사항을 반영하기 위해 db.commit() 메서드를 호출하고, 이후 db.refresh(db_user)로 업데이트된 정보를 가져온다.
• 최종적으로 업데이트된 사용자 정보를 JSON 형식으로 클라이언트에게 반환한다.

POST MAN을 통해 테스트를 해보면, id 2인 "Kim"의 이메일 수정이 성공적으로 실행되었다.

◆ 데이터 삭제 기능 구현하기

# 데이터 삭제 기능 추가 (DELETE)
@app.delete("/users/{user_id}")
def delete_user(user_id: int, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if db_user is None:
        return {"error": "User not found"} # 사용자가 존재하지 않을 경우 에러 메시지 반환
    
    # 사용자 삭제
    db.delete(db_user)
    db.commit()
    return {"message": "User deleted successfully"} # 사용자 삭제 성공 메시지 반환

[주요 코드]

• @app.delete("/users/{user_id}"): 이 데코레이터는 해당 함수가 user_id 경로에 포함하는 DELETE 메소드 엔드포인트임을 FastAPI에 알린다.
• delete_user() 함수: user_id와 데이터베이스 Session 객체를 인자로 받으며, Session 객체는 Depends(get_db) 의존성을 통해 제공된다.
• 해당 함수는 주어진 user_id로 데이터베이스에서 사용자를 검색하여 사용자를 찾지 못하면 에러 메시지를, 사용자를 찾으면 데이터베이스 세션의 delete() 메소드를 사용하여 사용자를 삭제한 다음 트랜잭션을 db.commit()으로 커밋하고, 사용자가 삭제되었다는 메시지를 반환한다.

역시 POST MAN을 통해 테스트를 해보면

2번 사용자를 DELETE 메소드를 통해 [SEND] 버튼을 누르면 사용자가 성공적으로 삭제되었다는 메시지가 뜨고,

전체 사용자를 조회해보면 이전의 3명에서 2명만 남아있는 것이 확인된다.

---

Depends 의존성 주입

Depends는 FastAPI에서 의존성 주입(Dependency Injection)을 구현하기 위해 사용되는 매우 유용한 기능이다. 이를 통해 특정 함수나 클래스의 인스턴스를 다른 함수에 자동으로 주입할 수 있으며, 코드의 재사용성과 관리성을 높이는 데 도움이 된다.

 

[Depends 의존성 주입의 개념]

• 의존성 주입: 의존성 주입은 특정 함수가 실행되기 위해 필요한 의존 객체를 외부에서 주입받는 방식이다. 이를 통해 코드의 의존성을 줄일 수 있으며, 테스트와 유지보수가 용이해진다.
• FastAPI에서의 활용: FastAPI는 이러한 의존성 주입 기술을 사용하여 데이터베이스 세션, 인증 정보, 기타 서비스 객체를 쉽게 관리할 수 있도록 돕는다.

[기본 사용법]

1. 의존성 함수 정의:
먼저, 의존성으로 사용할 함수를 정의한다. 예: 데이터베이스 세션을 반환하는 get_db() 함수

from sqlalchemy.orm import Session
from fastapi import Depends

def get_db():
    db = SessionLocal()  # 데이터베이스 세션 생성
    try:
        yield db  # 세션을 반환
    finally:
        db.close()  # 세션 종료

 

2. 사용할 엔드포인트에서 의존성 주입:
FastAPI의 라우터 함수에서 Depends를 사용하여 의존성을 주입 받는다.

@app.post("/users/")
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    # db는 get_db() 함수에서 생성된 데이터베이스 세션
    new_user = User(**user.dict())
    db.add(new_user)
    db.commit()
    return new_user

- Depends(get_db):
이 부분은 create_user 함수에서 데이터베이스 세션이 필요하다는 것을 의미한다. Depends(get_db)를 사용하여 get_db 함수가 의존성을 제공하도록 한다. 요청이 들어오면 FastAPI는 자동으로 get_db를 호출하고 반환된 데이터베이스 세션을 db 매개변수에 주입한다.

 

- Yield 사용:
get_db 함수에서는 yield 문을 사용하여 세션을 반환한다. 이는 FastAPI가 주입된 후, 함수 실행이 끝나면 제공된 세션을 닫도록 보장한다.

※ yield 문
yield 문은 Python에서 제너레이터(generator)를 생성하는 데 사용되는 키워드.
제너레이터는 이터레이터(Iterator)의 일종으로, 일반적인 함수와는 다르게 여러 값들을 순차적으로 반환할 수 있는 특성을 가지고 있다.

※ 제너레이터
제너레이터는 이터레이션(Iteration)을 지원하는 특수한 종류의 함수로, 함수를 호출할 때 실행이 완료되지 않고 중간에 일시 중지할 수 있고. 이때 yield 문을 사용하여 값을 반환하게 된다.
일반 함수는 return 문을 사용하여 값을 반환하고 함수 실행이 종료되지만, 제너레이터는 yield 문에서 실행이 중단되고, 이후 다시 호출될 때 이전 상태에서 계속 실행할 수 있다.

def count_up_to(max):
    count = 1
    while count <= max:
        yield count  # 현재 count 값을 반환하고, 함수 상태를 유지
        count += 1  # count 값을 증가시킴

▶ count_up_to 함수는 주어진 최대값까지의 숫자를 하나씩 반환하는 제너레이터이다.

---

db.query() 문법

SQLAlchemy의 db.query() 메서드는 데이터베이스에서 객체를 조회하기 위해 사용되는 유용한 기능입니다. 이 메서드는 다양한 쿼리를 작성하는 데 활용될 수 있으며, 여러 가지 문법이 있습니다. 아래에 db.query()의 다양한 사용 예시를 정리하였습니다.

1. 기본 조회
특정 테이블의 모든 레코드 조회: all() 메서드는 해당 테이블의 모든 레코드를 리스트 형태로 반환.

all_users = db.query(User).all()

 

2. 필터링
조건을 추가하여 특정 레코드 조회: filter() 메서드로 조건을 추가하고, 해당 조건을 만족하는 첫 번째 레코드를 반환. 사용자가 존재하지 않으면 None을 반환

user = db.query(User).filter(User.id == 1).first()

여러 조건을 조합하여 조회: 여러 filter() 조건을 콤마로 나열하여 복수의 조건을 결합 가능.

users = db.query(User).filter(User.age > 18, User.active == True).all()

 

3. 정렬 및 제한
정렬: order_by() 메서드를 사용하여 결과를 정렬 가능. (기본값: 오름차순)

sorted_users = db.query(User).order_by(User.username).all()

행 제한: limit() 메서드를 사용하여 반환할 레코드 수를 제한 가능

limited_users = db.query(User).limit(5).all()

4. 특정 열 선택
특정 열만 조회: 특정 열만 선택하여 조회 가능. 이 경우 반환 결과는 튜플의 리스트

usernames = db.query(User.username).all()

5. 그룹화
그룹화 및 집계 함수 사용: func를 사용하여 SQL의 집계 함수(예: count, sum, avg)를 이용 가능. scalar() 메서드는 단일 값을 반환

from sqlalchemy import func

user_count = db.query(func.count(User.id)).scalar()

6. 서브쿼리
서브쿼리 사용: 서브쿼리하여 결과를 다시 조회할 수 있으며, 보다 복잡한 쿼리 가능

from sqlalchemy.orm import aliased

subquery = db.query(User).filter(User.active == True).subquery()
active_users = db.query(subquery).all()

7. 조인
테이블 간 조인: join() 메서드를 사용하여 다른 테이블과 조인 수행 가능

users_with_posts = db.query(User).join(Post).filter(Post.user_id == User.id).all()

---

[참고]

가장 빠른 풀스택을 위한 플라스크 & FastAPI

 

---

다음 내용

[FastAPI] 인증과 세션