1. 개요
지난 포스팅에서 파이썬을 활용하여 업무 자동화에 대해 정리했었다.
https://soonmin.tistory.com/82
[fabric] fabirc 사용한 업무 자동화 구축
1. 개요 프로세스를 개발하고, 서버에 배포한 뒤 정상적으로 동작하는지 확인이 필요하다. 나는 매일 로컬환경에서 다수의 배포서버에 SSH로 접속해서 로그파일들을 확인해야 했다. 반복 작업을
soonmin.tistory.com
이번에는 자동화 구축된 결과를 파일저장, 메일전송뿐만 아니라 로컬 db(sqlite)로도 관리하고 싶어, 파이썬에서 DB와 상호 작용하기 위한 기술들을 찾아보았다. 다양한 DB(PostgreSQL, sqlite 등)에 호환되어야 하므로 ORM 기반의 SQLAlchemy를 사용하기로 결정했다.
나는 모델(테이블)도 하나만 사용하기 때문에 모델 간의 관계에 대해서 다루지 않았으며, 기본적인 것들만 예시와 함께 정리할 것이다. 조인을 사용해야 할 상황이 오면 정리하겠다.
SQLAlchemy에 대해 자세히 학습하고 싶으면 공식문서나 다른 포스팅을 참고해주세요.
SQLAlchemy
The Database Toolkit for Python
www.sqlalchemy.org
2. SQLAlchemy이란?
- 파이썬에서 사용하는 여러 ORM(Object-Relational-Mapping) 중 하나이다. (Java에서는 대표적으로 JPA)
- ORM은 쿼리문을 직접 작성하지 않고, 객체와 RDBMS를 매핑하여, 쿼리를 생성한다.
- ORM의 장점은 객체지향적으로 RDBMS와 상호작용할 수 있다는 점이다.
- DB에 종속적이지 않다는 점은 장점이면서 단점이 될 수 있다.
- 장점: 다양한 DB 마이그레이션이 가능하다.
- 단점: DB 고유 기능을 사용하기 어렵다. => 고유 기능이 사용 가능 하다면 마이그레이션이 어려움.
3. SQLAlchemy 사용하기
a. 라이브러리 설치
- pip install로 sqlalchemy를 설치한다.
pip install sqlalchemy
b. DB 정의 및 세션 생성
- create_engine으로 db url 접속 정보를 넣어준다.
- echo를 True로 설정하면 콘솔에 실행 쿼리를 출력시킨다.
- sessionmaker으로 db 세션을 설정한다.
- autocommit=Fasle: 자동커밋 비활성화
- autoflush=False: 자동 flush 비활성
- bind로 create_engine으로 생성한 db 엔진을 바인딩
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
# url = 'postgresql://postgres:1234@127.0.0.1:5433/test' # postgreSql
url = 'sqlite:///./test.db' #sqlite
engine = create_engine(url, echo=True)
session_local = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def get_db_session():
session = session_local()
try:
return session
finally:
session.close()
c. 모델 정의 및 DDL 생성
- declarative_base는 테이블을 정의하기 위한 base 클래스를 생성하는 데 사용(모든 모델 객체들은 Base를 상속한다.)
- User라는 모델을 생성하고, 테이블 이름과 필드(컬럼)을 선언.
- id는 auto_increment로 설정.
- Base.metadata.create_all(engine)은 모델로 정의된 모든 테이블을 db에 생성. (존재하면 생성 x)
Base = declarative_base()
class User(Base):
__tablename__ = "user"
id = Column(Integer, primary_key=True, unique=True, autoincrement=True)
user_id = Column(String(50))
user_pw = Column(String(50))
name = Column(String(50))
date = Column(DateTime)
def __init__(self, user_id, user_pw, name, date):
self.user_id = user_id
self.user_pw = user_pw
self.name = name
self.date = date
Base.metadata.create_all(engine)
d. CRUD 예제
select
- filter_by(), filter()로 조건을 걸어줄 수 있다.
- all() 은 조건에 해당하는 모든 데이터를 조회한다.
- first()는 조건에 해당하는 데이터 하나를 조회한다.
result1 = session.query(User).filter_by(name='test_user').all()
for user in result1:
print(f'user_id: {user.user_id}')
print(f'user_pw: {user.user_pw}')
print(f'name: {user.name}')
print(f'date: {user.date}')
result2 = session.query(User).filter_by(id='3').first()
print(f'user_id: {result2.user_id}')
print(f'user_pw: {result2.user_pw}')
print(f'name: {result2.name}')
print(f'date: {result2.date}')
result3 = session.query(User).filter_by(name='test_user', user_id='user1').first()
print(f'user_id: {result3.user_id}')
print(f'user_pw: {result3.user_pw}')
print(f'name: {result3.name}')
print(f'date: {result3.date}')
insert
- db session을 가져와 모델 객체를 add 한 뒤 commit만 호출하면 된다.
session = get_db_session()
user1 = User(user_id='user1', user_pw='pw1', name='test_user', date=datetime.now())
session.add(user1)
session.commit()
bulk insert
- 대량의 데이터를 한 번에 삽입할 때는 bulk_save_objects를 사용할 수 있다.
- 대량의 데이터를 삽입할 때는 add()나 add_all()을 사용하는 것보다 성능에 좋다.
- ORM 자체가 bulk insert에 최선의 방법은 아니다. 성능이 중요하다면 core를 사용해서 bulk 작업을 진행하자.
session = get_db_session()
user_list = []
user1 = User(user_id='user1', user_pw='pw1', name='test_user', date=datetime.now())
user2 = User(user_id='user1', user_pw='pw1', name='test_user', date=datetime.now())
user_list.append(user1)
user_list.append(user2)
session.bulk_save_objects(user_list)
session.commit()
update
- 업데이트할 대상을 조회한 뒤 객체의 속성을 바꾼 뒤, session을 커밋하면 update가 완료된다.
session = get_db_session()
user_to_update = session.query(User).filter_by(id='1').first()
if user_to_update:
# 사용자 정보 업데이트
user_to_update.user_pw = 'new_password'
user_to_update.name = 'updated_name'
user_to_update.date = datetime.now()
# 변경 내용을 세션에 반영
session.commit()
else:
print("사용자를 찾을 수 없습니다.")
delete
- filter_by로 조건을 걸고 delete()를 호출하여 삭제를 진행한다.
- 삭제 대상이 없으면 0을 반환한다.
session = get_db_session()
result = session.query(User).filter_by(id='2').delete()
if result:
session.commit()
print("사용자를 삭제했습니다.")
else:
print("삭제할 사용자를 찾을 수 없습니다.")
4. 마무리
지금까지 sqlalchemy에 대해서 정리해 보았다. 하나의 모델로 간단한 예제를 구현했기 때문에 이해하기에는 큰 어려움은 없을 것이다. 추후 테이블이 추가되거나 조인을 사용할 기회가 생긴다면 조인 쿼리도 정리해 보겠다.
