[SQLAlchemy] Raw SQL

목차

 

개요 

SQLALchemy는 ORM과 많이 밀접하다. 그러나 가끔씩 Raw SQL을 사용해야 될 때가 있다.

 

이 글은 SQLALchemy 2.0에서 Raw SQL 사용예제를 모아둔 글이다.

 

1.  Setup

Raw SQL을 작성하기에 앞서 간단한 테이블을 만들자.

CREATE TABLE `member` (
  `id` int(1) unsigned NOT NULL AUTO_INCREMENT,
  `random_id` char(36) NOT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8

아래는 이 테이블에 매핑되는 class이다. 어떤 의미로써는 Entity이다.

@dataclasses.dataclass
class Member:
    id: int = dataclasses.field(init=False, default=None)
    random_id: uuid.UUID

    def to_dict(self):
        return dataclasses.asdict(self)

 

2.  SQLALChemy 1.x 버전과  2.0의 Raw SQL 차이점

SQLALchemy 1.x 버전에서는 Raw SQL을 사용하려면 문자열 쿼리를 만들어 사용하면 된다. 하지만 SQLAlChemy 2.0에선 "text"라는 함수를 사용해야 한다.

from sqlalchemy import text
query = conn.execute(text("select * from member"))

참조 : https://docs.sqlalchemy.org/en/20/core/sqlelement.html#sqlalchemy.sql.expression.TextClause

Column Elements and Expressions — SQLAlchemy 2.0 Documentation

 

3. Raw SQL 반환 : CursorResult와 MappingResult

Raw SQL을 실행한 뒤 얻은 결과는 CursorResult 객체다.

with engine.connect() as conn:
    query = conn.execute(text("select * from member"))
    print(type(query))
    

# OutPut
<class 'sqlalchemy.engine.cursor.CursorResult'>

이는 반환 결과가 tuple 형식이다.

[(13, '9b74f5ed-13c9-46e2-ae34-a562748ed255'), ...]

Column의 이름과 함께 가져오려면 mappings()를 이용하자.

with engine.connect() as conn:
    query = conn.execute(text("select * from member")).mappings()
    print(query.fetchall())

# OutPut
[{'id': 13, 'random_id': '9b74f5ed-13c9-46e2-ae34-a562748ed255'}, ...]

mappings()를 통해 얻은 객체의 타입은 다음과 같다.

<class 'sqlalchemy.engine.result.MappingResult'>

 

4.  데이터 얻어오기

Raw SQL로 얻어온 결과는 PEP 248에 따라. fetch, fetchmany, fetchall 메서드를 사용할 수 있다.

 

4.1 fetch

# 단일값 가져오기 (첫번째 row)
with engine.connect() as conn:
    query = conn.execute(text("select * from member")).mappings()
    print(query.fetchone())

 

4.2 fetchmany

# 특정 개수만큼 가져오기
with engine.connect() as conn:
    query = conn.execute(text("select * from member")).mappings()
    print(query.fetchmany(size=5))

 

4.3 fetchall

# 전체 가져오기
with engine.connect() as conn:
    query = conn.execute(text("select * from member")).mappings()
    print(query.fetchall())

 

5.  Raw SQL에 value를 넘겨야 할 때

5.1 단일 값 매핑

Raws SQL에 WHER 절을 이용할 때 단일값을 넘기는 방법은 다음과 같다. 

with engine.connect() as conn:
    select_sql = "SELECT * FROM `member` WHERE random_id = :random_id;"
    params = {"random_id": "9b74f5ed-13c9-46e2-ae34-a562748ed255"}
    query = conn.execute(text(select_sql), params).mappings().fetchone()
    print(query)

dictionary로 key와 value를 만들고 execute 함수에 두 번째 인자로 넘긴다.

 

5.2 list를 넘겨야 할 때

list를 넘기는 방식은 다음과 같다.

with engine.connect() as conn:
    random_ids = [str(uuid.uuid4()) for _ in range(5)]

    select_sql = "SELECT * FROM `member` WHERE random_id IN :random_id;"
    params = {"random_id": random_ids}
    query = conn.execute(text(select_sql), params).mappings().fetchone()

dictionary 방식과는 별 차이는 없다. 

5.3 bindparams를 이용한 방식

dictionary를 이용한 방식 이외에도 bindparams 함수를 이용하면 다음과 같이도 이용할 수 있다.

with engine.connect() as conn:
    random_ids = [str(uuid.uuid4()) for _ in range(5)]

    select_sql = "SELECT * FROM `member` WHERE random_id IN :random_id;"
    query = conn.execute(text(select_sql).bindparams(random_id=random_ids)).mappings().fetchone()

5.4 Operator를 해야 할 때

and  나 or과 같이 특정 oeprator를 이용해야 하는 경우이다. 이 부분은 SQLALchemy가 지원해 주는 방식을 찾지 못했다. 그러나 코드를 이용해 구현하면 다음과 같다.

with engine.connect() as conn:

    select_sql = "SELECT * FROM `member` WHERE :conditions;"
    conditions = ' AND '.join(["random_id = '{}'".format([str(uuid.uuid4()) for _ in range(5)])])

    query = conn.execute(text(select_sql).bindparams(conditions=conditions)).mappings().fetchone()

 

5.5 INSERT VALUES를 만들어야 할 때

지금까지는 SELECT 절을 이용한 예제로 값을  넘기는 경우 이를 차용하면 된다. 그러나 Mutiple Values를 통해 insert 절을 실행시켜야 하는 경우는 용법이 조금 다르다.

with engine.connect() as conn:
    members = [Member(random_id=uuid.uuid4()).to_dict() for _ in range(10)]
    insert_sql = "INSERT INTO member (id, random_id) VALUES (:id, :random_id);"

    query = conn.execute(text(insert_sql), members)

생성된 쿼리를 보면 다음과 같다.

2023-10-19 11:30:37,236 INFO sqlalchemy.engine.Engine INSERT INTO member (id, random_id) VALUES (%(id)s, %(random_id)s);
2023-10-19 11:30:37,236 INFO sqlalchemy.engine.Engine [generated in 0.00017s] [{'id': None, 'random_id': UUID('6f37071f-3703-430e-bac9-37123aa81c81')}, ...]

 

6. 마치며

SQLALchemy에서 Raw SQL의 string mark type은 pep-0249에 따르면 named mark 방식이다. 그러나 Engine을 연결할 때 사용하는 DB API에 따라 권장되는 mark 방식이 다르다고 한다. 

 

즉 SQLAlchemy의 mark type과 사용하는 DB API의 mark type을 동일하게 맞춰주면 좋을 텐데 이 부분에 대한 자료를 찾지 못해 아쉽다.