[Python] SQLalchemy ORM

1. 서론

회사 백오피스 서비스 개발에서 뉴스기사 크롤링 모듈 개발을 담당 하였다.

크롤링 테스트 코드를 작성하고 비즈니스 로직을 작성하려고 하니, 막상 'DB 에 어떻게 쌓아야 하지?' 라는 생각이 들었다. 

주력 프레임워크인 Springboot 에선 JPA + QueryDSL 을 사용하였는데, Flask 프레임워크 기반인 Python 프로젝트에는 없을까?

이러한 이유로 SQLalchemy 라는 패키지를 알게되었고, 이에 대해 기술한 내용이다.

 

버전

* Python: v3.13 
* SQLAlchemy: v1.3
* PostgreSQL: v17

 

2. 본론

SQLAlchemy 는 SQL의 모든 기능과 유연성을 제공하는 Python 기반 SQL 툴킷이자 객체 관계형 매퍼이다.

장점은 여타 ORM 과 마찬가지로 sql 쿼리를 직접 작성하지 않아도 로직 내에서 테이블을 조작할 수 있다는 것이다. 

 

2.1. 패키지 설치

다음과 같은 명령어로 설치 가능하다.

pip install sqlalchemy

 

2.2. 모델 선언

모델단에선 데이터베이스에서 쿼리할 구조를 형성하는 모듈 수준의 구성 요소를 정의한다.

예시로 모델을 정의하면 다음과 같다.

 

Model.py

from sqlalchemy import Column, Integer, String, ForeignKey, DateTime, Date, LargeBinary, Boolean, BigInteger, inspect, func, create_engine
from datetime import datetime, date
from sqlalchemy.orm import relationship, declarative_base, sessionmaker
from src.conf.config import DATABASE_URL

Base = declarative_base()

"""언론사 테이블

   Note
   ----
   크롤링 언론사 링크 룩업 테이블
"""
class Company(Base):
    __tablename__ = "company"
    cp_id = Column(BigInteger, primary_key=True, autoincrement=True)
    cp_name = Column(String(255)) # 언론사 명
    cp_url = Column(String(1000)) # 검색 url
    cg_id = Column(BigInteger, ForeignKey("category.cg_id")) # 카테고리 테이블 FK
    category = relationship("Category")

    cp_created_date = Column(DateTime(timezone=True), server_default=func.now()) # 데이터 저장 일자
    cp_modified_date = Column(DateTime(timezone=True), onupdate=func.now()) # 데이터 저장 일자
    
Base.metadata.create_all(engine)

 

2.2.1. Base

'선언적 매핑' 구조에서 가장 중요한 클래스이다.

declarative_base() 함수를 통해 인스턴스 선언이 가능하고, 인스턴스 생성시 MetaData 라는 객체가 포함된다.

이 객체에는 테이블 이름, 컬럼 타입, 제약 조건 등 DB 스키마에 대한 모든 정보를 담을 수 있다. 이러한 이유로, Base.create_all() 함수 호출시 실제 DB 테이블을 생성할 수 있다.

 

2.2.2. __tablename__

실제 DB 테이블에 매핑되는 테이블 명이다.

 

2.2.3. Column

테이블의 필드는 Column() 을 통해 선언이 가능하다.

실제 클래스를 뜯어보면 들어가는 파라미터로 이름, 데이터 유형, 제약조건을 조작할 수 있다.

대표적으로 사용하는 name (필드명), primary_key (PK 정의), autoincrement(ID 자동 생성) 이 있다.

class Column(DialectKWArgs, SchemaItem, ColumnClause[_T]):
    """Represents a column in a database table."""

    __visit_name__ = "column"

    inherit_cache = True
    key: str

    server_default: Optional[FetchedValue]

    def __init__(
        self,
        __name_pos: Optional[
            Union[str, _TypeEngineArgument[_T], SchemaEventTarget]
        ] = None,
        __type_pos: Optional[
            Union[_TypeEngineArgument[_T], SchemaEventTarget]
        ] = None,
        *args: SchemaEventTarget,
        name: Optional[str] = None,
        type_: Optional[_TypeEngineArgument[_T]] = None,
        autoincrement: _AutoIncrementType = "auto",
        default: Optional[Any] = _NoArg.NO_ARG,
        insert_default: Optional[Any] = _NoArg.NO_ARG,
        doc: Optional[str] = None,
        key: Optional[str] = None,
        index: Optional[bool] = None,
        unique: Optional[bool] = None,
        info: Optional[_InfoType] = None,
        nullable: Optional[
            Union[bool, Literal[SchemaConst.NULL_UNSPECIFIED]]
        ] = SchemaConst.NULL_UNSPECIFIED,
        onupdate: Optional[Any] = None,
        primary_key: bool = False,
        server_default: Optional[_ServerDefaultArgument] = None,
        server_onupdate: Optional[_ServerOnUpdateArgument] = None,
        quote: Optional[bool] = None,
        system: bool = False,
        comment: Optional[str] = None,
        insert_sentinel: bool = False,
        _omit_from_statements: bool = False,
        _proxies: Optional[Any] = None,
        **dialect_kwargs: Any,
    ):

 

2.2.4 매핑

Company (언론사) 와 News (뉴스 기사) 테이블 간 1:N 매핑관계를 갖는다고 가정한다.

 

주인 입장에서 ForeignKey 를 지정해야, 테이블 간에 물리적인 매핑이 생성된다.

from sqlalchemy import Column, Integer, String, ForeignKey, DateTime, Date, LargeBinary, Boolean, BigInteger, inspect, func, create_engine
from datetime import datetime, date
from sqlalchemy.orm import relationship, declarative_base, sessionmaker
from src.conf.config import DATABASE_URL

"""언론사 테이블

   Note
   ----
   크롤링 언론사 링크 룩업 테이블
"""
class Company(Base):
    __tablename__ = "company"
    cp_id = Column(BigInteger, primary_key=True, autoincrement=True)
    cp_name = Column(String(255)) # 언론사 명
    cp_url = Column(String(1000)) # 검색 url
    cg_id = Column(BigInteger, ForeignKey("category.cg_id")) # 카테고리 테이블 FK
    category = relationship("Category")

    cp_created_date = Column(DateTime(timezone=True), server_default=func.now()) # 데이터 저장 일자
    cp_modified_date = Column(DateTime(timezone=True), onupdate=func.now()) # 데이터 저장 일자
    
    
"""뉴스 테이블

   Note
   ----
   크롤링 뉴스 저장 테이블
"""
class News(Base):
    __tablename__ = "news"
    news_id = Column(BigInteger, primary_key=True, autoincrement=True)

    cg_id = Column(BigInteger, ForeignKey("category.cg_id")) # 카테고리 테이블 FK
    category = relationship("Category")

    cp_id = Column(BigInteger, ForeignKey("company.cp_id")) # 언론사 테이블 FK
    company = relationship("Company")

    news_title = Column(String(1000)) # 뉴스 제목
    news_link = Column(String(1000)) # 뉴스 원문 링크
    news_img = Column(String(255)) # 뉴스 썸네일
    news_contents = Column(String (1000)) # 뉴스 본문
    news_published_date = Column(String(255)) # 뉴스 발행 일자
    news_created_date = Column(DateTime(timezone=True), server_default=func.now()) # 데이터 저장 일자
    news_modified_date = Column(DateTime(timezone=True), onupdate=func.now()) # 데이터 수정 일자

    def toDict(self):
        result = {}
        for c in inspect(self).mapper.column_attrs:
            value = getattr(self, c.key)

            if isinstance(value, (datetime, date)):
                result[c.key] = value.isoformat()
            else:
                result[c.key] = value

        result["category_name"] = self.category.cg_name if self.category else None
        result["company_name"] = self.company.cp_name if self.company else None

        return result

 

이때 relationship() 이 Springboot JPA 에서 없는 개념이라 조금 생소하였다.

 

차이를 보자면 다음과 같다.

- Column (cp_id): 실제 DB 테이블의 값이다. ex. News.cp_id = 10 으로 나올것이다.
- relationship (company): 파이썬 객체 세상에서의 연결고리. 숫자가 아니라 실제 Company 객체 자체 가리킨다.

 

만약 cp_id 만 있을때, 특정 뉴스의 언론사 이름을 가져오고 싶다면 로직이 길어진다.

# 뉴스를 가져온 뒤, 다시 DB를 뒤져서 언론사를 찾아야 함
news = session.query(News).first()
company = session.query(Company).filter(Company.cp_id == news.cp_id).first()
print(company.cp_name)

 

relationship 이 있다면 객체로 바로 접근 가능하기 때문에 위와 같은 과정을 겪지 않아도 된다.

# 객체 타고 바로 접근 
news = session.query(News).first()
print(news.company.cp_name)

 

이러한 이유로, relationship() 은 매핑관계에서 좀 더 쉽게 조회하기 위해 사용한다 라고 이해하였다.

 

2.3. DB 연결

DB 에 매핑될 테이블을 정의했으니, 실제로 DB 에 연결하여 정의된 내용을 전달해야 한다. 

DB 에 연결하기 위해선 create_engine() 명령어를 사용한다.

engine = create_engine("postgresql://postgres:esbgw123!@127.0.0.1/ibl_rookie", echo=True)

Base.metadata.create_all(engine) # 테이블 생성

 

데이터베이스 연결 후 조작을 할 때 session 이 필요하다.

session 이 SQLAlchemy 공식문서에서 비유하기론 사용자의 접시이고, 변경사항(객체들) 이 음식들 이라고 한다.

session 에 update 할 테이블, 값 이나 delete 할 값 등을 넣고 커밋하면서 데이터 베이스를 조작하는 컨셉이다.

 

세션은 이전에 생성한 엔진을 기반으로 생성한다.

SessionLocal = sessionmaker(bind=engine)

 

2.4. CRUD 메서드

실제 공식문서를 보면 해당 메서드가 SQL 로 어떤 구문인지 비교하여 기술해 두어 읽기 편하다.

 

2.4.1. CREATE

추가하고자 하는 테이블에 매핑된 class 인스턴스를 생성하여 session 에 추가 후 commit 시 변경사항이 반영된다.

session.add({추가하고 하는 Class})
session.commit()

 

실제 사용 예시는 다음과 같다.

""" 뉴스 내용 저장 메서드
   Parameters
   ----------
   result (dictionary)
       크롤링 후 정제 딕셔너리
       
   cg_id (BigInteger)
       뉴스 카테고리 id
       
   cp_id (BigInteger)
       언론사 id
       
   Returns
   --------
   None

   Raises
   -------
   AttributeError
       save error 시 orm error 출력
"""
def news_save(result: dict, cg_id, cp_id):
    new_article = News(
        cg_id = cg_id,
        cp_id = cp_id,
        news_title = result["title"],
        news_link = result["link"],
        news_img = result["img"],
        news_contents = result["contents"],
        news_published_date = result["published_date"]
    )

    try:
        orm.add(new_article)
        orm.commit()
    except Exception as e:
        print(f"fail to save data! error : {e}")
        orm.rollback()

 

2.4.2. READ

조회는 오히려 QueryDSL 구문과 유사하여 편했다. 자주 사용하는 메서드만 정리해보았다.

# 1. 전체 조회
all_news = session.query(News).all()

# 2. 조건 조회
news = session.query(News).filter(News.news_id == 1).first()

# 3. 조인 조회 (이미 설정한 relationship 활용)
results = session.query(News).join(Company).filter(Company.cp_name == "전자신문").all()

 

2.4.3. UPDATE

update 는 JPA 와 유사하게 따로 메서드가 존재하지 않고 조회 후 add 및 commit 한다.

 

2.4.4. DELETE

삭제는 delete() 메서드를 사용 후 commit 한다.

# 먼저 삭제할 대상을 조회
target_news = session.query(News).filter(News.news_id == 123).first()

if target_news:
    # 세션에서 삭제 표시
    session.delete(target_news)
    
    # 트랜잭션 커밋
    session.commit()
    logging.info(f"뉴스 삭제 완료")

 

3. 참고자료

https://docs.sqlalchemy.org/en/13/orm/tutorial.html