본문 바로가기
Web 개발/FAST API (인프런 강의 내용)

5 실습3 ORM JOIN

by yororing 2024. 10. 16.

앞 단계 참조 링크:

00 개요

  • Docker MySQL에 접속하여 실제 user 테이블 생성한 후 user 데이터 (사용자 계정: user_id ("admin"), password ("password") 생성) 생성 후 user 테이블의 user 데이터를 todo 테이블 데이터와 연결 완료 상태
  •  SQLAlchemy의 ORM을 이용해 user 테이블과 todo 테이블을 JOIN하는 실습 진행
  • 앞으로 user를 조회할 때마다 그 user가 생성한 todo들을 전부 다 함께 조회하여 사용할 것

 

01 User 클래스에 todos 속성 연결하기

1. orm.py에 다음과 같이 작성

  • relationship을 이용해 todo와 user 연결할 것
  • from sqlalchemy.orm import relationship 추가 
  • relationship("연결할_클래스", lazy="joined") 사용 
# /c/Users/관리자/Desktop/projects/todos/src/database/orm.py 내용
from sqlalchemy import Boolean, Column, Integer, String, ForeignKey
from sqlalchemy.orm import declarative_base, relationship			# relationship 추가
from src.schema.request import CreateToDoRequest

Base = declarative_base()

# ToDo 클래스 모델링 한 것
class ToDo(Base):
    __tablename__ = 'todo'
    
    id = Column(Integer, primary_key=True, index=True)
    contents = Column(String(256), nullable=False)
    is_done = Column(Boolean, nullable=False)
    user_id = Column(Integer, ForeignKey("user.id"))
    
    def __repr__(self):
        return f"ToDo(id={self.id}, contents={self.contents}, is_done={self.is_done})"
    
    @classmethod
    def create(cls, request: CreateToDoRequest) -> "ToDo":
        return cls(
            # id값은 DB에 의해 자동으로 결정 되서 별도로 지정안해줌
            contents=request.contents,
            is_done=request.is_done,
        )
        
    def done(self) -> "ToDo":    # ToDo의 is_done 값을 True로 변경 후 ToDo 반환
        self.is_done = True
        return self
    
    def undone(self) -> "ToDo":  # ToDo의 is_done 값을 False로 변경 후 ToDo 반환
        self.is_done = False
        return self

# User 클래스 모델링 한 것
class User(Base):
    __tablename__ = 'user'
    
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(256), unique=True, nullable=False) # unique=True는 내가 넣은 것임
    password = Column(String(256), nullable=False)
    todos = relationship("ToDo", lazy="joined")			# 추가
  • 설명:
    • from sqlalchemy.orm import relationship 추가
    • todos = relationship("ToDo", lazy="joined") 추가
      • "ToDo"는 class User과 연결할 클래스명임
      • lazy라는 속성을 joined라고 줌
      • 이 todos 속성은, 실제로 어떤 Column이 생성되는 것이 아니라 가상의 relationship이 생성되는 것
      • ORM을 통해 user라는 테이블을 조회했을 때 user.todos라고 하면 이 user가 처음에 조회되는 시점에 todos를 함께 JOIN을 해와서 user.todos 했을 때 이 "ToDo" 클래스/테이블의 데이터를 한 번에 우리가 사용할 수 있게 됨 

2. Python Console에서 다음을 실행

>>> from src.database.connection import SessionFactory
>>> from src.database.orm import User
>>> from sqlalchemy import select
>>> session = SessionFactory()
>>> session
<sqlalchemy.orm.session.Session object at 0x0000016331D18E90>

>>> user = session.scalar(select(User))
2024-10-16 22:57:38,607 INFO sqlalchemy.engine.Engine SELECT DATABASE()
2024-10-16 22:57:38,608 INFO sqlalchemy.engine.Engine [raw sql] {}
2024-10-16 22:57:38,613 INFO sqlalchemy.engine.Engine SELECT @@sql_mode
2024-10-16 22:57:38,614 INFO sqlalchemy.engine.Engine [raw sql] {}
2024-10-16 22:57:38,616 INFO sqlalchemy.engine.Engine SELECT @@lower_case_table_names
2024-10-16 22:57:38,616 INFO sqlalchemy.engine.Engine [raw sql] {}
2024-10-16 22:57:38,620 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2024-10-16 22:57:38,627 INFO sqlalchemy.engine.Engine SELECT user.id, user.username, user.password, todo_1.id AS id_1, todo_1.contents, todo_1.is_done, todo_1.user_id
FROM user LEFT OUTER JOIN todo AS todo_1 ON user.id = todo_1.user_id
2024-10-16 22:57:38,628 INFO sqlalchemy.engine.Engine [generated in 0.00100s] {}
  • src.database.connection에서 SessionFactory import
  • src.database.orm에서 User 클래스 import
  • sqlalchemy에서 select import
  • session이라는 변수에 SessionFactory() 객체 생성
  • user라는 변수에 session에 scalar를 이용하여 그 안에 select(User) 하여 User() 클래스 생성
  • 엔터 누를 시 쿼리가 쭉 생성됨
    • SELECT user.id, user.username, user.password, todo_1.id AS id_1, todo_1.contents, todo_1.is_done, todo_1.user_id FROM user LEFT OUTER JOIN todo AS todo_1 ON user.id = todo_1.user_id
    • 위 쿼리문: todo 테이블이 user 테이블에 LEFT OUTER JOIN 되어있음
    • 즉, user 테이블과 todo 테이블이 JOIN되어서 조회가 될 것
    • 이로 인해 가능한 것: user.todos 를 Python Console에 입력 시 앞서 데이터를 연결하였던 3개의 todo가 한 번에 다음과 같이 조회될 것
>>> user.todos
[ToDo(id=1, contents=FastAPI Section 0, is_done=True), ToDo(id=2, contents=FastAPI Section 2, is_done=True), ToDo(id=3, contents=FastAPI Section 3, is_done=True)]

 

  • 이것은 우리가 joined를 relationship의 속성으로 지정해줬기 때문에 우리가 항상 user.todo를 조회할 때 이 ToDo 데이터를 한 번에 사용할 수 있게 됨 

 

이때까지의 코드들: 9개

  • /c/Users/관리자/Desktop/projects/todos/src/main.py
  • /c/Users/관리자/Desktop/projects/todos/src/api/todo.py
  • /c/Users/관리자/Desktop/projects/todos/src/database/orm.py
  • /c/Users/관리자/Desktop/projects/todos/src/database/connection.py
  • /c/Users/관리자/Desktop/projects/todos/src/database/repository.py
  • /c/Users/관리자/Desktop/projects/todos/src/schema/request.py
  • /c/Users/관리자/Desktop/projects/todos/src/schema/response.py
  • /c/Users/관리자/Desktop/projects/todos/src/tests/test_main.py
  • /c/Users/관리자/Desktop/projects/todos/src/tests/conftest.py

 

이때까지의 디렉토리

TODOS

├── Include

├── Lib

├── Scripts

├── src

│     ├── api

│     │      └── todo.py

│     ├── database

│     │      ├── connection.py

│     │      ├── orm.py

│     │      └── repository.py

│     ├── schema

│     │      ├── request.py

│     │      └── response.py

│     ├── tests

│     │      ├── __init__.py

│     │      ├──  conftest.py

│     │      └── test_main.py

│      └──  main.py

└── pyvenv.cfg

# /c/Users/관리자/Desktop/projects/todos/src/main.py 내용
from fastapi import FastAPI
from src.api import todo

app = FastAPI()
app.include_router(todo.router)

# 첫 화면 API
@app.get("/")
def health_check_handler():
    return {"ping": "pong"}
# /c/Users/관리자/Desktop/projects/todos/src/api/todo.py 내용
from fastapi import Body, HTTPException, Depends, APIRouter
from typing import List	

from src.database.repository import ToDoRepository, ToDo
from src.schema.response import ToDoSchema, ToDoListSchema
from src.schema.request import CreateToDoRequest

router = APIRouter(prefix='/todos')


# GET Method 사용하여 전체 조회 API
@router.get("", status_code=200)
def get_todos_handler(
    order: str | None = None,
    todo_repo: ToDoRepository = Depends(),
) -> ToDoListSchema:
    todos: List[ToDo] = todo_repo.get_todos()
    if order and order == "DESC":
        return ToDoListSchema(
            todos=[ToDoSchema.from_orm(todo) for todo in todos[::-1]]
        )
    return ToDoListSchema(
        todos=[ToDoSchema.from_orm(todo) for todo in todos]
    )

# GET Method 사용하여 단일 조회 API
@router.get("/{todo_id}", status_code=200)
def get_todo_handler(
    todo_id: int,
    todo_repo: ToDoRepository = Depends(),
    ) -> ToDoSchema:
    todo:ToDo | None = todo_repo.get_todo_by_todo_id(todo_id = todo_id)
    if todo:
        return ToDoSchema.from_orm(todo)
    raise HTTPException(status_code=404, detail="ToDo Not Found")

# POST Medthod 사용하여 todo 생성 API
@router.post("", status_code=201)
def create_todo_handler(
    request: CreateToDoRequest,
    todo_repo: ToDoRepository = Depends(),
) -> ToDoSchema:
    todo: ToDo = ToDo.create(request=request)          # id=None
    todo: ToDo = todo_repo.create_todo(todo=todo) # id=int
    return ToDoSchema.from_orm(todo)

# PATCH Method 사용하여 is_done 값 수정 API
@router.patch("/{todo_id}", status_code=200)
def update_todo_handler(
    todo_id: int,
    is_done: bool = Body(..., embed=True),
    todo_repo: ToDoRepository = Depends(),
    ):
    todo:ToDo | None = todo_repo.get_todo_by_todo_id(todo_id = todo_id)
    if todo:
        # update - is_done값이 True이면 todo.done() 실행, False이면 todo.undone() 실행
        todo.done() if is_done else todo.undone()
        todo: ToDo = todo_repo.update_todo(todo=todo)
        return ToDoSchema.from_orm(todo)
    raise HTTPException(status_code=404, detail="ToDo Not Found")

# DELETE Method 사용하여 todo 아이템 삭제 API
@router.delete("/{todo_id}", status_code=204)
def delete_todo_handler(
    todo_id: int,
    todo_repo: ToDoRepository = Depends(),
    ):
    todo:ToDo | None = todo_repo.get_todo_by_todo_id(todo_id = todo_id)
    if not todo:
        raise HTTPException(status_code=404, detail="ToDo Not Found")
    todo_repo.delete_todo(todo_id=todo_id)
  • orm.py 수정 (내용 추가!)
# /c/Users/관리자/Desktop/projects/todos/src/database/orm.py 내용
from sqlalchemy import Boolean, Column, Integer, String, ForeignKey
from sqlalchemy.orm import declarative_base, relationship		# 추가
from src.schema.request import CreateToDoRequest

Base = declarative_base()

# ToDo 클래스 모델링 한 것
class ToDo(Base):
    __tablename__ = 'todo'
    
    id = Column(Integer, primary_key=True, index=True)
    contents = Column(String(256), nullable=False)
    is_done = Column(Boolean, nullable=False)
    user_id = Column(Integer, ForeignKey("user.id"))
    
    def __repr__(self):
        return f"ToDo(id={self.id}, contents={self.contents}, is_done={self.is_done})"
    
    @classmethod
    def create(cls, request: CreateToDoRequest) -> "ToDo":
        return cls(
            # id값은 DB에 의해 자동으로 결정 되서 별도로 지정안해줌
            contents=request.contents,
            is_done=request.is_done,
        )
        
    def done(self) -> "ToDo":    # ToDo의 is_done 값을 True로 변경 후 ToDo 반환
        self.is_done = True
        return self
    
    def undone(self) -> "ToDo":  # ToDo의 is_done 값을 False로 변경 후 ToDo 반환
        self.is_done = False
        return self

# User 클래스 모델링 한 것
class User(Base):
    __tablename__ = 'user'
    
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(256), unique=True, nullable=False) # unique=True는 내가 넣은 것임
    password = Column(String(256), nullable=False)
    todos = relationship("ToDo", lazy="joined")		# 추가

 

# /c/Users/관리자/Desktop/projects/todos/src/database/repository.py 내용
from sqlalchemy import select, delete
from sqlalchemy.orm import Session
from typing import List
from fastapi import Depends

from src.database.orm import ToDo
from src.database.connection import get_db

class ToDoRepository:
    def __init__(self, session:Session = Depends(get_db)):
        self.session = session 
        
    def get_todos(self,) -> List[ToDo]:
        return list(self.session.scalars(select(ToDo)))

    def get_todo_by_todo_id(self, todo_id: int) -> ToDo | None:
        return self.session.scalar(select(ToDo).where(ToDo.id == todo_id))

    def create_todo(self, todo: ToDo) -> ToDo:
        self.session.add(instance=todo)
        self.session.commit()
        self.session.refresh(instance=todo)
        return todo

    def update_todo(self, todo: ToDo) -> ToDo:
        self.session.add(instance=todo)
        self.session.commit()
        self.session.refresh(instance=todo)
        return todo

    def delete_todo(self, todo_id: ToDo) -> None:
        self.session.execute(delete(ToDo).where(ToDo.id == todo_id))
        self.session.commit()
# /c/Users/관리자/Desktop/projects/todos/src/database/connection.py 내용
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "mysql+pymysql://root:todos@127.0.0.1:3306/todos"

engine = create_engine(DATABASE_URL, echo=True)
SessionFactory = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db():
    session = SessionFactory()
    try:
        yield session
    finally:
        session.close()
# /c/Users/관리자/Desktop/projects/todos/src/schema/request.py 내용

from pydantic import BaseModel

class CreateToDoRequest(BaseModel):
    contents: str
    is_done: bool
# /c/Users/관리자/Desktop/projects/todos/src/schema/response.py 내용
from pydantic import BaseModel
from typing import List

class ToDoSchema(BaseModel):
    id: int
    contents: str
    is_done: bool

    class Config:
        from_attributes = True
        orm_mode = True # 이거 추가 해야지 ConfigError 안남

class ToDoListSchema(BaseModel):
    todos: List[ToDoSchema]
# /c/Users/관리자/Desktop/projects/todos/src/tests/test_main.py 내용
from src.schema.response import ToDoSchema
from src.database.orm import ToDo
from src.database.repository import ToDoRepository      # 추가

# GET Method 사용하여 "/" 검증
def test_health_check(client):      # client 는 내가 conftest.py에서 정의한 fixture
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"ping": "pong"}

# GET Method 사용하여 전체 조회 API 검증
def test_get_todos(client, mocker):
    # order = ASC
    mocker.patch.object(ToDoRepository, "get_todos", return_value=[          # 수정
        ToDo(id=1, contents="FastAPI Section 0", is_done=True),
        ToDoSchema(id=2, contents="FastAPI Section 2", is_done=False),
    ])
    response = client.get("/todos")
    assert response.status_code == 200
    assert response.json() == {
        "todos": [
            {"id": 1, "contents": "FastAPI Section 0", "is_done": True},
            {"id": 2, "contents": "FastAPI Section 2", "is_done": False},
        ]
    }
    
    # order = DESC
    response = client.get("/todos?order=DESC")
    assert response.status_code == 200
    assert response.json() == {
        "todos": [
            {"id":2, "contents": "FastAPI Section 2", "is_done": False},
            {"id":1, "contents": "FastAPI Section 0", "is_done": True},
        ]
    }

# GET Method 사용하여 단일 조회 API 검증
def test_get_todo(client, mocker):
    # 상태코드 200
    mocker.patch.object(ToDoRepository, "get_todo_by_todo_id",        # 수정
        return_value = ToDo(id=1, contents="todo", is_done=True),
    )   
    response = client.get("/todos/1")   # path에 하위 서브 path 적어주기
    assert response.status_code == 200
    assert response.json() == {"id":1, "contents": "todo", "is_done": True }
    
    # 상태코드 404
    mocker.patch.object(ToDoRepository, "get_todo_by_todo_id", return_value = None,) # 수정   
    response = client.get("/todos/1")
    assert response.status_code == 404
    assert response.json() == {"detail":"ToDo Not Found"}

# POST Medthod 사용하여 todo 생성 API 검증
def test_create_todo(client, mocker):
    create_spy = mocker.spy(ToDo, "create")     # mocker의 spy 기능 사용
    mocker.patch.object(ToDoRepository, "create_todo",              # 수정
        return_value=ToDo(id=1, contents="todo", is_done=False),
    )
    
    body = {
        "contents": "test",
        "is_done": False,
    }
    response = client.post("/todos", json=body)
    
    assert create_spy.spy_return.id is None
    assert create_spy.spy_return.contents == "test"
    assert create_spy.spy_return.is_done is False
    
    assert response.status_code == 201
    assert response.json() == {"id":1, "contents":"todo", "is_done":False}


# PATCH Method 사용하여 is_done 값 수정 API 검증
def test_update_todo(client, mocker):
    # 상태코드 200
    
    # get_todo_by_todo_id 검증
    mocker.patch.object(ToDoRepository, "get_todo_by_todo_id",      # 수정
        return_value = ToDo(id=1, contents="todo", is_done=True),
    )
    
    # done() 또는 undone() 검증
    # done = mocker.patch.object(ToDo, "done") # ToDo 객체의 done() 메소드 호출
    undone = mocker.patch.object(ToDo, "undone") # ToDo 객체의 undone() 메소드 호출
    
    # update_todo 검증
    mocker.patch.object(ToDoRepository, "update_todo",              # 수정
        return_value = ToDo(id=1, contents="todo", is_done=False),   # done 경우 is_done = True, undone 경우 is_done=False 주기
    )
    
    response = client.patch("/todos/1", json={"is_done": False})   # done 경우 "is_done": True, undone 경우 "is_done": False 주기
    
    # done() 또는 undone() 검증 - done 또는 undone이 한 번 호출되었는지 확인. 아닐 경우 AssertionError 발생
    # done.assert_called_once_with()
    undone.assert_called_once_with()
    
    assert response.status_code == 200
    assert response.json() == {"id":1, "contents": "todo", "is_done": False }
    
    # 상태코드 404
    mocker.patch.object(ToDoRepository, "get_todo_by_todo_id",        # 수정
        return_value = None,
    )   
    response = client.patch("/todos/1", json={"is_done": True})
    assert response.status_code == 404
    assert response.json() == {"detail":"ToDo Not Found"}
  
# DELTE Method API 검증  
def test_delete_todo(client, mocker):
    # 상태코드 204
    mocker.patch.object(ToDoRepository, "get_todo_by_todo_id",        # 수정
        return_value = ToDo(id=1, contents="todo", is_done=True),
    )
    mocker.patch.object(ToDoRepository, "delete_todo", return_value = None,)  # 수정
    response = client.delete("/todos/1")    # path에 하위 서브 path 적어주기
    assert response.status_code == 204      # status_code만 주고 response.json()는 assert 안 해봐도 됨! 반환값이 없을 것이기에
    
    # 상태코드 404
    mocker.patch.object(ToDoRepository, "get_todo_by_todo_id",return_value = None,)  # 수정
    response = client.delete("/todos/1")
    assert response.status_code == 404
    assert response.json() == {"detail":"ToDo Not Found"}
# /c/Users/관리자/Desktop/projects/todos/src/tests/conftest.py 내용
import pytest
from fastapi.testclient import TestClient
from src.main import app

@pytest.fixture
def client():
	return TestClient(app=app)

'Web 개발 > FAST API (인프런 강의 내용)' 카테고리의 다른 글

5 실습4 회원가입 API 생성 & 비밀번호 암호화(bcrypt)  (1) 2024.10.18
5 기능 고도화 Lazy Loading / Eager Loading  (0) 2024.10.18
5 실습2 기능 고도화 User 테이블 생성  (0) 2024.08.15
5 실습1 기능 고도화 User 모델링  (0) 2024.08.12
5 기능 고도화 SQL JOIN  (0) 2024.08.11

관련글

티스토리툴바