[python] python의 ORM, SQLAlchemy

 

 SQLAlechemy

파이썬 코드에서 DB연결을 할 수 있도록 다양한 라이브러리를 지원하는데, 그중에서 널리 사용되는 라이브러리 중 하나로

파이썬 SQL 툴킷이자 객체 관계 매퍼(Object Relational Mapper, ORM)이다.

ORM이란 관계형 데이터베이스(RDBMS)의 테이블들을 프로그래밍 언어간의 불일치를 해소시키는데,

SQL을 잘 알지 못한다면 ORM을 먼저 사용하기보다, SQL을 먼저 공부해야한다.

 

• JAVA에서의 ORM도 함께 살펴보자.

2024.10.28 - [Development/Spring eGov4.0 (Java11, Tomcat9)] - Spring , 영속 계층의 프레임 워크, myBatis와 JPA차이 , ORM이란

Spring , 영속 계층의 프레임 워크, myBatis와 JPA차이 , ORM이란

 

1. SQLAlchemy 설치

pip 명령어를 활용하여 설치한다.

물론 가상 환경을 사용했다면, 해당 환경에 맞도록 가상 환경을 활성화 하는것을 잊지 말것.

# 가상환경이름 SQLAlchemy_venv 가정.

# 가상 환경 활성화 (Windows)
SQLAlchemy_venv\Scripts\activate
# 가상 환경 활성화 (macOS/Linux)
source SQLAlchemy_venv/bin/activate

# SQLAlchemy 설치
pip install sqlalchemy

# SQLite -> 별도 설치 필요 없음.

# mysql용 DBAPI
pip install mysql-connector-python

 

• SQLite 참고

2025.07.14 - [Development/Python] - [python] SQLite, sqlite3을 활용한 CRUD

[python] SQLite, sqlite3을 활용한 CRUD

 

 

2. 사용 예시

a. SQLite (테이블 생성 전)

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker

# 1. 데이터베이스 연결
# SQLite 데이터베이스 파일을 생성.
engine = create_engine('sqlite:///user.db')

# 2. ORM 기본 클래스 정의
# 모든 모델 클래스가 상속받을 기본 클래스.
Base = declarative_base()

# 3. 데이터베이스 테이블 모델 정의 (테이블과 매핑)
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String, unique=True)
    password = Column(String)

# 4. 모델을 기반으로 테이블 생성
Base.metadata.create_all(engine)

# 5. 세션 생성 및 데이터 추가
Session = sessionmaker(bind=engine)
session = Session()

# 새 사용자 객체 생성
user1 = User(name='Alice', email='alice@example.com', password='abc1234')
user2 = User(name='Bob', email='bob@example.com', password='1234abc')

# 세션에 객체 추가 (아직 데이터베이스에 저장되지 않음)
session.add(user1)
session.add(user2)

# 데이터베이스에 변경 사항 커밋
session.commit()

# 6. 데이터 조회
users = session.query(User).all()
for user in users:
    print(f"ID: {user.id}, Name: {user.name}, Email: {user.email}")

Base.metadata.create_all(engine)은 기본적으로 테이블이 존재하지 않을 때만 테이블을 생성한다.

이러한 동작은 "CREATE TABLE IF NOT EXISTS" SQL 문과 동일하며, 애플리케이션을 여러 번 실행해도 테이블이 중복으로 생성되지 않도록 해준다.

하지만 주의할 사항은 기존 테이블의 스키마(schema) 변경사항을 자동으로 반영하지는 않는다.

예를 들어, User 모델에 새로운 컬럼을 추가한 후 create_all()을 다시 실행해도 기존 테이블에는 새로운 컬럼이 추가되지 않는다.

 

b. mysql 사용시

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker

# 1. 데이터베이스 연결 설정
# MySQL 데이터베이스에 연결하기 위한 엔진을 생성한다.
engine = create_engine('mysql+mysqlconnector://your_user:your_password@your_host/your_database')

# 2. ORM 기본 클래스 정의
Base = declarative_base()

# 3. 데이터베이스 테이블 모델 정의 
class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50))  # VARCHAR(50)과 매핑.
    email = Column(String(100), unique=True) # VARCHAR(100)과 매핑되며, 고유 값.
    password = Column(String(255)) # 암호화된 비밀번호를 저장하기 위한 컬럼.

# 4. 모델을 기반으로 테이블 생성
# 이미 테이블이 존재하면 아무런 동작을 하지 않음.
Base.metadata.create_all(engine)

# 5. 세션 생성 및 데이터 추가
Session = sessionmaker(bind=engine)
session = Session()

# 새 사용자 객체를 생성.
user1 = User(name='Alice', email='alice@example.com', password='abc1234')
user2 = User(name='Bob', email='bob@example.com', password='1234abc')

# 생성한 객체를 세션에 추가. 미반영 상태.
session.add(user1)
session.add(user2)

# 세션의 변경사항을 데이터베이스에 반영(커밋).
session.commit()

# 6. 데이터 조회
users = session.query(User).all()
for user in users:
    print(f"ID: {user.id}, 이름: {user.name}, 이메일: {user.email}")

SQLite와 create_engin의 방법이 다를 뿐 SQLAlchemy의 기본 동작은 동일하다는 것을 알 수 있다.

 

3. SQLAlchemy의 CRUD

a. SELECT ALL

users = session.query(User).all()
# 모든 사용자를 리스트 형태로 반환.

 

b. SELECT ONE

user = session.query(User).filter_by(name='Bob').first()
# 이름이 'Bob'인 첫 번째 사용자를 반환.

데이터베이스에서 조회된 정보가 User 클래스의 객체(instance) 형태로 반환된다.

user_to_update = session.query(User).get(1)
# ID가 1인 사용자 조회

get() 메서드는 인자로 받은 값이 해당 테이블의 Primary Key(기본 키) 값이라고 가정하고 데이터를 조회한다.

만약 PK가 아닌 것을 찾기 위해 filter_by(column=value).first()를 사용할 수 있다.

 

c. SELECT ... WHERE ...

users = session.query(User).filter(User.id > 1).all()
# ID가 1보다 큰 모든 사용자를 조회.

 

d. INSERT INTO

# 새 User 객체 생성
new_user = User(name='Charlie', email='charlie@example.com', password='asdf1234')

# 세션에 객체 추가
session.add(new_user)

# 변경사항을 데이터베이스에 커밋
session.commit()

 

e. UPDATE SET

# ID가 1인 사용자 조회
user_to_update = session.query(User).get(1)

if user_to_update:
    # 이메일 주소 변경
    user_to_update.email = 'alice_updated@example.com'
    
    # 변경사항 커밋
    session.commit()

 

f. DELETE

# ID가 2인 사용자 조회
user_to_delete = session.query(User).get(2)

if user_to_delete:
    # 세션에서 객체 삭제
    session.delete(user_to_delete)
    
    # 변경사항 커밋
    session.commit()

1. 데이터를 삭제할 때는 삭제할 객체를 조회한 후,
2. session.delete() 메서드로 세션에서 삭제하고,
3. session.commit()으로 데이터베이스에서 최종적으로 제거한다.

4. SQLAlchemy의 다양한 쿼리

우선 다양한 쿼리를 확인하기 위해 테이블 두개를 준비한다.

해당 테이블은 userid를 참조하여 FK를 생성하고 두 테이블간의 관계를 묘사한다.

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import declarative_base, sessionmaker, relationship

# 데이터베이스 설정
engine = create_engine('sqlite:///User2.db')
Base = declarative_base()

# User와 Post 모델
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    password = Column(String)
    # relationship()은 SQLAlchemy에게 Post 모델과 어떻게 연결되는지 알려준다.
    posts = relationship("Post", back_populates="user")

class Post(Base):
    __tablename__ = 'posts'
    id = Column(Integer, primary_key=True)
    title = Column(String)
    content = Column(String)
    # ForeignKey는 이 컬럼이 'users' 테이블의 'id' 컬럼과 연결되어 있음을 명시.
    user_id = Column(Integer, ForeignKey('users.id'))
    # 이 relationship은 User 모델로 다시 연결된다.
    user = relationship("User", back_populates="posts")

Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

# 예제 데이터 추가
user1 = User(name='Alice', password='123')
post1 = Post(title='My First Post', content='Hello, World!', user=user1)
post2 = Post(title='My Second Post', content='Another post!', user=user1)
user2 = User(name='Bob', password='345')
post3 = Post(title='My First Post', content="Hello, I'm Bob!", user=user2)
post4 = Post(title='My Second Post', content='Nice ~', user=user2)

session.add_all([user1, post1, post2, user2, post3, post4])
session.commit()

 

a. JOIN ON

# User와 Post 테이블을 조인하여 모든 게시글과 작성자 이름을 가져오기.
result = session.query(User.name, Post.title).join(Post).all()

for user_name, post_title in result:
    print(f"사용자: {user_name}, 게시글: {post_title}")

SELECT users.name, posts.title
FROM users JOIN posts ON users.id = posts.user_id

사용자: Alice, 게시글: My First Post
사용자: Alice, 게시글: My Second Post
사용자: Bob, 게시글: My First Post
사용자: Bob, 게시글: My Second Post

 

c. JOIN ON WHERE

from sqlalchemy import and_

# 이름이 'Alice'이고, 게시글 제목에 'Hello, World!'가 포함된 사용자를 찾기.
result = session.query(User).join(Post).filter(
    and_(
        User.name == 'Alice',
        Post. content.like('%Hello%')
    )
).first()

if result:
    print(f"사용자 발견: {result.name}")

SELECT users.id, users.name, users.password, posts.id, posts.title, posts.content
FROM users JOIN posts ON users.id = posts.user_id 
WHERE users.name = 'Alice' AND posts.title LIKE '%Hello%'
LIMIT 1

사용자 발견: Alice

 

d. ORDER BY

# 모든 게시글을 내용을 기준으로 내림차순으로 정렬.
posts = session.query(Post).order_by(Post.content.desc()).all()

for post in posts:
    print(f"[content.desc] : {post.user.name} : {post.content}")

SELECT posts.id, posts.title, posts.content, posts.user_id, users.id AS users_id, users.name AS users_name, users.password AS users_password 
FROM posts 
JOIN users ON users.id = posts.user_id 
ORDER BY posts.content DESC

[content.desc] : Bob : Nice ~
[content.desc] : Alice : Hello, World!
[content.desc] : Bob : Hello, I'm Bob!
[content.desc] : Alice : Another post!

posts.user.name에 접근하는 방식은 지연 로딩(Lazy Loading)이라는 기술을 사용한다.

코드가 post.user.name에 접근하는 순간, SQLAlchemy는 post 객체의 user 속성이 비어있다는 것을 감지하고 쿼리를 실행시켜 user에 채워 넣게 된다.복잡한 조인 쿼리를 직접 작성할 필요 없이, 객체 지향적인 방식으로 편리하게 데이터를 다룰 수 있다.

* join() 과 relationship의 차이

1. relationship
객체 간의 연결 관계를 정의하는 데 사용한다. 이를 통해 post.user와 같이 객체 속성으로 데이터를 참조할 수 있다.
지연 로딩을 사용하므로, 필요한 시점에만 쿼리가 실행되어 편리하다.

2. join()
명시적으로 조인을 수행하고 원하는 컬럼을 미리 한 번에 가져오는 데 사용한다.
이는 여러 게시글에 대한 사용자의 이름을 반복해서 가져와야 할 때 지연 로딩보다 효율적일 수 있다.
매번 쿼리를 실행하는 대신, 한 번의 쿼리로 모든 데이터를 가져오기 때문이다.

 

e. GROUP BY

from sqlalchemy import func

# 각 사용자의 게시글 수를 가져온다.
# 사용자 이름별로 그룹화한다.
result = session.query(User.name, func.count(Post.id)).join(Post).group_by(User.name).all()

for user_name, post_count in result:
    print(f"사용자: {user_name}, 게시글 수: {post_count}")

SELECT users.name, count(posts.id)
FROM users
JOIN posts ON users.id = posts.user_id
GROUP BY users.name

사용자: Alice, 게시글 수: 2
사용자: Bob, 게시글 수: 2

 

 

5. 테스트 환경

위의 SQLite의 경우 Colab을 사용할 수 있다.

SQLAlchemy도 마찬가지로 이용할 수 있으므로, Colab이 사용 가능한 환경이라면 쉽게 테스트가 가능하니,

노트북이나 데스크탑 환경이 아니고 패드여도 사용 가능하다.