00 개요
- 목적: sqlalchemy를 이용하여 앞서 생성한 데이터베이스와 Python project를 연결하는 실습 진행
01 데이터베이스 연결
0. 환경
- mysql에 접속하고 있다면 Ctrl + Z 해서 접속 해제한 후 진행하기
- 가상환경 활성화하기
$ source ~/Desktop/projects/todos/bin/activate # 강사님
$ source ~/Desktop/projects/todos/Scripts/activate # 나
# 잘 작동하는지 확인
(todos)$ python --version
Python 3.12.1
1. 필요한 library 설치: sqlalchemy, pymysql, cryptography
(todos)$ pip install sqlalchemy
(todos)$ pip install pymysql
(todos)$ pip install cryptography
- pymysql: Python과 MySQL 연동 시 사용되는 드라이버
- cryptography: pymysql을 통해 MySQL에 접속 시 인증 또는 암호 관련된 처리를 해주는 library
2. src 안에 데이터베이스 폴더(database) 생성
(todos)$ mkdir ~/Desktop/projects/todos/src/database
3. database 폴더 안에 파일(connection.py) 생성
(todos)$ touch ~/Desktop/projects/todos/src/database/connection.py
4. connection.py에 쓰기
1) 데이터베이스 관련된 인증 정보를 상수로 하나 선언해주기
# .../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)
- DATABASE_URL 상수 생성
- root:todos → root 계정에 todos라는 비밀번호 사용
- @127.0.0.1:3306 → 127.0.0.1(local host)에 3306포트 사용
- /todos → todos 데이터베이스 사용
- engine 개체 생성
- sqlalchemy를 이용하여 데이터베이스에 접속하기 위해서는 engine이라는 개체를 생성해야됨
- sqlalchemy의 create_engine()를 이용하여 engine 객체 생성
- create_engine()에 DATABASE_URL 전달
- echo=True → sqlalchemy에 의해 query들이 대신 처리될 때 어떤 sql이 사용됐는지 그 사용되는 sql을 출력해주는 옵션
- SessionFactory 변수 생성
- sqlalchemy.orm의 sessionmaker()을 이용하여 SessionFactory 변수 생성
- 이렇게 만든 SessionFactory를 이용하여 Session 생성하고, 이 Session 인스턴스를 통해 데이터베이스와 통신할 것
- sessionmaker()의 옵션
- autocommit=False → 자동으로 commit하지 않고 명시적으로 commit하겠다는 옵션
- autoflush=False → 자동으로 flush하지 않고 명시적으로 flush하겠다는 옵션
- bind=engine → (우리가 전달한 DATABASE_URL을 통해 생성된) engine이라는 객체를 통해 Session을 만들 수 있게 하는 옵션
5. 데이터베이스 잘 연결 확인 - Python Console 실행하여 확인
0) 경로는 src에서 python console 실행하기
1) Python Console 실행
(todos)$ python
Python 3.12.1 (tags/v3.12.1:2305ca5, Dec 7 2023, 22:03:25) [MSC v.1937 64 bit (AMD64)] on win32
Type "help", "copyright", "credits" or "license" for more information.
>>>
2) 다음을 입력
>>> from database.connection import SessionFactory
>>> from sqlalchemy import select
>>> session = SessionFactory()
>>> session.scalar(select(1))
2024-04-25 20:28:01,430 INFO sqlalchemy.engine.Engine SELECT DATABASE()
2024-04-25 20:28:01,431 INFO sqlalchemy.engine.Engine [raw sql] {}
2024-04-25 20:28:01,436 INFO sqlalchemy.engine.Engine SELECT @@sql_mode
2024-04-25 20:28:01,437 INFO sqlalchemy.engine.Engine [raw sql] {}
2024-04-25 20:28:01,439 INFO sqlalchemy.engine.Engine SELECT @@lower_case_table_names
2024-04-25 20:28:01,440 INFO sqlalchemy.engine.Engine [raw sql] {}
2024-04-25 20:28:01,444 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2024-04-25 20:28:01,445 INFO sqlalchemy.engine.Engine SELECT 1
2024-04-25 20:28:01,446 INFO sqlalchemy.engine.Engine [generated in 0.00035s] {}
1
>>>
- from database.connection import SessionFactory
- database.connection에서 SessionFactory( ) 불러오기
- from sqlalchemy import select
- sqlalchemy에서 select 함수 불러오기
- session = SessionFactory()
- SessionFactory 사용하여 session 객체 생성하기
- session.scalar(select(1))
- SessionFactory의 scalar를 통해 1을 select하여 출력하는 query를 데이터베이스에 날려보기
- 어떤 의미가 있는 데이터가 출력이 되진 않지만 이 session을 통해 이렇게 쿼리를 날리는 것 만으로도 데이터베이스가 잘 연결됐는지 확인 가능
- 출력된 로그
- echo=True 옵션을 줬기에 출력되는 로그
- 로그를 보면 데이터베이스에 연결을 해서 select 1 query가 발생했다는 것 확인 가능
- 정상적으로 처리가 됐기에 마지막에 1이 출력 됨
- 만약 정상적이지 않은 경우 (예 - docker가 정상적으로 실행되고 있지 않거나 todos 데이터베이스가 잘 생성되지 않았을 경우 등) select 1 query를 날려도 아무것도 출력되지 않을 것
- 이렇게 출력된 결과는 데이터베이스가 잘 돌고 있고 데이터베이스와 sqlalchemy가 잘 연결됐다는 의미
이때까지의 코드들:
# .../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)
# .../todos/src/main.py 내용
from fastapi import FastAPI, Body, HTTPException
from pydantic import BaseModel
app = FastAPI()
# 첫 화면 API
@app.get("/")
def health_check_handler():
return {"ping": "pong"}
# POST 생성을 위해 사용자로부터 전달받을 request 클래스 생성
class CreateToDoRequest(BaseModel):
id: int
contents: str
is_done: bool
# 데이터베이스 역할하는 딕셔너리 생성
todo_data = {
1: {
"id": 1,
"content": "실전! FastAPI 섹션 0 수강",
"is_done": True,
},
2: {
"id": 2,
"content": "실전! FastAPI 섹션 1 수강",
"is_done": False,
},
3: {
"id": 3,
"content": "실전! FastAPI 섹션 2 수강",
"is_done": False,
},
}
# GET Method 사용하여 전체 조회 API
@app.get("/todos", status_code=200)
def get_todos_handler(order: str | None = None):
rt = list(todo_data.values())
if order and order == "DESC":
return rt[::-1]
return rt
# GET Method 사용하여 단일 조회 API
@app.get("/todos/{todo_id}", status_code=200)
def get_todo_handler(todo_id: int):
todo = todo_data.get(todo_id)
if todo:
return todo
raise HTTPException(status_code=404, detail="ToD Not Found")
# POST Medthod 사용하여 todo 생성 API
@app.post("/todos", status_code=201)
def create_todos_handler(request: CreateToDoRequest):
todo_data[request.id] = request.dict()
return todo_data[request.id]
# PATCH Method 사용하여 is_done 값 수정 API
@app.patch("/todos/{todo_id}", status_code=200)
def update_todo_handler(
todo_id: int,
is_done: bool = Body(..., embed=True)
):
todo = todo_data.get(todo_id)
if todo:
todo["is_done"] = is_done
return todo
raise HTTPException(status_code=404, detail="ToDo Not Found")
# DELETE Method 사용하여 todo 아이템 삭제 API
@app.delete("/todos/{todo_id}", status_code=204)
def delete_todo_handler(todo_id: int):
todo = todo_data.pop(todo_id, None)
if todo:
return
raise HTTPException(status_code=404, detail="ToDo Not Found")
