[Python] SQLAlchemy

SQLAlchemy는 파이썬 환경에서 가장 많이 사용되는 ORM 라이브러리다.

기본적인 구성 방법과 CRUD 쏘는 방법 정도만 정리해보겠다.

설치하기

알케미를 쓰려면 모듈 2개가 필요하다.
깔아준다.

pip install sqlalchemy psycopg2

됐으면 프로젝트 구조를 잡아보자.
나는 다음과 같이 알케미 관리용 서브모듈 "models"를 두고, 공통적인 기능을 base에 두고 테이블별로 별도의 소스파일을 두는 형태를 사용했다.

기본 연결 설정

base.py를 먼저 보겠다.
여기서 데이터베이스 연결이나, 그와 관련된 기본적인 요소들을 다 넣어놨다.

from sqlalchemy.orm import declarative_base

from sqlalchemy import create_engine, MetaData
from sqlalchemy.orm import sessionmaker, scoped_session

engine = create_engine("postgresql://postgres:패스워드@localhost:5432/alchemy_test", echo=True)
get_db_session = scoped_session(sessionmaker(autocommit=False, autoflush=False, bind=engine))

Base = declarative_base()

def sync_db():
    Base.metadata.create_all(bind=engine)

get_db_session는 session을 가져와주는 함수인데, DB 커넥션 개념이라 생각하면 된다.
Base는 엔티티를 정의할때 사용할 기반 클래스다. 이걸 상속받는 것으로 이후에 정의할 모든 엔티티가 하나로 제어된다.

그리고 테이블 초기화용 함수 sync_db도 하나 정의해뒀다. 정의한 엔티티를 create table해주는 기능이다.
여기서 바로 하지 않고 함수로만 둔 것은, 엔티티가 전부 import를 통해서 load된 시점에만 유효하게 동작하기 때문이다. 저건 최상위 모듈에서 실행하겠다.

엔티티 정의

테이블은 이런식으로 정의하면 된다.
Base를 상속받고, __tablename__으로 테이블명을 정의한다.
그리고 각각의 컬럼들을 필드로 정의해서 Column 객체를 넣어주면 된다.

from sqlalchemy import String, Integer, Column
from .base import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    name = Column(String(30), nullable=True)
    age = Column(Integer, nullable=False, default=0)

테이블 동기화하기

엔티티를 정의헀으니, 그게 실제 테이블로 생겨나도록 동기화를 해보자.
import해온 상태에서 sync_db를 해주면

주르륵 뜨면서

이렇게 만들어준다.

데이터 삽입하기 (add, add_all)

이제는 만들어놨던걸 기반으로 CRUD 작업을 하나씩 해보겠다.

세션을 가져와서 이래저래 객체를 생성하고, add로 추가하고, 커밋치고, 세션을 닫으면 된다.

from models import user, base

def main(): 
    db_session = base.get_db_session()

    new_user = user.User(name="John Doe", age=555)

    db_session.add(new_user)
    db_session.commit()

    db_session.close()

main()

커밋을 쳐야먄 진짜 커밋이 들어간다는 점 잊지 말자.
세션도 꼭 닫아주자.

아무튼 저렇게 실행하면

기대한대로 들어갈 것이다.

배열로 만들어서 add_all을 하면 여러개를 한번에 넣을 수 있다.

from models import user, base

def main(): 
    db_session = base.get_db_session()

    users = [
        user.User(name="John Doe", age=555), 
        user.User(name="Phone Doe", age=666), 
        user.User(name="Alexander", age=444), 
        user.User(name="Jane Doe", age=10)
    ]

    db_session.add_all(users)
    db_session.commit()

    db_session.close()

main()

잘 들어갔다.

데이터 조회하기

조회를 할때는 주로 query 함수를 사용해서 쿼리를 생성한다.

아래는 id가 1인 객체만 하나 찾아오는 간단한 예제다.

from models import user, base
from sqlalchemy.engine.row import Row

def main(): 
    db_session = base.get_db_session()

    result = db_session.query(user.User).filter_by(id = 1).first()
    print(result.id, result.name, result.age)

    db_session.close()

main()

여기서 유의할 점은, query나 filter_by 같은건 그냥 쿼리 텍스트만 만들어주는 역할이란 것이다.
저런 메서드 체이닝을 통해서 생성한 쿼리 객체를, first나 all 등의 함수로 실행시키고, 그 결과를 받아온다.

all을 사용하면 여러개를 리스트로 가져올 수 있다.
아래 코드는 age가 555, 666, 777인 데이터를 전부 select해오는 코드다.

from models import user, base
from sqlalchemy.engine.row import Row

def main(): 
    db_session = base.get_db_session()

    result = db_session.query(user.User) \
        .where(user.User.age.in_([555, 666, 777])) \
        .all()

    for e in result: 
        print(e.id, e.name, e.age)

    db_session.close()

main()

잘 가져온다.

데이터 수정하기 (update)

데이터에 update를 치는것에는 크게 2가지 방법이 있다.

정직하게 update 함수를 호출해서 처리하는 방법이 있고

from models import user, base
from sqlalchemy.engine.row import Row

def main(): 
    db_session = base.get_db_session()

    result = db_session.query(user.User) \
        .where(user.User.id.__eq__(1)) \
        .update({"name": "FooBar"})

    db_session.commit()

    db_session.close()

main()

select로 가져온 값에서 필드만 조작해서 얼렁뚱땅 수정되게 하는 방법이 있다.

from models import user, base
from sqlalchemy.engine.row import Row

def main(): 
    db_session = base.get_db_session()

    found = db_session.query(user.User) \
        .where(user.User.id.__eq__(3)) \
        .first()

    found.name = "asdf" // 값 변경!

    db_session.commit() // DB에 반영!

    db_session.close()

main()

데이터 삭제하기 (delete)

이번에는 데이터를 하나 지워보자.
update 함수를 호출하는 것과 별반 다를 것은 없다.

from models import user, base
from sqlalchemy.engine.row import Row

def main(): 
    db_session = base.get_db_session()

    result = db_session.query(user.User) \
        .where(user.User.id.__eq__(2)) \
        .delete()

    db_session.commit()

    db_session.close()

main()

해서 실행시키면

잘 사라진다.

참조
https://www.sqlalchemy.org/
https://blog.neonkid.xyz/253