[SQLAlchemy/Python] python바탕의 Query로 데이터 베이스 조작 (CRUD)

※ 일본의 한 블로그 글을 번역한 포스트입니다. 오역 및 직역, 의역이 있을 수 있으며 틀린 내용은 지적해주시면 감사하겠습니다.

 

 SQLAlchemy는 Python에서 자주 사용되는 OR Mapper이다. 이번 포스팅에서 SQLAlchemy를 이용하여 DB의 CRUD를 조작하는 방법에 대해서 설명하도록 하겠다. 

 

개발환경

---

 참고로 db_session이라는 명명은 딱히 별 다른 의미가 없으므로 포스트나 작성법에 따라 명명이 다르다.

• Windows10
• Python 3.8
• SQLite
• SQLAlchemy 2.0

 또한, Flask에서 SQLAlchemy을사용한 경우 Django와는 조금 사용법이 달라지지만 기본적인 작성법은 동일하므로, 프레임워크에 따라 적당히 변경하길 바란다.

# 이번 포스팅에서의 SQLAlchemy 작성법 예
employees = db_session.query(Employee).all()
employee = db_session.query(Employee).get(1)
employees = db_session.query(Employee).filter(Employee.id == 2)

# Flask SQLAlchemy의 작성법 예
employees = Employee.query.all()
employee = Employee.query.get(1)
employee = Employee.query.filter(Employee.id == 2)

 

 

get : 하나의 데이터 취득

---

 filter와 비슷하지만 데이터를 1개만 취득해보자.

• 개체의 오브젝트 취득
• 주키를 지정
• 데이터가 없을 경우 None을 반환

 복잡해 보이지만, 예를 들어 filter로 취득한 것은 리스트와 같이 for등으로 데이터를 뽑아 내지 않으면 각가의 데이터(레코드와 같은 이미지)는 사용할 수 없는 반면, get은 그러한 과정을 거치지 않더라도 각각의 데이터(레코드)를 사용할 수 있다고 생각하면 된다.

 일반적으로는 id를 주키로 하는 경우가 많으며, 데이터가 없어도 에러가 발생하지 않으므로, Django의 ORM과는 차이가 있다.

 아래의 코드는 주키를 이용하여 하나의 데이터를 취득하는 경우이다.

employee = db_session.query(Employee).get(1)
print(f'{employee.name}씨 근속연수{employee.year}년')

 참고로 버전 1.4이상을 사용하는 경우는 아래의 문법으로 사용하는 것을 추천한다. 달라진 점은 구문에 query를 사용하지 않는다는 점이다.

employee = db_session.get(Employee, 1)

 

 

all ; 모든 데이터를 획득하기

---

employees = db_session.query(Employee).all()
for employee in employees:
    print(f'{employee.name}씨 근속연수{employee.year}년')

홍길동씨 근속연수2년
김철수씨 근속연수3년
김영희씨 근속연수10년
강인한씨 근속연수5년

 

 

filter ; 조건에 따른 데이터를 획득하기

---

 

기본적인 구문은 다음과 같다.

세션.query(모델).filter(조건)

employees = db_session.query(Employee).filter(Employee.id == 1)

 

 

1. AND 조건

 여러개의 조건으로 검색하고 싶을 때 두 가지 방법이 있다.

1) 패턴1

 단순히 컴마를 붙여서 여러 가지 조건을 구분하여 작성하는 방법이다.

employees = db_session.query(Employee).filter(Employee.name == '홍길동', Employee.year == 10)

2) 패턴2

and_ 이라는 것을 import할 필요가 있다. 컴마로 여러 개의 조건을 지정한다.

from sqlalchemy import and_

employees = db_session.query(Employee).filter(and_(Employee.name == '홍길동', Employee.year == 10))

 

2. OR 조건

or_이라는 것을 import 할 필요가 있다. and_와 사용법은 동일하게 컴마로 여러 개의 조건을 지정하여 구분한다.

employees = db_session.query(Employee).filter(or_(Employee.name == '홍길동', Employee.year == 2))

 

3. IN 구문(Where in)

 filter로 조건을 이용한 검색 방법이 있지만, 조건을 리스트로 전달하고, 조건에 일치하는 데이터를 획득할 수 있는 스마트한 작성법이 존재한다. 그것은 in_([...])을 사용하는 것이다.

 기본 작성 구문은 다음과 같다.

 세션.query(모델).filter(모델.필드.in_([리스트 데이터]))

employees = db_session.query(Employee).filter(Employee.name.in_(['홍길동', '김영희', '강인한']))

 

4. not in 구문 (where not in)

 where in과 반대되는 구문. 리스트로 조건을 전달하며, 조건에 해당하지 않는 데이터만을 획득할 수 있다. 즉, 아래의 예문의 경우 이름이 홍길동, 김영희, 강인한이 아닌 데이터만 획득하게 된다.

employees = db_session.query(Employee).filter(~Employee.name.in_(['홍길동', '김영희', '강인한']))

 모델명 앞에 ~을 붙이므로 한눈에 알기 어려울 수 있다. 따라서 아래와 같은 문법으로 작성할 수 있다.

employees = db_session.query(Employee).filter(Employee.name.notin_(['홍길동', '김영희', '강인한']))

 

5. 부등호를 사용한 조건(「>」「<」「>=」「<=」)

 부등호는 알다시피 비교 조건시에 사용하므로 설명은 생략하도록 하겠다.

employees = db_session.query(Employee).filter(Employee.year > 2)
employees = db_session.query(Employee).filter(Employee.year >= 2)
employees = db_session.query(Employee).filter(Employee.year < 2)
employees = db_session.query(Employee).filter(Employee.year <= 2)
employees = db_session.query(Employee).filter(Employee.year > 2, Employee.year <= 5)

 

6. BETWEEN 구문

 값의 범위는 지정하는 방법은 위의 부등호를 사용한 방법도 가능하지만, BETWEEN 구문을 사용할 수 있다. 숫자의 범위나 일시시간을 start와 end로 지정한다.  from sqlalchemy import between 을 import할 필요가 있다.

 기본 구문은 다음과 같다.

 세션.query(모델).filter(betwenn(비교할 대상, start, end))

from sqlalchemy import between
employees = db_session.query(Employee).filter(between(Employee.year, 2, 5))

 

7. LIKE 구문

 부분 일치 조건을 이용하여 데이터를 획득할 때 사용한다.

1) 문자열에 일정한 문자가 포함되어 있는 경우

like나 contains를 사용하면 된다.

employees = db_session.query(Employee).filter(Employee.name.like('%ha%'))
employees = db_session.query(Employee).filter(Employee.name.contains('ha'))

2) 전방 일치

이것도 두 가지 작성법이 있으며, 결과는 동일하므로 원하는 것을 사용하면 된다.

employees = db_session.query(Employee).filter(Employee.name.startswith('T'))
employees = db_session.query(Employee).filter(Employee.name.ilike('T%'))

3) 후방 일치

 동일하게 두 가지 작성법이 존재한다.

employees = db_session.query(Employee).filter(Employee.name.endswith('a'))
employees = db_session.query(Employee).filter(Employee.name.like('a%'))

4) 대문자와 소문자

 사용한 like()는 대문자 소문자를 구문하지만 이번에 DB로 사용한 SQLite의 경우는 대문자와 소문자를 구분하지 않으므로 둘 다 획들 할 수 있다.

 ilike()를 사용하면, 지정한 문자열을 소문자로 변환한 후에 SQL로 획득하므로 주의할 필요가 있다. 아래의 구문을 실행하면 생성되는 SQL는 두 번째와 같다.

employees = db_session.query(Employee).filter(Employee.name.ilike('TA%'))

WHERE lower(employee.name) LIKE lower(?)

 따라서 like()를 사용한다면 뒤의 lower()가 없다.

 

8. NULL 검색(is null), 제외(is not null)

 단순히 == None으로 IS NULL이 된다. 그러나 is_(None)도 사용할 수 있다. 따라서 기본 구문은 아래의 두 가지가 존재할 수 있다.

세션.query(모델).filter(모델.필드 == None)

세션.query(모델).filter(모델.필드.is_(None))

employees = db_session.query(Employee).filter(Employee.name == None)
employees = db_session.query(Employee).filter(Employee.name.is_(None))

 Null 데이터를 제외하고 싶은 경우는 

세션.query(모델).filter(모델.필드 != None)

세션.query(모델).filter(모델.필드.is_not(None))

employees = db_session.query(Employee).filter(Employee.name != None)
employees = db_session.query(Employee).filter(Employee.name.is_not(None))

 

9. order_by  데이터 정렬

1) 오름차순

두 가지 방법이 있다. 두 번째의 경우는 asc를 import해야할 필요가 있다.

employees = db_session.query(Employee).filter(Employee.is_remote == False).order_by(Employee.year.asc())

from sqlalchemy import asc
employees = db_session.query(Employee).filter(Employee.is_remote == False).order_by(asc(Employee.year))

동시에 여러 열의 정렬하고 싶은 경우에는 단순히 컴마로 필드를 추가하면 된다.

employees = db_session.query(Employee).filter(Employee.is_remote == False).order_by(Employee.year, Employee.id)

2) 내림차순

내림차순의 경우 desc()를 사용한다.

employees = db_session.query(Employee).filter(Employee.is_remote == False).order_by(Employee.year.desc())

또는 asc()의 경우와 같이 desc를 import해서 작성하는 방법도 존재한다.

employees = db_session.query(Employee).filter(Employee.is_remote == False).order_by(desc(Employee.year))

 

10. count : 데이터 수를 획득

데이터의 개수를 획득하고 싶은 경우에는 count를 사용하면 된다.

ctn = db_session.query(Employee).count()

 

11. exists : 존재 여부 확인(True 혹은 False로 반환)

query = db_session.query(Employee).filter(Employee.is_remote == True)
if db_session.query(query.exists()).scalar():
    print('리모트 워크하는 사람이 있습니다.')
else:
    print('리모트 워크하는 사람이 없습니다.')

 

12. limit, offset : 획득하는 데이터 개수를 지정

 limit을 사용하면 최신 데이터 3개를 획득하는 경우 등에 사용되며 개수 지정이 필수이다. 그러나 offset은 지정이 필수가 아니다.. 

employees = db_session.query(Employee).order_by(Employee.created_at.desc()).limit(3)
print_all(employees)

employees = db_session.query(Employee).order_by(Employee.created_at.desc()).limit(3).offset(1)
print_all(employees)

 

 

데이터만 1개만 획득하고 싶은 경우

---

 맨 처음에 get을 사용한 방법을 소개했지만 filter한 후에 그 결과에서 데이터를 1개만 획득하고 싶은 경우가 있을 것이다. 그럴 경우를 위한 방법을 소개하고자 한다.

 

one ; 데이터를 1개만 획득하기

 앞서 설명했듯 filter한 결과에서 데이터를 사용하고 싶은 경우에 다음과 같이 쓸 수 있다.

employee = db_session.query(Employee).filter(Employee.id == 1).one()
print(f'{employee.name}씨 근속연수{employee.year}년')

 그러나 데이터가 존재하지 않는 경우나 데이터가 여러 개일 경우 에러가 발생한다. 그러한 에러가 발생할 경우를 대비해서 다음과 같이 작성할 수 있다.

from sqlalchemy.exc import NoResultFound
try:
    employee = db_session.query(Employee).filter(Employee.id == 100).one()
    print(f'{employee.name}씨 근속연수{employee.year}년')
except NoResultFound:
    print('데이터가 없습니다.')
except MultipleResultsFound:
    print('데이터가 2개이상 존재합니다.')

 

one_or_none()

 거의 one()과 비슷하지만 한 가지 다른 점이 있다.

• 데이터가 1개만 획득되지 않는다면 None

 데이터가  존재하지 않으면 에러대신에 None이 반환된다. 주의할 점은 여러 개가 존재하는 경우 one과 동일하게 에러가 발생한다는 점이다. 

employee = db_session.query(Employee).filter(Employee.id == 100).one_or_none()

 

firs()

 획득된 데이터 중 제일 처음의 데이터 1개만 반환한다.

employee = db_session.query(Employee).filter(Employee.year > 3).first()
print(f'{employee.name}씨 근속연수{employee.year}년')

---

참고자료

https://qiita.com/Bashi50/items/e3459ca2a4661ce5dac6

https://qiita.com/Bashi50/items/7f2214550e80b87fb922