[SQLAlchemy] (4) DML 예시 (by sqlalchemy 2.0)

sqlalchemy 2.0 특징

SQL Core의 select(), insert(), update(), delete()

- DML 함수들이 Core와 ORM 상호작용 공통의 방식으로 채택됨

 

Query 객체

- defrecated 되었음. (링크)

from sqlalchemy.orm import Query

# legacy 처리 됨 (2.0 에서)
q = session.query(SomeMappedClass)

 

execute

- core방식, orm 방식 둘다 사용 가능

- core 방식 : conn.execute(stmt)

- orm 방식 : session.execute(stmt)

- session.execute(stmt) => Result 객체를 반환

- Result 객체를 이용해 다양한 형태의 결과 처리 및 제어 가능 ex) fetchall(), scalars(), fetchone() 등

- 보다 유연하게 여러 형태의 데이터 처리 가능. (단일 열의 스칼라 값 반환하는 경우 사용 가능)

 

scalars

- session.execute() 의 축약형으로 => ScalarResult 객체를 반환한다.

- 덜 유연하지만 보다 짧은 코드 제공

- session.execute(stmt).scalars() = session.scalars(stmt)

 

SELECT 문 

방법 1 - scalars (쿼리문) 방식

• session.scalars(stmt)는 session.execute(stmt).scalars()의 축약형
• first() 메서드는 쿼리의 첫 번째 결과를 반환하며, 결과가 없는 경우 None을 반환
• one() 메서드는 쿼리의 결과가 정확히 하나일 때 사용
• all() 메서드는 쿼리의 결과를 모두 리스트 형태로 반환

from sqlalchemy import select
# 방법 1 - scalars(쿼리문) 방식
# 단일 객체 조회
print(session.scalars(select(User))) # ScalarResult
print(session.scalars(select(User)).first()) # User 객체
print(session.scalars(select(User).filter_by(age=30))) # ScalarResult

# filter(_by)와 where 는 같음
print(session.scalars(select(User).where(User.age == 30)).first()) # User 객체
print(session.scalars(select(User).where(User.age <= 10)).all()) # [User] - 1개여도 무조건 list

# filter, filter_by
print(session.scalars(select(User).filter(User.age <= 10))) # ScalarResult
print(session.scalars(select(User).filter_by(age=30)).all()) # [User, User..]

# one
print(session.scalars(select(User).filter_by(age=30)).one()) # Error -> MultipleResultsFoundError

# get 함수
print(session.get(entity=User, ident=2)) # User 객체 (ident 필수)

 

 

방법 2 - execute (쿼리문) 방식

from sqlalchemy import select

# 방법 2 - execute(쿼리문) 방식
print(session.execute(select(User))) # ChunkedIteratorResult
print(session.execute(select(User)).scalars()) # ResultScalars
print(session.execute(select(User)).scalars().first()) # User 객체
print(session.execute(select(User)).scalars().all()) # [User, User ...]

# filter(_by)와 where는 같음
print(session.execute(select(User).filter_by(age=10)).scalar_one()) # User 객체 - scalar_one()은 반드시 1개만 가능
print(session.execute(select(User).where(User.age==100)).scalar_one_or_none()) # None - filter와 where 는 같음

 

 

예시 - 특정 필드만 조회하기

- scalars는 단일 열의 스칼라 값 반환한다

# 방법 3 - 특정 필드만 조회 하는 방법
stmt = select(User.age, User.name)
print(session.execute(stmt).all()) # [(30, 'Alice'), (10, 'Gee1'), (30, 'Alice'), (14, 'Kee1')]
print(session.execute(stmt).scalars().all()) # [30, 10, 30, 14]

 

예시 - 연산자 비교

# 논리 연산자
stmt = select(User.age, User.name).filter(User.name=='Alice', User.age >=20)
print(session.execute(stmt).all()) # [(30, 'Alice'), (30, 'Alice')]

# Like 연산자
stmt = select(User.age, User.name).filter(User.name.like('%ee%'))
print(session.execute(stmt).all()) # [(10, 'Gee1'), (14, 'Kee1')]

# In 연산자, ~In(=Not In)연산자)
stmt = select(User.age, User.name).filter(~User.age.in_([10, 30]))
print(session.execute(stmt).all()) # [(14, 'Kee1')]

# is null or is not null
stmt = select(User.age, User.name).filter(User.name.is_(None))
print(session.execute(stmt).all()) # []
stmt = select(User.age, User.name).filter(User.name.isnot(None))
print(session.execute(stmt).all()) # [(30, 'Alice'), (10, 'Gee1'), (30, 'Alice'), (14, 'Kee1')]

 

INSERT 문 

1. orm 방식 - add

- add(), add_all() 메서드를 사용해 객체를 세션에 추가하고, commit(), flush()를 호출하여 db에 반영한다.

# ORM 방식으로 데이터 삽입
with Session(engine) as session:
    # 새로운 객체 생성
    new_user = User(name="Alice", age=30)
    
    # 세션에 객체 추가
    session.add(new_user)
    
    # 변경 사항을 커밋하여 데이터베이스에 반영
    session.commit()

 

2. core 방식 - insert 문

- core의 api를 사용하여 쿼리를 작성하는 방식

# Core 방식으로 데이터 삽입
with engine.connect() as conn:
    # insert 문을 정의하고 실행
    stmt = insert(users_table).values(name="Alice", age=30)
    conn.execute(stmt)
    conn.commit()

 

두 방법의 차이는

- 객체 지향적으로 사용하느냐 -> add()

- SQL 문 직접 작성하여 보다 복잡합 쿼리 작성에 이점을 가진다거나 -> insert()

- ORM 오버 헤드 대신 더 빠른 SQL 방식을 선택 -> insert()

등의 기준이 있다. 

 

UPDATE 문 

1. orm 방식

- 조회해온 user의 값 객체를 변경 후에 commit() 

# ORM 방식으로 데이터 업데이트
with Session(engine) as session:
    user = session.scalars(select(User)).first()
    user.age = 35
    session.commit()

 

2. core 방식

- update 쿼리문을 작성, engine.connect()가 아닌 session을 이용해서도 가능

# Core 방식으로 데이터 업데이트
with engine.connect() as conn:
    # update 문을 정의하고 실행
    stmt = (
        update(User)
        .where(User.c.name == "Alice")
        .values(age=35)
    )
    conn.execute(stmt)
    conn.commit()

 

 

UPSERT 문

- on_duplicate_key_update 구문을 이용해 unique_key를 기준으로 데이터가 없으면 추가, 데이터가 있으면 수정을 함

- Insert 객체의 inserted 로 기존 데이터에 접근 가능

from sqlalchemy.dialects.mysql import insert

insert_stmt = insert(my_table).values(id=3, column1='더미더미값')

# 유형 1 
on_duplicate_key_stmt = insert_stmt.on_duplicate_key_update(
        column1=insert_stmt.inserted.data,
        status='U'
	)
# INSERT INTO my_table (id, column1) VALUES (%s, %s) ON DUPLICATE KEY UPDATE column1 = VALUES(column1), status = %s

print(on_duplicate_key_stmt)

# 유형 2
on_duplicate_key_stmt2 = insert_stmt.on_duplicate_key_update(
	{"column1": "some data", "updated_at": func.current_timestamp()},
)
print(on_duplicate_key_stmt2)
# INSERT INTO my_table (id,column1data) VALUES (%s, %s) ON DUPLICATE KEY UPDATE column1 = %s, updated_at = CURRENT_TIMESTAMP

 

 

 

DELETE 문 

1. orm 방식 - delete()

- 조회해온 user의 값 객체를 삭제 후 commit() 

# ORM 방식으로 데이터 업데이트
with Session(engine) as session:
    user = session.scalars(select(User)).first()
    session.delete(user)
    session.commit()

 

2. core 방식

- delete 쿼리문을 작성, engine.connect()가 아닌 session을 이용해서도 가능

# Core 방식으로 데이터 업데이트
with engine.connect() as conn:
    # delete 문을 정의하고 실행
    stmt = (
        delete(User)
        .where(User.c.name == "Alice")
    )
    conn.execute(stmt)
    conn.commit()