앞 단계 참조 링크:
00 개요
- SQLAlchemy를 이용하여 User 테이블 모델링 및 모델링한 코드를 기반으로 MySQL에 User 테이블을 생성하는 실습 진행
01 SQLAlchemy 기능 - CREATE TABLE
- 이전에 우리는 CREATE TABLE이라는 명령어를 직접 작성하여 DB에 실행하였는데, sqlalchemy에는 SQL문을 자동으로 만들어주는 기능 존재
1. SQL문 생성하기
- 클래스/테이블을 먼저 모델링한 후 거기에 맞는 SQL문 생성 가능
1) Class (Table) 모델링하기
- 우리는 orm.py에 이미 ToDo라는 클래스(테이블)을 모델링 해놨음
# /c/Users/관리자/Desktop/projects/todos/src/database/orm.py 내용
from sqlalchemy import Boolean, Column, Integer, String
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)
...
2) CREATE TABLE 문 출력하기
- Python Console에서 실행:
- 필요한 모듈 및 클래스 참조하기
- from sqlalchemy.schema import CreateTable
- from src.database.orm import ToDo
- from src.database.connection import engine
- 우리가 src.database.orm에 모델링한 테이블 (ToDo 클래스)에 맞게 테이블 생성하는 SQL문 출력하기
- print(CreateTable(ToDo.__table__).compile(engine))
>>> from sqlalchemy.schema import CreateTable
>>> from src.database.orm import ToDo
>>> from src.database.connection import engine
>>> print(CreateTable(ToDo.__table__).compile(engine))
CREATE TABLE todo (
id INTEGER NOT NULL AUTO_INCREMENT,
contents VARCHAR(256) NOT NULL,
is_done BOOL NOT NULL,
PRIMARY KEY (id)
)
02 User 테이블 생성하기
- User 테이블을 생성할 때 SQL문을 직접 적지 않고 User 테이블을 모델링 한 다음 SQLAlchemy를 이용하여 SQL문으로 변환하여 그 CREATE TABLE 명령어를 사용할 것
1. orm.py에 User 테이블 모델링하기
- 먼저 User라는 클래스/테이블을 모델링 해줘야 함
- 그 후 ToDo 테이블과 User 테이블을 JOIN하여 사용하기 위해 ToDo 클래스/테이블에 Foreign Key를 생성해야 함
- orm.py에 다음과 같이 작성하기:
- 추가 내용:
- ForeignKey 생성하기 위해 sqlachemy에서 ForeignKey import
- User라는 클래스/테이블 모델링
- ToDo 클래스/테이블에 user 테이블의 id 컬럼 값을 참조하는 Foreign Key (이름: user_id) 추가
# /c/Users/관리자/Desktop/projects/todos/src/database/orm.py 내용
from sqlalchemy import Boolean, Column, Integer, String, ForeignKey # 추가 - 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")) # 추가 - user 테이블의 id컬럼 값과 FK로 연결
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(
contents=request.contents,
is_done=request.is_done,
)
def done(self) -> "ToDo":
self.is_done = True
return self
def undone(self) -> "ToDo":
self.is_done = False
return self
class User(Base): # 추가 - Base 모델을 상속 받음
__tablename__ = "user" # 추가 - tablename이라는 속성에 "user"라는 이름 선언
id = Column(Integer, primary_key=True, index=True) # 추가 - id 컬럼 생성
username = Column(String(256), nullable=False) # 추가 - username 컬럼 생성
password = Column(String(256), nullable=False) # 추가 - password 컬럼 생성
- 설명:
- from sqlalchemy import ForeignKey
- user_id = Column(Integer, ForeignKey("user.id"))
- ToDo 클래스에 user_id라는 컬럼 생성
- user_id 값이 user라는 테이블의 id 컬럼 값에 연결되도록 Foreign Key로설정
- class User(Base):
- Base 모델을 상속 받는 User라는 클래스 생성
- __tablename__ = "user"
- User 클래스의 __tablename__이라는 속성에 "user"라는 이름 선언하여 테이블명을 "user"로 설정
- id = Column(Integer, primary_key=True, index=True)
- username = Column(String(256), nullable=False)
- password = Column(String(256), nullable=False)
2. User 테이블을 생성하는 SQL문 출력하기
>>> from sqlalchemy.schema import CreateTable
>>> from src.database.orm import ToDo, User
>>> from src.database.connection import engine
>>> print(CreateTable(User.__table__).compile(engine))
CREATE TABLE user (
id INTEGER NOT NULL AUTO_INCREMENT,
username VARCHAR(256) NOT NULL,
password VARCHAR(256) NOT NULL,
PRIMARY KEY (id),
UNIQUE (username)
)
- CREATE TABLE 문을 복사하여 이후에 user 테이블 생성 시 사용할 것!
추가)
- ToDo 클래스/테이블도 변경되었기에 CREATE TABLE 문 출력해보기
- Python Console에서 실행:
>>> from sqlalchemy.schema import CreateTable
>>> from src.database.orm import ToDo, User
>>> from src.database.connection import engine
>>> print(CreateTable(ToDo.__table__).compile(engine))
CREATE TABLE todo (
id INTEGER NOT NULL AUTO_INCREMENT,
contents VARCHAR(256) NOT NULL,
is_done BOOL NOT NULL,
user_id INTEGER,
PRIMARY KEY (id),
FOREIGN KEY(user_id) REFERENCES user (id)
)
- todo라는 테이블은 우리가 이미 생성하였기에 해당 CREATE TABLE 명령어를 실행할 것이 아니라, ALTER TABLE 명령어를 이용하여 우리가 새로 추가한 user_id 추가 및 Foreign Key 관계만 추가할 것!
SQL문)
CREATE TABLE user (
id INTEGER NOT NULL AUTO_INCREMENT,
username VARCHAR(256) NOT NULL,
password VARCHAR(256) NOT NULL,
PRIMARY KEY (id)
);
ALTER TABLE todo ADD COLUMN user_id INTEGER;
ALTER TABLE todo ADD FOREIGN KEY(user_id) REFERENCES user (id);
INSERT INTO user (username, password) VALUES ("admin", ”password”);
UPDATE todo SET user_id = 1 WHERE id = 1;
SELECT * FROM todo t JOIN user u ON t.user_id = u.id;
이때까지의 코드들: 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":
self.is_done = True
return self
def undone(self) -> "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)
