앞 단계 참조 링크:
00 개요
- 목적: 데이터베이스 테이블을 sqlalchemy의 ORM을 이용해 모델링을 하는 실습 진행
01 ORM 사용하여 모델링
0. 환경
1. database 폴더 안에 파일 (orm.py) 생성
- C:\Users\관리자\Desktop\projects\todos\src\database\orm.py 생성
$ touch /c/Users/관리자/Desktop/projects/todos/src/database/orm.py
2. orm.py에 쓰기
# /c/Users/관리자/Desktop/projects/todos/src/database/orm.py
from sqlalchemy import Boolean, Column, Integer, String
from sqlalchemy.orm import declarative_base
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)
def __repr__(self):
return f"ToDo(id={self.id}, contents={self.contents}, is_done={self.is_done})"
- from sqlalchemy import Boolean, Column, Integer, String
- 테이블 모델링에 사용될 데이터 타입 설정을 위해 sqlalchemy에서 Boolean, Column, Integer, String 가져오기
- from sqlalchemy.orm import declarative_base
- Base라는 클래스 생성하기 위해 sqlalchemy.orm에서 declarative_base 가져오기
- Base 클래스를 상속받아 데이터베이스 테이블을 어떤 클래스로 모델링 해주도록 할 것
- class ToDo(Base):
- ToDo라는 이름의 테이블을 모델링할 것
- Base를 상속받음
- __tablename__ = 'todo'
- __tablename__ 속성을 todo로 설정
- 앞서 ToDo 테이블 (i.e., class ToDo(Base))을 ToDo라는 이름으로 생성했기 때문에 여기에 동일한 tablename의 이름을 연결시켜줘야됨
- (다음 부분은 앞서 생성한 실제 데이터베이스와 동일한 옵션들을 사용할 수 있도록 연결 해주면 됨)
- id = Column(Integer, primary_key=True, index=True)
- id 라는 이름의 Column 설정: Integer 데이터 타입, primay key = true, index = true (i.e., 이 column을 인덱스로 설정한다는 의미) 설정
- contents = Column(String(256), nullable=False)
- contents 라는 이름의 Column 설정: String(256) 데이터 타입, nullable=False (i.e., 빈 값 불가능) 설정
- is_done = Column(Boolean, nullable=False)
- is_done 이라는 이름의 Column 설정: Boolean 데이터 타입, nullable=False (i.e., 빈 값 불가능) 설정
- def __repr__(self):
- repr을 오버라이드해서 써줌
- 실슬 진행 시 객체를 출력하는 경우가 종종 있는데 그럴 때마다 어떤 todo 객체가 출력되는지 쉽게 보기 위해 python 클래스의 repr이란 매직 메소드를 오버라이드해서 사용할 것
- return f"ToDo(id={self.id}, contents={self.contents}, is_done={self.is_done})"
- f string을 이용해 ToDo 클래스 안에 있는 값들을 출력하도록설정
- self.id를 통해 id 값 출력
- self.contents를 통해 contents 값 출력
- self.is_done을 통해 is_done 값 출력
3. 데이터베이스에 있는 데이터 조회 (앞서 만든 ToDo 클래스 사용)
0) 가상환경 활성화하기
$ source ~/Desktop/projects/todos/bin/activate # 강사님
$ source ~/Desktop/projects/todos/Scripts/activate # 나
# 잘 작동하는지 확인
(todos)$ python --version
Python 3.12.1
관리자@DESKTOP-THDM2MN MINGW64 ~/Desktop/projects/todos # 내 실습
$ . ~/Desktop/projects/todos/Scripts/activate # 내 실습
(todos) # 내 실습
관리자@DESKTOP-THDM2MN MINGW64 ~/Desktop/projects/todos # 내 실습
$ # 내 실습
1) docker 컨테이너 실행
$ docker ps -a # 모든 프로세스 출력 (실행 중 & 실행 안하고 있는 것들도)
CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES
0c3283e97292 mysql:8.0 "docker-entrypoint.s…" 8 days ago Exited (0) About a minute ago todos
2597a124c27d docker/welcome-to-docker:latest "/docker-entrypoint.…" 8 days ago Exited (0) 8 days ago welcome-to-docker
$ docker start 컨테이너ID # 꺼져 있는 프로세스 실행
$ docker start todos # 내 실습
$ docker ps -a # 내 실습
CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES
0c3283e97292 mysql:8.0 "docker-entrypoint.s…" 8 days ago Up 6 seconds 0.0.0.0:3306->3306/tcp, 33060/tcp todos
2597a124c27d docker/welcome-to-docker:latest "/docker-entrypoint.…" 8 days ago Exited (0) 8 days ago welcome-to-docker
1) 경로는 src에서 Python Console 실행하기
(todos)
관리자@DESKTOP-THDM2MN MINGW64 ~/Desktop/projects/todos/src
$ 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 database.orm import ToDo
>>> from sqlalchemy import select
>>> session = SessionFactory()
>>> list(session.scalar(select(ToDo)))
- from database.connection import SessionFactory
- database.connection (database 폴더 안에 connection.py)에서 SessionFactory( ) 불러오기
- from database.orm import ToDo
- database.orm (database 폴더 안에 orm.py)에서 ToDo( ) 불러오기
- from sqlalchemy import select
- sqlalchemy에서 select 함수 불러오기
- session = SessionFactory()
- SessionFactory 사용하여 session 객체 생성하기
- list(session.scalars(select(ToDo)))
- SessionFactory의 scalars (scalar가 아닌 scalars 임을 주의)를 통해 ToDo을 select하여 출력하는 query를 데이터베이스에 날려보기
- session.scalars(select(ToDo))만 입력 시
-
- 이런 경우는 ScalarResult라는 객체(object)로 반환됨
- 그래서 list()로 감싸줘야 됨
2024-05-02 09:36:50,594 INFO sqlalchemy.engine.Engine SELECT DATABASE()
2024-05-02 09:36:50,595 INFO sqlalchemy.engine.Engine [raw sql] {}
2024-05-02 09:36:50,599 INFO sqlalchemy.engine.Engine SELECT @@sql_mode
2024-05-02 09:36:50,599 INFO sqlalchemy.engine.Engine [raw sql] {}
2024-05-02 09:36:50,601 INFO sqlalchemy.engine.Engine SELECT @@lower_case_table_names
2024-05-02 09:36:50,602 INFO sqlalchemy.engine.Engine [raw sql] {}
2024-05-02 09:36:50,606 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2024-05-02 09:36:50,608 INFO sqlalchemy.engine.Engine SELECT todo.id, todo.contents, todo.is_done
FROM todo
2024-05-02 09:36:50,609 INFO sqlalchemy.engine.Engine [generated in 0.00052s] {}
ToDo(id=1, contents=FastAPI Section 0, is_done=True)
2024-05-02 09:41:34,852 INFO sqlalchemy.engine.Engine [cached since 284.2s ago] {}
<sqlalchemy.engine.result.ScalarResult object at 0x00000252B827BCF0>- list(session.scalars(select(ToDo)))를 입력 시
-
>>> list(session.scalars(select(ToDo)))
2024-05-02 09:46:44,769 INFO sqlalchemy.engine.Engine SELECT todo.id, todo.contents, todo.is_done
FROM todo
2024-05-02 09:46:44,774 INFO sqlalchemy.engine.Engine [cached since 594.2s ago] {}
[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=False)]
>>>
- 이렇게 하므로서 ToDo 테이블에 있는 모든 객체/레코드를 list 형태로 반환된 것을 확인
- 출력 결과 내용("SELECT todo.id, todo.contents, todo.is_done FROM todo" 등)을 보면 sqlalchemy에 의해 ORM을 사용하여 'SELECT ~ FROM 테이블명' 식으로 ToDo의 모든 컬럼들을 조회하는 query가 자동으로 생성 및 실행됨을 확인
- Error:
- MySQL과 연결이 안 된다는 error 메세지 출력 경우
Traceback (most recent call last):
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\pymysql\connections.py", line 644, in connect
sock = socket.create_connection(
^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\socket.py", line 852, in create_connection
raise exceptions[0]
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\socket.py", line 837, in create_connection
sock.connect(sa)
ConnectionRefusedError: [WinError 10061] 대상 컴퓨터에서 연결을 거부했으므로 연결하지 못했습니다
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\engine\base.py", line 146, in __init__
self._dbapi_connection = engine.raw_connection()
^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\engine\base.py", line 3304, in raw_connection
return self.pool.connect()
^^^^^^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\pool\base.py", line 449, in connect
return _ConnectionFairy._checkout(self)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\pool\base.py", line 1263, in _checkout
fairy = _ConnectionRecord.checkout(pool)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\pool\base.py", line 712, in checkout
rec = pool._do_get()
^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\pool\impl.py", line 179, in _do_get
with util.safe_reraise():
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\util\langhelpers.py", line 146, in __exit__
raise exc_value.with_traceback(exc_tb)
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\pool\impl.py", line 177, in _do_get
return self._create_connection()
^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\pool\base.py", line 390, in _create_connection
return _ConnectionRecord(self)
^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\pool\base.py", line 674, in __init__
self.__connect()
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\pool\base.py", line 900, in __connect
with util.safe_reraise():
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\util\langhelpers.py", line 146, in __exit__
raise exc_value.with_traceback(exc_tb)
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\pool\base.py", line 896, in __connect
self.dbapi_connection = connection = pool._invoke_creator(self)
^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\engine\create.py", line 643, in connect
return dialect.connect(*cargs, **cparams)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\engine\default.py", line 617, in connect
return self.loaded_dbapi.connect(*cargs, **cparams)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\pymysql\connections.py", line 358, in __init__
self.connect()
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\pymysql\connections.py", line 711, in connect
raise exc
pymysql.err.OperationalError: (2003, "Can't connect to MySQL server on '127.0.0.1' ([WinError 10061] 대상 컴퓨터에서 연결을 거부했으므로 연결하지 못했습니다)")
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\orm\session.py", line 2354, in scalar
return self._execute_internal(
^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\orm\session.py", line 2181, in _execute_internal
conn = self._connection_for_bind(bind)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\orm\session.py", line 2050, in _connection_for_bind
return trans._connection_for_bind(engine, execution_options)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "<string>", line 2, in _connection_for_bind
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\orm\state_changes.py", line 139, in _go
ret_value = fn(self, *arg, **kw)
^^^^^^^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\orm\session.py", line 1144, in _connection_for_bind
conn = bind.connect()
^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\engine\base.py", line 3280, in connect
return self._connection_cls(self)
^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\engine\base.py", line 148, in __init__
Connection._handle_dbapi_exception_noconnection(
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\engine\base.py", line 2444, in _handle_dbapi_exception_noconnection
raise sqlalchemy_exception.with_traceback(exc_info[2]) from e
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\engine\base.py", line 146, in __init__
self._dbapi_connection = engine.raw_connection()
^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\engine\base.py", line 3304, in raw_connection
return self.pool.connect()
^^^^^^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\pool\base.py", line 449, in connect
return _ConnectionFairy._checkout(self)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\pool\base.py", line 1263, in _checkout
fairy = _ConnectionRecord.checkout(pool)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\pool\base.py", line 712, in checkout
rec = pool._do_get()
^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\pool\impl.py", line 179, in _do_get
with util.safe_reraise():
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\util\langhelpers.py", line 146, in __exit__
raise exc_value.with_traceback(exc_tb)
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\pool\impl.py", line 177, in _do_get
return self._create_connection()
^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\pool\base.py", line 390, in _create_connection
return _ConnectionRecord(self)
^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\pool\base.py", line 674, in __init__
self.__connect()
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\pool\base.py", line 900, in __connect
with util.safe_reraise():
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\util\langhelpers.py", line 146, in __exit__
raise exc_value.with_traceback(exc_tb)
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\pool\base.py", line 896, in __connect
self.dbapi_connection = connection = pool._invoke_creator(self)
^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\engine\create.py", line 643, in connect
return dialect.connect(*cargs, **cparams)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\engine\default.py", line 617, in connect
return self.loaded_dbapi.connect(*cargs, **cparams)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\pymysql\connections.py", line 358, in __init__
self.connect()
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\pymysql\connections.py", line 711, in connect
raise exc
sqlalchemy.exc.OperationalError: (pymysql.err.OperationalError) (2003, "Can't connect to MySQL server on '127.0.0.1' ([WinError 10061] 대상 컴퓨터에서 연결을 거부했으므로 연결 하지 못했습니다)")
(Background on this error at: https://sqlalche.me/e/20/e3q8)
- 이유: MySQL이 설치된 docker 컨테이너가 실행되고 있지 않았었음
- 해결: docker 컨테이너 (컨테이너ID = todos)를 실행시키고 난 후 다시 실습해본 결과 잘 출력됨
3) 다음을 입력
- todos 변수에 ToDo 테이블의 모든 객체/레코드를 list 형태로 할당
>>> todos = list(session.scalars(select(ToDo)))
2024-05-02 10:02:11,177 INFO sqlalchemy.engine.Engine SELECT todo.id, todo.contents, todo.is_done
FROM todo
2024-05-02 10:02:11,178 INFO sqlalchemy.engine.Engine [cached since 1521s ago] {}
>>> 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=False)]
- todoes에서 각 todo 반복문 돌려서 각 todo 출력
>>> for todo in todos:
... print(todo)
...
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=False)
- todo가 정상적으로 Python 안에서 사용 가능한 것 확인
결론
- 데이터베이스 테이블(ToDo 테이블)과 생성한 클래스(ToDo 클래스)를 연동시켜 사용하는 기술을 ORM이라고 함
- ORM을 이용하여 데이터베이스에서 테이블에 있는 데이터를 조회하는 실습을 진행해 봄
이때까지의 코드들:
# .../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")
# .../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/database/orm.py
from sqlalchemy import Boolean, Column, Integer, String
from sqlalchemy.orm import declarative_base
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)
def __repr__(self):
return f"ToDo(id={self.id}, contents={self.contents}, is_done={self.is_done})"
