[Python] SQLAlchemy: ORM을 이용한 테이블 조회 활용 (5) - 응용 예시

[Python] SQLAlchemy: ORM을 이용한 테이블 조회 활용 (5) - 응용 예시

---

[SQLAlchemy  활용 목차]
[Python] SQLAlchemy: ORM(Object Relational Mapping)
1. 객체 기반 데이터 활용 SQLAlchemy ORM 개요
2. ORM Filter로 조회 조건 활용
3. ORM filter와 filter_by 사용법 비교
4. ORM 쿼리 결과 부분 조회 처리
​ 5. ORM을 이용한 테이블 조회 활용 예시​​

---

 

sqlalchemy를 이용하여 Data 입력과 데이터 수정 삭제에 대한 예제를 만들어 보았습니다.
각 영역별로 데이터 DDL, DML을 이용하는 예시로 참조 바랍니다.
 

# 파이썬 참조 모듈 선언

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session

from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm.exc import FlushError
from sqlalchemy import desc

 
 

# 데이터베이스 연결 설정

# 데이터베이스 연결 설정
engine = create_engine('sqlite:///db/_lec04_bulk.db')

# SQLite 데이터베이스 사용
db_session = scoped_session(sessionmaker(autocommit=False, autoflush=False,bind=engine))

Base = declarative_base()

 
 

# 데이터 베이스 초기화

#데이터베이스 초기화 함수
def init_db():
    Base.metadata.create_all(bind=engine)

 

# 테이블 객체 선언

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String(10), unique=True)
    email= Column(String(120), unique=True)

 

# 테이블이 생성됨

## 테이블이 생성됨
init_db() 

 
 

# 데이터 중복 입력시 업데이트와 삽입을 처리 하는 방법

- User 테이블은 name, email에 unique  설정이 되어 프로그램을 여러번 실행하면 데이터 중복이 발생 됩니다. 기존 값이 있는지 확인하고 있다면 업데이트(update)를 하고 없다면 삽입(insert)를 하는 방식으로 처리하는 방법입니다.
물론 이러한 방식은 대량의 데이터 처리를 하는 경우 불리합니다. 이유는 데이터를 건별로 검색을 하고 업데이트와 삽입을 선별해야하는 Loop가 발생되기 때문입니다.

user_list= [ {'name': 'test', 'email': 'btest00@email.com'},
                  {'name': 'test11', 'email': 'btest11@email.com'},
                  {'name': 'test22', 'email': 'btest22@amail.com'},
                  {'name': 'test33', 'email': 'btest33@dmail.com'},
                  {'name': 'test44', 'email': 'btest44@dmail.com'},
]

# bulk_update_mappings 대신 for 루프를 사용하여 업데이트
for data in user_list:
    try:
          db_session.query(User).filter_by(name=data['name']).update(data)
     except (IntegrityError, FlushError):
          db_session.rollback() continue # 새로운 데이터인 경우
          if  db_session.query(User).filter_by(name=data['name']).count() == 0:
              db_session.add(User(**data))

# 데이터베이스 커밋
db_session.commit()

 

# Data 조회

쿼리 결과를 조회 하기 위한 filter와 .first() .all()   like('%string%') . limit(N) 등을 이용한 조회는 다음과 같이 사용하면 됩니다.

# Data 조회 
#------------------------
testUsers = db_session.query(User).filter(User.name == 'test').first()
print( testUsers )testUsers = db_session.query(User).filter(User.name == 'test1').all()
print( testUsers )

# 이름에 'test' 문자열이 포함된 사용자 중에서 10명까지 조회
testUsers = db_session.query(User).filter(User.name.like('%test%')).limit(10).all()
print(testUsers)

 
 

# Data 삭제 : like를 이용한 데이터 삭제

# Data 삭제 : like를 이용한 데이터 삭제
#------------------------
# like와 limit를 이용한 데이터 삭제
query = db_session.query(User).filter(User.email.like('%@email.com')).limit(2)users = query.all()for user in users: db_session.delete(user)db_session.commit()

 

# Data 조회 : like를 이용한 데이터 조회

 

# 이름에 'test' 문자열이 포함된 사용자 중에서 20명만 조회
testUsers = db_session.query(User).filter(User.name.like('%test%')).limit(20).all()
print(testUsers)

# id 열을 기준으로 내림차순으로 정렬한 결과 반환
testUsers = db_session.query(User).filter(User.name.like('test%')).order_by(desc(User.id)).all()
print('> like :', testUsers)

 

# 조건문 : IS NULL, IS NOT NULL : None,  and_ 이용

일반적으로 filter(조건1, 조건2, ...) 와 ( and_(조건1, 조건2) ) 의 조건문과 Null 데이터 조회방법을 참고 하면 좋겠습니다.

# IS NULL, IS NOT NULL : None
testUsers = db_session.query(User).filter(User.email.is_(None)).all()
print('> IS NULL :', testUsers)

testUsers = db_session.query(User).filter(User.email.isnot(None)).all()
print('> IS NOT NULL :', testUsers)

# and_ 를 이용한 조건문 
from sqlalchemy import and_
for user in db_session.query(User).filter(and_(User.name == 'test2',User.email == 'btest20@amail.com')):
    print('> _and :', testUsers)

 

# Sqlalchemy를 이용한 ORM 예제

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session

from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm.exc import FlushError
from sqlalchemy import desc



# 데이터베이스 연결 설정
engine = create_engine('sqlite:///db/_lec04_bulk.db')  # SQLite 데이터베이스 사용
db_session = scoped_session(sessionmaker(autocommit=False, autoflush=False,bind=engine))

Base = declarative_base()

#데이터베이스 초기화 함수

def init_db():
    Base.metadata.create_all(bind=engine)

class User(Base):
    __tablename__ = "users"

    id   = Column(Integer, primary_key=True)
    name = Column(String(10), unique=True)
    email= Column(String(120), unique=True)

    def __init__(self, name=None, email=None):
        self.name = name
        self.email= email

    def __repr__(self):
        return '<User (%r, %r, %r)>' %(self.id, self.name, self.email)

init_db()   ## 테이블이 생성됨

user_list= [
    {'name': 'test',  'email': 'btest00@email.com'},
    {'name': 'test11', 'email': 'btest11@email.com'},
    {'name': 'test22', 'email': 'btest22@amail.com'},
    {'name': 'test33', 'email': 'btest33@dmail.com'},
    {'name': 'test44', 'email': 'btest44@dmail.com'},
]

# bulk_update_mappings 대신 for 루프를 사용하여 업데이트
for data in user_list:
    try:
        db_session.query(User).filter_by(name=data['name']).update(data)
    except (IntegrityError, FlushError):
        db_session.rollback()
        continue

    # 새로운 데이터인 경우
    if db_session.query(User).filter_by(name=data['name']).count() == 0:
        db_session.add(User(**data))

# 데이터베이스 커밋
db_session.commit()

# Data 조회
#------------------------
testUsers = db_session.query(User).filter(User.name == 'test').first()
print( testUsers )

testUsers = db_session.query(User).filter(User.name == 'test1').all()
print( testUsers )

# 이름에 'test' 문자열이 포함된 사용자 중에서 10명까지 조회
testUsers = db_session.query(User).filter(User.name.like('%test%')).limit(10).all()
print(testUsers)


# Data 삭제 : like를 이용한 데이터 삭제
#------------------------
# like와 limit를 이용한 데이터 삭제
query = db_session.query(User).filter(User.email.like('%@email.com')).limit(2)
users = query.all()
for user in users:
    db_session.delete(user)
db_session.commit()


# 이름에 'test' 문자열이 포함된 사용자 중에서 첫 번째 사용자를 조회
testUser = db_session.query(User).filter(User.name.like('%test%')).first()
print(testUser)

# 이름에 'test' 문자열이 포함된 사용자 중에서 20명만 조회
testUsers = db_session.query(User).filter(User.name.like('%test%')).limit(20).all()
print(testUsers)

# id 열을 기준으로 내림차순으로 정렬한 결과 반환
testUsers = db_session.query(User).filter(User.name.like('test%')).order_by(desc(User.id)).all()
print('> like :', testUsers)



# id 열을 기준으로 내림차순으로 정렬한 결과 반환
testUsers = db_session.query(User).filter(User.name.in_(['test1','test2','test3'])).order_by(desc(User.id)).all()
print('> IN :', testUsers)



# IS NULL, IS NOT NULL : None

testUsers = db_session.query(User).filter(User.email.is_(None)).all()
print('> IS NULL :', testUsers)

testUsers = db_session.query(User).filter(User.email.isnot(None)).all()
print('> IS NOT NULL :', testUsers)


# and
from sqlalchemy import and_
for user in db_session.query(User).filter(and_(User.name == 'test2',User.email == 'btest20@amail.com')):
    print('> _and :', testUsers)

 

---

 

---

[SQLAlchemy  활용 목차]
[Python] SQLAlchemy: ORM(Object Relational Mapping)
1. 객체 기반 데이터 활용 SQLAlchemy ORM 개요
2. ORM Filter로 조회 조건 활용
3. ORM filter와 filter_by 사용법 비교
4. ORM 쿼리 결과 부분 조회 처리
​ 5. ORM을 이용한 테이블 조회 활용 예시​​

---