앞 단계 참조 링크:
00 개요
- Docker MySQL에 접속하여 실제 user 테이블 생성한 후 todo 테이블과 연결하는 실습 진행
- FOREIGN KEY 추가
- JOIN 사용하여 테이블 조회
01 User 테이블 생성하기
1. Python Console에서 Docker 접속 후 MySQL 접속하기
$ docker ps -a # 모든 컨테이너 목록 확인 (실행 + 중지 + 종료)
$ docker start todos # todos 컨테이너 실행
todos
$ docker exec -it todos bash # 실행중인 todos 컨테이너 내부 접속 (todos 컨테이너의 bash 실행)
bash-4.4#
$ mysql -u root -p # mysql 접속 (root 사용자로)
Enter password: # 비밀번호 (todos) 입력
Welcome to the MySQL monitor. Commands end with ; or \g.
Your MySQL connection id is 8
Server version: 8.0.36 MySQL Community Server - GPL
Copyright (c) 2000, 2024, Oracle and/or its affiliates.
Oracle is a registered trademark of Oracle Corporation and/or its
affiliates. Other names may be trademarks of their respective
owners.
Type 'help;' or '\h' for help. Type '\c' to clear the current input statement.
mysql>
2. MySQL에서 todos 데이터베이스로 이동하기
mysql> use todos; # todos 데이터베이스로 이동
Reading table information for completion of table and column names
You can turn off this feature to get a quicker startup with -A
Database changed
mysql> select * from todo;
+----+-------------------+---------+
| id | contents | is_done |
+----+-------------------+---------+
| 1 | FastAPI Section 0 | 1 |
| 2 | FastAPI Section 2 | 1 |
| 3 | FastAPI Section 3 | 1 |
+----+-------------------+---------+
3 rows in set (0.00 sec)
3. User 테이블 생성하기
mysql> CREATE TABLE user ( # user 테이블 생성
-> id INTEGER NOT NULL AUTO_INCREMENT,
-> username VARCHAR(256) NOT NULL,
-> password VARCHAR(256) NOT NULL,
-> PRIMARY KEY (id),
-> UNIQUE (username)
-> );
Query OK, 0 rows affected (0.12 sec)
mysql> select * from user; # user 테이블 조회
Empty set (0.00 sec)
02 todo 테이블과 user 테이블 연결하기
1. todo 테이블 수정하기
mysql> ALTER TABLE todo ADD COLUMN user_id INTEGER; # todo 테이블에 user_id 컬럼 추가
Query OK, 0 rows affected (0.05 sec)
Records: 0 Duplicates: 0 Warnings: 0
mysql> select * from todo; # todo 테이블 조회
+----+-------------------+---------+---------+
| id | contents | is_done | user_id |
+----+-------------------+---------+---------+
| 1 | FastAPI Section 0 | 1 | NULL |
| 2 | FastAPI Section 2 | 1 | NULL |
| 3 | FastAPI Section 3 | 1 | NULL |
+----+-------------------+---------+---------+
3 rows in set (0.00 sec)
mysql>
2. todo 테이블의 user_id컬럼을 user 테이블의 id컬럼과 연결시키기
- ALTER TABLE 변경할테이블1 ADD FOREIGN KEY (변경할컬럼1) REFERENCES 참조할테이블2 (참조할컬럼2);
mysql> ALTER TABLE todo ADD FOREIGN KEY (user_id) REFERENCES user (id);
Query OK, 3 rows affected (0.22 sec)
Records: 3 Duplicates: 0 Warnings: 0
3. user 테이블에 사용자 추가하기
- 원래 비밀번호 같은 경우는 해쉬를 해서 넣어줘야 하는데 지금은 확인용으로 해쉬 없이 생성해보기
- INSERT INTO 명령어 사용
mysql> INSERT INTO user (username, password) VALUES ("admin", "password"); # user 테이블에 계정 생성
Query OK, 1 row affected (0.02 sec)
mysql> select * from user; # user 테이블 조회
+----+----------+----------+
| id | username | password |
+----+----------+----------+
| 1 | admin | password |
+----+----------+----------+
1 row in set (0.00 sec)
- 적용된 외래키 제약조건 (FOREIGN KEY) 확인해보기:
mysql> SELECT
-> TABLE_SCHEMA,
-> COLUMN_NAME,
-> CONSTRAINT_NAME,
-> REFERENCED_TABLE_NAME,
-> REFERENCED_COLUMN_NAME
-> FROM
-> INFORMATION_SCHEMA.KEY_COLUMN_USAGE
-> WHERE
-> TABLE_NAME = 'todo'
-> AND REFERENCED_TABLE_NAME IS NOT NULL;
+--------------+-------------+-----------------+-----------------------+------------------------+
| TABLE_SCHEMA | COLUMN_NAME | CONSTRAINT_NAME | REFERENCED_TABLE_NAME | REFERENCED_COLUMN_NAME |
+--------------+-------------+-----------------+-----------------------+------------------------+
| todos | user_id | todo_ibfk_1 | user | id |
+--------------+-------------+-----------------+-----------------------+------------------------+
1 row in set (0.00 sec)
3. user 테이블의 admin 사용자를 todo 테이블에 매핑해주기
- todo 테이블의 id = 1, 2, 3 행들의 user_id 값을 1로 업데이트 해주기
- UPDATE 명령어 사용
mysql> select * from todo; # todo 테이블 조회
+----+-------------------+---------+---------+
| id | contents | is_done | user_id |
+----+-------------------+---------+---------+
| 1 | FastAPI Section 0 | 1 | NULL |
| 2 | FastAPI Section 2 | 1 | NULL |
| 3 | FastAPI Section 3 | 1 | NULL |
+----+-------------------+---------+---------+
3 rows in set (0.00 sec)
mysql> UPDATE todo SET user_id = 1 WHERE id = 1; # todo 테이블 업데이트
Query OK, 1 row affected (0.01 sec)
Rows matched: 1 Changed: 1 Warnings: 0
mysql> UPDATE todo SET user_id = 1 WHERE id = 2;
Query OK, 1 row affected (0.01 sec)
Rows matched: 1 Changed: 1 Warnings: 0
mysql> UPDATE todo SET user_id = 1 WHERE id = 3;
Query OK, 1 row affected (0.02 sec)
Rows matched: 1 Changed: 1 Warnings: 0
mysql> select * from todo;
+----+-------------------+---------+---------+
| id | contents | is_done | user_id |
+----+-------------------+---------+---------+
| 1 | FastAPI Section 0 | 1 | 1 |
| 2 | FastAPI Section 2 | 1 | 1 |
| 3 | FastAPI Section 3 | 1 | 1 |
+----+-------------------+---------+---------+
3 rows in set (0.00 sec)
- user_id가 다 1로 업데이트 된 것 확인
4. todo 테이블과 user 테이블 JOIN해보기
- SELECT * FROM 테이블1 테이블1별명 JOIN 테이블2 테이블2별명 ON 테이블1별명.테이블1의컬럼 = 테이블2별명.테이블2의컬럼;
mysql> SELECT * FROM todo t JOIN user u ON t.user_id = u.id;
+----+-------------------+---------+---------+----+----------+----------+
| id | contents | is_done | user_id | id | username | password |
+----+-------------------+---------+---------+----+----------+----------+
| 1 | FastAPI Section 0 | 1 | 1 | 1 | admin | password |
| 2 | FastAPI Section 2 | 1 | 1 | 1 | admin | password |
| 3 | FastAPI Section 3 | 1 | 1 | 1 | admin | password |
+----+-------------------+---------+---------+----+----------+----------+
3 rows in set (0.00 sec)
- 왼쪽에는 todo 테이블의 컬럼값들, 오른쪽에는 user 테이블의 컬럼값들이 합쳐져서 출력되는 것 확인
- 특정 컬럼값만 출력해보기
mysql> SELECT u.username, t.contents, t.is_done
-> FROM todo t
-> JOIN user u
-> ON t.user_id = u.id;
+----------+-------------------+---------+
| username | contents | is_done |
+----------+-------------------+---------+
| admin | FastAPI Section 0 | 1 |
| admin | FastAPI Section 2 | 1 |
| admin | FastAPI Section 3 | 1 |
+----------+-------------------+---------+
3 rows in set (0.00 sec)
이때까지의 코드들: 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
# /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)
# /c/Users/관리자/Desktop/projects/todos/src/database/orm.py 내용
from sqlalchemy import Boolean, Column, Integer, String, ForeignKey
from sqlalchemy.orm import declarative_base
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)
# /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/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.ession.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/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)
