[파이썬] FastAPI - ORM 연동하기(SQLAlchemy)

이전 내용

[파이썬] FastAPI - 웹소켓

---

데이터베이스와 ORM

 

ORM은 복잡한 SQL 쿼리 대신 객체지향 방식으로 데이터베이스를 조작할 수 있게 해주는 프로그래밍 기법이다.

보다 상세한 내용은 하단 링크 참고

[Java] Spring Boot: ORM, JPA, 하이버네이트

---

FastAPI와 ORM

 

FastAPI 자체는 특정 ORM을 내장하고 있지 않지만, SQLAlchemy와 같은 ORM 라이브러리와 통합된다. SQLAlchemy를 사용하면 FastAPI 애플리케이션 내에서 데이터베이스 모델을 정의하고, 이를 통해 데이터베이스 작업을 손쉽게 수행할 수 있다.

 

◆ SQLAlchemy 사용하기

SQLAlchemy 공식 사이트에 나와 있는 자료에 따르면,

https://docs.sqlalchemy.org/en/20/intro.html#installation-guide

 

지원되는 설치 방법은 아래와 같다.

 

또한, 데이터베이스 종류에 따라 알맞는 드라이버(라이브러리)가 필요하며, pip install (또는 conda install)을 통해 같이 설치해주면 된다.

데이터베이스 종류	드라이버
MySQL	pymysql, mysqlclient
PostgreSQL	psycopg2
SQLite	sqlite3
Oracle	cx_Oracle
Microsoft SQL Server	pyodbc, pymssql
MongoDB	pymongo
Cassandra	cassandra-driver
Redis	redis-py

 

---

SQLAlchemy의 주요 문법

 

- 데이터베이스 연결 설정: 데이터베이스에 연결하기 위해서는 create_engine 함수 사용.

from sqlalchemy import create_engine

- 데이터베이스 URL 설정

engine = create_engine('mysql+pymysql://user:password@host:port/dbname')

- 모델 선언, 테이블 정의: SQLAlchemy에서 데이터베이스 모델을 선언하는 과정은 데이터베이스 테이블의 스키마를 파이썬 클래스로 표현하는 것이다. 이때, declarative_base 클래스를 사용하여 모델을 생성할 수 있다.

• declarative_base() 함수는 SQLAlchemy의 모든 모델이 상속받아야 하는 기본 클래스를 생성하며, 이 기본 클래스는 SQLAlcheymy에게 해당 클래스의 객체들이 데이터베이스 테이블을 어떻게 나타내는지 알려준다.
• __tablename__ 을 클래스에 명시적으로 선언하면 실제 데이터베이스에서 사용될 테이블 이름으로 지정된다. 만약 선언하지 않을 경우, 클래스 이름의 소문자 변환을 디폴트 테이블명으로 사용한다.

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String

# 기본 클래스 생성
Base = declarative_base()

class User(Base):
    __tablename__ = 'users'  # 테이블 이름 설정
    
    # 칼럼 정의
    id = Column(Integer, primary_key=True, index=True) # 자동으로 인덱스 생성
    name = Column(String(50), unique=True, index=True) # unique: 고윳값
    age = Column(Integer)

Column 클래스는 SQLAlchemy에서 데이터베이스 테이블의 각 열(컬럼)을 정의할 때 사용하는 중요한 기능이다.

[Column 클래스에서 사용할 수 있는 다양한 옵션]

1. 기본 옵션

• type: 컬럼의 데이터 타입 정의. 예: Integer, String, Float, Boolean 등.
• primary_key: 해당 컬럼이 기본 키(primary key)임을 지정하는 옵션. True로 설정하면 기본 키로 사용.
• nullable: 해당 컬럼이 NULL 값을 허용하는지 여부 설정. 기본값은 True.
• default: 컬럼의 기본값을 설정할 수 있는 옵션.
• index: 인덱스를 생성할지를 지정. True로 설정하면 해당 컬럼에 대해 인덱스 생성.
• unique: 해당 컬럼의 값이 유일해야 함을 지정. True로 설정하면 중복된 값이 허용되지 않는다.

2. 제약 조건 옵션

• foreign_key: 외래 키를 정의할 때 사용. 다른 테이블의 특정 컬럼을 참조할 수 있다.
• server_default: 데이터베이스에서 기본값을 설정할 때 사용. (default와의 차이점은 이 옵션이 SQL이 서버에서 실행될 때 적용.)

3. 문자열 타입 옵션

• length: String 타입에서 열의 길이를 제한할 때 사용.

4. 기타 옵션

• onupdate: 데이터가 업데이트될 때 적용될 기본값을 설정.
• comment: 컬럼에 대한 설명을 추가. 데이터베이스에서 주석으로 저장.
• info: 사용자가 정의할 수 있는 추가 정보를 제공하는 딕셔너리. 예를 들어, 문서화 및 메타데이터 관리에 사용할 수 있다.

- 테이블 생성: 정의한 테이블을 데이터베이스에 생성.

Base.metadata.create_all(engine)

 - 데이터 추가: 세션을 사용하여 데이터를 추가.

from sqlalchemy.orm import sessionmaker

# 세션 생성
Session = sessionmaker(bind=engine)
session = Session()

# 데이터 추가
new_user = User(name='John Doe', age=30)
session.add(new_user)
session.commit()  # 변경 사항을 커밋

- 데이터 조회: 데이터베이스에서 데이터를 조회하는 방법.

# 모든 사용자 조회
users = session.query(User).all()
for user in users:
    print(f'ID: {user.id}, Name: {user.name}, Age: {user.age}')

# 특정 조건으로 조회
specific_user = session.query(User).filter(User.name == 'John Doe').first()

- 데이터 수정: 기존 데이터 수정 방법.

user_to_update = session.query(User).filter(User.name == 'John Doe').first()
user_to_update.age = 31  # 나이를 31로 수정
session.commit()  # 변경 사항을 커밋

- 데이터 삭제.

user_to_delete = session.query(User).filter(User.name == 'John Doe').first()
session.delete(user_to_delete)
session.commit()  # 변경 사항을 커밋

 

- 세션 종료: 작업 완료 후에는 세션을 종료해야 한다.

session.close()

---

◆ 모델 생성 코드 예

from sqlalchemy import Column, Integer, String, Boolean, DateTime, Float, ForeignKey
from datetime import datetime
from sqlalchemy.ext.declarative import declarative_base

# 베이스 클래스 생성
Base = declarative_base()

class User(Base):
    # 테이블 이름 설정
    __tablename__ = 'users'
    
    # id 필드 설정
    id = Column(Integer, primary_key=True, autoincrement=True, comment="기본키")
    # username 필드 설정
    username = Column(String(50), unique=True, nullable=False, index=True, comment="사용자 이름")
    # 'eamil' 필드 설정
    email = Column(String(150), unique=True, nullable=False, index=True, comment="이메일")
    # is_active 필드 설정
    is_active = Column(Boolean, default=True, comment="활성화 여부")
    # created_at 필드 설정
    crated_at = Column(DateTime, default=datetime.utcnow, comment="생성일")
    # grade 필드 설정
    grade = Column(Float, comment="사용자 등급")
    # pared_id 필드 설정
    parent_id = Column(Integer, ForeignKey('parents.id'), comment="상위 ID")

---

FastAPI 데이터베이스 연동 과정

 

먼저 FastAPI 애플리케이션을 작성하고, SQLAlchemy로 데이터베이스(MySQL, PostgreSQL 등)에 접속하여 데이터베이스를 생성한다.

CREATE DATABASE 데이터베이스이름;

 

생성 후에는 FastAPI 코드로 데이터베이스 연결을 한다.

from fastapi import FastAPI
from sqlalchemy import create_engine

# 예시: MySQL 연결
DATABASE_URL = "mysql+pymysql://username:password@host/db_name"

engine = create_engine(DATABASE_URL)

 

데이터베이스 연결은 SQLAlchemy 라이브러리의 create_engine() 함수를 사용한다. 이 함수는 데이터베이스 엔진을 생성하고, 이 엔진은 실제로 데이터베이스 작업을 수행할 때 사용된다.

from sqlalchemy import Column, Integer, String, Boolean, DateTime, Float, ForeignKey, create_engine
from datetime import datetime
from sqlalchemy.ext.declarative import declarative_base
from fastapi import FastAPI

app = FastAPI()

DATABASE_URL = "mysql+pymysql://root:1234@localhost/test.db"  # MySQL 데이터베이스 URL
engine = create_engine(DATABASE_URL)

# 베이스 클래스 생성
Base = declarative_base()

class User(Base):
    # 테이블 이름 설정
    __tablename__ = 'users'
    
    # id 필드드 설정
    id = Column(Integer, primary_key=True, autoincrement=True, comment="기본키")
    # username 필드 설정
    username = Column(String(50), unique=True, nullable=False, index=True, comment="사용자 이름")
    # 'eamil' 필드 설정
    email = Column(String(150), unique=True, nullable=False, index=True, comment="이메일")
    # is_active 필드 설정
    is_active = Column(Boolean, default=True, comment="활성화 여부")
    # created_at 필드 설정
    crated_at = Column(DateTime, default=datetime.utcnow, comment="생성일")
    # grade 필드 설정
    grade = Column(Float, comment="사용자 등급")
    # pared_id 필드 설정
    parent_id = Column(Integer, ForeignKey('parents.id'), comment="상위 ID")
    
# 데이터베이스 엔진을 사용하여 데이터베이스에 테이블 생성
Base.metadata.create_all(bind=engine) # 서버 시작 시 테이블에 해당 테이블이 없으면 새로 생성

@app.get("/")
def read_root():
    # 루트 경로로 HTTP GET 요청이 오면 메시지 반환
    return {"message": "Hello, FastAPI!"}

• Base.metadata.create_all(bind=engine): 해당 테이블이 없을 경우 자동으로 생성해주는 코드. 만약 기존 테이블이 있다면 아무런 변화 없음.

---

[참고]

가장 빠른 풀스택을 위한 플라스크 & FastAPI

SQLAlchemy 공식 사이트

---

다음 내용

[FastAPI] SQLAlchemy와 CRUD (Depends, db.query)