[Flask] SQLAlchemy ORM 활용법

geunyeongii·2022년 1월 18일

ORM(Object Relational Mapper)?

객체와 관계형 데이터베이스의 데이터를 자동으로 매핑(연결)해주는 것을 말한다.
- 객체 지향 프로그래밍은 클래스를 사용하고, 관계형 데이터베이스는 테이블을 사용한다.
- 객체 모델과 관계형 모델 간에 불일치가 존재한다.
- ORM을 통해 객체 간의 관계를 바탕으로 SQL을 자동으로 생성하여 불일치를 해결한다.
데이터베이스 데이터 <—매핑—> Object 필드
- 객체를 통해 간접적으로 데이터베이스 데이터를 다룬다.
Persistant API라고도 할 수 있다.
- Ex) JPA, Hibernate 등

ORM 활용법

User 객체에 db.Model을 상속했을 경우 User객체에서 바로 query 함수를 불러올 수 있다.
ex ) WHERE part에서 -> user = User.query.filter(User.name != name)

SELECT

session.query(Model).all() # SELECT * FROM model
session.query(Model.id, Model.name, Model.age).all() # SELECT id, name, age FROM model
session.query(Model).first() # SELECT * FROM model LIMIT 1

from sqlalchemy import func # count 함수
session.query(func.count(Model.id))
# SELECT COUNT(id) FROM model

# AS 
session.query(Model.id.label('model_id')).all()
# SELECT id AS model_id FROM model

WHERE

filter를 사용!!

session.query(Model).filter(Model.name == 'lowell').all() 
# SELECT * FROM model WHERE name = 'lowell'

session.query(Model).filter(Model.name == 'lowell', Model.age == 20).all() 
# SELECT * FROM model WHERE name = 'lowell' AND age = 20

from sqlalchemy import or_ # OR 연산자
session.query(Model).filter(or_(Model.mame == 'lowell', Model.age == 20)).all()
# SELECT * FORM model WHERE name = 'lowell' OR age = 20

filter_by를 사용!!

session.query.filter(User.name != name)
# user = User.query.filter(User.name != name)

session.query.filter((User.email == email) | (User.name == name)).first()
# user = User.query.filter((User.email == email) | (User.name == name)).first()

session.query.filter(User.name.like('%minsoo%'))
# user = User.query.filter(User.name.like('%minsoo%'))

INSERT

user = Model(name='lowell', age=20)
session.add(user)
session.commit() # session.rollback()

# INSERT INTO model(name, age) VALUES ('lowell', 20)
# COMMIT

UPDATE

1) 첫 번째 방법

user = session.query(Model).filter(Model.name == 'lowell').first()
user.age += 1
session.commit()

# select 후 update 함

2) 두 번째 방법

user = session.query(Model).filter(Model.name == 'lowell').update({'age': User.age + 1});
session.commit()

# select를 하지 않는 update 방법
# UPDATE model SET age = age + 1 WHERE name = 'lowell'

DELETE

user = session.query(Model).filter(Model.name == 'lowell').first()
session.delete(user)
session.commit()

ORDER BY

session.query(Model).filter(Model.name == 'lowell').order_by(Model.created_at)
# SELECT * FROM model WHERE name = 'lowell' ORDER BY created_at

session.query(Model).filter(Model.name == 'lowell').order_by(Model.created_at.desc(), Model.status) 
# SELECT * FROM model WHERE name = 'lowell' ORDER BY created_at DESC, status

JOIN

1) INNER JOIN

session.query(Model1, Model2).filter(Model1.id == Model2.id).all() 
# SELECT * FROM model1 JOIN model2 ON model1.id = model2.id

session.query(Model1).join(Model2, Model1.id == Model2.id).all()
# SELECT * FROM model1 JOIN model2 ON model1.id = model2.id

2) OUTER JOIN

session.query(Model1). \
    outerjoin(Model2, Model1.id == Model2.id).\
    all()
    
# SELECT * FROM model1 LEFT JOIN model2 ON model1.id = model2.id

여러개 조인을 하기 위해 그냥 이어 붙이기!

session.query(Model1.name, Model2.student_id, Model3.account).\
	outerjoin(Model2, Model1.id == Model2.id).\
	outerjoin(Model3, Model1.id == Model3.id).\
	all()
	
# SELECT model1.name, model2.student_id , model3.account 
# FROM model1
# LEFT JOIN model2 ON model1.id = model2.id
# LEFT JOIN model3 ON model1.id = model3.id

3) SELF JOIN

하나의 테이블을 조인하는 것

aliased를 사용

model2 = aliased(Model)

self.session.query(Model).\
    join(model2, model2.id == Model.id).\
    all()
    
# SELECT model.* FROM model JOIN model model2 ON model2.id = model.id;

GROUP BY

session.query(Model).group_by(Model.id).all()

# SELECT * FROM model GROUP BY id

SUBQUERY

from sqlalchemy import subquery

stmt = query.session(Model2).filter(Model2.grade == 'A').subquery() 
# SELECT id, grade FROM model2 WHERE grade = 'A'

session.query(Model1, stmt.c.id, stmt.c.grade).\
	outerjoin(stmt, stmt.c.id = Model1.id)
	
# SELECT model1.*, model2.id, model2.grade
# FROM mode1l LEFT JOIN (SELECT id, grade FROM model2 WHERE grade = 'A') model2 
# ON model1.id = model2.id

Tip!!!

1) CASE문

from sqlalchemy import case
	
session.query(
	case(
        	[
            	(Model.age >= 20, 'adult'),
                (Model.age >= 10, 'teenager')
        	], 
        	else_='not adult, not teenager'
    	)
    ).\
    filter(Model.sex='female').\
    all()

# SELECT CASE WHEN age >= 20 THEN 'adult' WHEN age >= 10 THEN 'teenager' ELSE 'not adult, not teenager' FROM model WHERE sex = 'female';

2) last_row_id 얻기

user = Model(name='lowell', age=20)
session.add(user)
session.flush() # DB connection 일어남

id = user.id # auto_encrement로 생성된 id

session.commit()

3) 검색(LIKE)

results = session.query(Model).\
	filter(Model.name.like('김%')).all() # 성이 김씨인 사람 찾음
    
# 응용해보긔 

keyword = kwargs.get('keyword', '')
search = True 

if keyword:
    search = Model.name.like(f'{keyword}%')
    
results = session.query(Model).\
	filter(search).all() # 검색어가 있으면 검색, 없으면 모두 가져옴

4) IN

session.query(Model).filter(Model.name.in_(('lowell', 'yejin'))).all() 
# SELECT * FROM model WHERE name IN ('lowell', 'yejin');

5) NOT IN

session.query(Model).filter(~Model.name.in_(('lowell', 'yejin'))).all() 
# SELECT * FROM model WHERE name NOT IN ('lowell', 'yejin');

6) COMMIT, ROLLBACK

session.commit() # commit
session.rollback() # rollback

7 DATE 계산하기 (DATE_ADD)

today = datetime.datetime.now().strftime('%Y-%m-%d')
session.query(Model).filter(today >= func.ADDDATE(Model.created_at, 30)).all()
# SELECT * FROM model WHERE NOW() >= DATE_ADD(created_at, INTERVAL 30 DAY);