SQLAlchemy Core로 네이티브 SQL의 강력함 활용하기

소개

데이터 관리를 위한 Python이라는 활기찬 생태계에서 SQLAlchemy의 ORM 컴포넌트와 같은 객체 관계형 매퍼(ORM)는 익숙한 객체 지향 패러다임으로 데이터베이스 상호 작용을 추상화하는 데 엄청난 인기를 얻었습니다. 이러한 편리함은 종종 개발을 간소화하여 엔지니어가 복잡한 SQL 구문보다는 비즈니스 로직에 집중할 수 있도록 합니다. 그러나 ORM의 추상화가 유익하더라도 병목 현상이 될 수 있는 시나리오가 있습니다. 복잡한 쿼리, 성능에 민감한 작업 또는 특정 SQL 기능에 크게 의존하는 기존 데이터베이스와의 통합에 직면했을 때 ORM은 제한적이라고 느껴질 수 있습니다. 이때 SQLAlchemy Core가 필수 도구로 등장하여 원시 SQL 표현식과 데이터베이스에 독립적인 구문을 사용하여 Pythonic한 방식으로 데이터베이스와 상호 작용할 수 있는 강력하고 Pythonic한 방법을 제공합니다. SQLAlchemy Core를 이해하고 활용함으로써 개발자는 네이티브 SQL의 전체 성능을 발휘하여 최적의 성능과 최대의 유연성을 보장할 수 있습니다. 이 가이드에서는 ORM의 한계를 뛰어넘어 데이터베이스 상호 작용을 직접 제어할 수 있도록 지원하는 방법을 설명하면서 SQLAlchemy Core를 심층적으로 살펴보겠습니다.

SQLAlchemy Core의 핵심 개념

실질적인 예제를 살펴보기 전에 SQLAlchemy Core를 뒷받침하는 기본 개념을 명확하게 이해해 봅시다.

Engine: Engine은 모든 SQLAlchemy Core 상호 작용의 시작점입니다. 데이터베이스 연결 및 방언 정보를 나타내며 데이터베이스에 대한 기본 인터페이스 역할을 합니다. 연결 풀링, 트랜잭션 관리 및 방언별 SQL 생성을 처리합니다.
Connection: Engine이 설정되면 SQL 문을 실행하기 위해 Connection 객체를 가져옵니다. Connection은 데이터베이스와의 활성 세션을 나타냅니다.
MetaData: MetaData 객체는 Table 및 Column과 같은 스키마 객체의 컨테이너입니다. Pythonic 방식으로 데이터베이스 테이블의 구조를 정의하는 데 사용됩니다.
Table: Table 객체는 데이터베이스 테이블을 나타냅니다. MetaData 객체 내에서 정의되며 Column 객체로 구성됩니다. 각 Column은 이름, 데이터 유형 및 제약 조건을 포함하여 테이블의 열을 정의합니다.
Column: Column 객체는 Table 내의 특정 열을 나타냅니다. 열의 이름, 데이터 유형(예: String, Integer, DateTime) 및 다양한 제약 조건(예: PrimaryKey, ForeignKey, Nullable)을 지정합니다.
Selectable: Selectable은 Table 또는 select() 구문과 같이 쿼리 가능한 모든 객체를 참조합니다.
SQL Expression Language: 이것이 SQLAlchemy Core의 핵심입니다. SQL 구문과 유사하지만 Python의 유연성을 갖춘 객체와 연산자를 사용하여 SQL 문을 Pythonic 방식으로 구성하는 방법입니다. SELECT, INSERT, UPDATE, DELETE 문을 빌드하고 복잡한 조건, 조인 및 집계를 정의할 수 있습니다.

Core를 사용하여 쿼리 구축 및 실행

이러한 개념을 사용하여 일반적인 데이터베이스 작업을 수행하는 방법을 살펴보겠습니다.

먼저 Engine 및 MetaData를 설정해야 합니다. 시연 목적으로 인메모리 SQLite 데이터베이스를 사용합니다.

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, DateTime, ForeignKey, select, insert, update, delete, func
from datetime import datetime

# 1. Engine 생성
engine = create_engine("sqlite:///:memory:")

# 2. MetaData 정의
metadata = MetaData()

# 3. 테이블 정의
users = Table(
    "users",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String(50), nullable=False),
    Column("email", String(100), unique=True),
)

orders = Table(
    "orders",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("user_id", Integer, ForeignKey("users.id")),
    Column("item", String(100), nullable=False),
    Column("quantity", Integer, default=1),
    Column("order_date", DateTime, default=datetime.utcnow),
)

# 4. 데이터베이스에 테이블 생성
metadata.create_all(engine)

print("테이블이 성공적으로 생성되었습니다!")

데이터 삽입

insert() 구문을 사용하여 데이터를 삽입하는 것은 간단합니다.

# 단일 사용자 삽입
with engine.connect() as connection:
    stmt = insert(users).values(name="Alice", email="alice@example.com")
    result = connection.execute(stmt)
    print(f"ID가 {result.lastrowid}인 사용자가 삽입되었습니다.")
    connection.commit() # 쓰기 작업 시 커밋을 잊지 마세요!

# 여러 사용자 삽입
with engine.connect() as connection:
    stmt = insert(users)
    connection.execute(stmt, [
        {"name": "Bob", "email": "bob@example.com"},
        {"name": "Charlie", "email": "charlie@example.com"},
    ])
    connection.commit()
    print("여러 사용자가 삽입되었습니다.")

# 주문 삽입
with engine.connect() as connection:
    stmt = insert(orders)
    connection.execute(stmt, [
        {"user_id": 1, "item": "Laptop", "quantity": 1},
        {"user_id": 2, "item": "Mouse", "quantity": 2},
        {"user_id": 1, "item": "Keyboard", "quantity": 1, "order_date": datetime(2023, 10, 26)},
    ])
    connection.commit()
    print("주문이 삽입되었습니다.")

`select()`를 사용한 데이터 검색

select() 구문은 쿼리를 수행하는 핵심 도구입니다. 이를 통해 SELECT 문을 프로그래밍 방식으로 복잡하게 구성할 수 있습니다.

# users 테이블에서 모든 열 선택
with engine.connect() as connection:
    stmt = select(users)
    result = connection.execute(stmt)
    print("\n모든 사용자:")
    for row in result:
        print(row) # Row 객체는 튜플 및 사전처럼 작동합니다.

# 특정 열 선택
with engine.connect() as connection:
    stmt = select(users.c.name, users.c.email).where(users.c.id == 1)
    result = connection.execute(stmt)
    print("\nID가 1인 사용자 (이름 및 이메일):")
    for row in result:
        print(row)

# `where()`를 사용한 필터링
with engine.connect() as connection:
    stmt = select(users).where(users.c.name.startswith("A"))
    result = connection.execute(stmt)
    print("\n이름이 'A'로 시작하는 사용자:")
    for row in result:
        print(row)

# 결과 정렬
with engine.connect() as connection:
    stmt = select(users).order_by(users.c.name.desc())
    result = connection.execute(stmt)
    print("\n이름별로 정렬된 사용자 (내림차순):")
    for row in result:
        print(row)

# 결과 제한
with engine.connect() as connection:
    stmt = select(users).limit(2)
    result = connection.execute(stmt)
    print("\n처음 두 사용자:")
    for row in result:
        print(row)

# 테이블 조인
with engine.connect() as connection:
    stmt = select(users.c.name, orders.c.item, orders.c.quantity).
           join(orders, users.c.id == orders.c.user_id).
           where(orders.c.quantity > 1)
    result = connection.execute(stmt)
    print("\n항목을 두 개 이상 주문한 사용자:")
    for row in result:
        print(row)

# 집계 및 그룹화
with engine.connect() as connection:
    stmt = select(users.c.name, func.count(orders.c.id).label("total_orders")).
           join(orders, users.c.id == orders.c.user_id).
           group_by(users.c.name).
           order_by(func.count(orders.c.id).desc())
    result = connection.execute(stmt)
    print("\n사용자별 총 주문 횟수:")
    for row in result:
        print(row)

데이터 업데이트

update() 구문은 기존 레코드를 수정하는 데 사용됩니다.

with engine.connect() as connection:
    stmt = update(users).where(users.c.id == 1).values(email="alice.updated@example.com")
    connection.execute(stmt)
    connection.commit()
    print("\nID가 1인 사용자의 이메일이 업데이트되었습니다.")

# 업데이트 확인
with engine.connect() as connection:
    stmt = select(users.c.name, users.c.email).where(users.c.id == 1)
    result = connection.execute(stmt).fetchone()
    print(f"업데이트된 사용자 1: {result}")

데이터 삭제

delete() 구문은 테이블에서 레코드를 제거하는 데 사용됩니다.

with engine.connect() as connection:
    stmt = delete(orders).where(orders.c.quantity == 1)
    result = connection.execute(stmt)
    connection.commit()
    print(f"\n수량이 1인 주문 {result.rowcount}개가 삭제되었습니다.")

# 삭제 확인
with engine.connect() as connection:
    stmt = select(orders)
    result = connection.execute(stmt)
    print("\n남은 주문:")
    for row in result:
        print(row)

원시 SQL 실행

SQLAlchemy Core는 강력한 표현 언어를 제공하지만 완전히 원시 SQL을 실행해야 하는 경우가 있습니다. 이 또한 지원됩니다.

with engine.connect() as connection:
    # 예 1: DDL 문
    connection.execute("CREATE TABLE products (id INTEGER PRIMARY KEY, name TEXT)")
    connection.commit()
    print("\n원시 SQL: 'products' 테이블 생성.")

    # 예 2: DML 문
    connection.execute("INSERT INTO products (name) VALUES (?)", ("Gadget A",))
    connection.execute("INSERT INTO products (name) VALUES (?)", ("Gadget B",))
    connection.commit()
    print("원시 SQL: 제품 삽입.")

    # 예 3: DQL 문
    result = connection.execute("SELECT * FROM products")
    print("원시 SQL: 모든 제품:")
    for row in result:
        print(row)

애플리케이션 시나리오

SQLAlchemy Core는 몇 가지 주요 시나리오에서 빛을 발합니다.

성능 최적화: ORM이 최적화하기 어려운 매우 성능에 민감한 쿼리의 경우 Core를 통해 직접 SQL을 작성하는 것이 상당한 속도 향상을 가져오는 경우가 많습니다. 이는 복잡한 조인, 하위 쿼리 또는 분석에 특히 그렇습니다.
레거시 데이터베이스 통합: 복잡한 저장 프로시저, 사용자 지정 함수 또는 특정 SQL 방언 기능을 가진 기존 데이터베이스와 함께 작업할 때 Core는 ORM 추상화를 강요하지 않고 직접 상호 작용할 수 있는 유연성을 제공합니다.
데이터 마이그레이션 및 ETL: 데이터 마이그레이션 스크립트 또는 추출, 변환, 로드(ETL) 파이프라인에서 Core는 효율적인 배치 작업과 데이터 조작에 대한 정밀한 제어를 허용하여 이러한 작업에 ORM보다 더 적합한 경우가 많습니다.
스키마 관리: ORM은 스키마를 추론할 수 있지만 Core는 데이터베이스 스키마를 프로그래밍 방식으로 정의, 검사 및 관리할 수 있는 강력한 방법을 제공하며, 이는 배포 파이프라인이나 데이터베이스 버전 관리에 유용합니다.
SQL 학습: Core를 사용하는 것은 Pythonic 환경에서 SQL을 학습하고 연습하는 좋은 방법으로, 원시 SQL과 ORM 추상화 사이의 격차를 해소합니다.
복잡한 사용자 지정 보고서: 복잡한 집계, 창 함수 또는 UNION 작업을 포함하는 복잡한 보고서를 생성하는 것은 Core를 사용하면 종종 더 자연스럽고 성능이 뛰어납니다.

결론

SQLAlchemy Core는 ORM의 강력한 대안이자 보완품입니다. 풍부하고 Pythonic한 SQL 표현 언어를 제공함으로써 개발자는 애플리케이션 내에서 네이티브 SQL의 전체 성능을 활용할 수 있습니다. 성능 병목 현상, 복잡한 쿼리 요구 사항 또는 세분화된 데이터베이스 제어가 필요한 경우 SQLAlchemy Core를 채택하면 Python 데이터 상호 작용의 효율성, 유연성 및 유지 관리성을 크게 향상시킬 수 있습니다. 고수준 추상화와 원시 데이터베이스 성능 사이의 간극을 해소하여 데이터를 진정으로 마스터할 수 있도록 지원합니다.