Sqlalchemy Hybrid Attributes


어떤 모델 테이블의 hybrid attribute 를 만들 때 sqlalchemy 에서는  hybrid_property 데코레이터를 이용해 코드를 줄여서 간단하게 만들 수 있다.


간단하게 시작 점과 끝 점을 저장하는 테이블이 있다고 가정했을 때, 두 점의 사이를 매번 계산하는 것은 반복적인 작업이다. 

그리고 table의 속성으로 지정해놓기에는 start 와 end 가 변경될 때 마다 변경해줘야 하므로 귀찮은 작업이다.

이럴 때 sqlalchemy의 hybrid attributes 를 사용하면 유용하다.


밑의 예시는 모두 Sqlalchemy 공식 documentation 의 문서

(출처 : http://docs.sqlalchemy.org/en/latest/orm/extensions/hybrid.html )에서 가져온 코드입니다.


밑의 코드를 보면,

from sqlalchemy import Column, Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session, aliased
from sqlalchemy.ext.hybrid import hybrid_property, hybrid_method
Base = declarative_base()


class Interval(Base):
    __tablename__ = 'interval'

    id = Column(Integer, primary_key=True)
    start = Column(Integer, nullable=False)
    end = Column(Integer, nullable=False)

    def __init__(self, start, end):
        self.start = start
        self.end = end

    @property
    def length(self):
        return self.end - self.start


위와 같이 만들 수 있겠지만, 위와 같은 경우는 query 를 이용할 때 해당 속성을 사용할 수 없다.

그렇기 때문에 아래와 같이 hybrid property 를 이용하여 attribute를 만들면 sqlalchemy orm 을 이용할 때도 해당 속성을 이용하여 

Query 를 작성할 수 있다.


Base = declarative_base()

class Interval(Base):
    __tablename__ = 'interval'

    id = Column(Integer, primary_key=True)
    start = Column(Integer, nullable=False)
    end = Column(Integer, nullable=False)

    def __init__(self, start, end):
        self.start = start
        self.end = end

    @hybrid_property
    def length(self):
        return self.end - self.start

    @hybrid_method
    def contains(self, point):
        return (self.start <= point) & (point <= self.end)

    @hybrid_method
    def intersects(self, other):
        return self.contains(other.start) | self.contains(other.end)

- 예시)
Print 함수를 통해 ORM 을 raw query로 보면

>>> print Session().query(Interval).filter(Interval.length > 10)
SELECT interval.id AS interval_id, interval.start AS interval_start,
interval."end" AS interval_end
FROM interval
WHERE interval."end" - interval.start > :param_1

>>> print Session().query(Interval).filter_by(length=5)
SELECT interval.id AS interval_id, interval.start AS interval_start,
interval."end" AS interval_end
FROM interval
WHERE interval."end" - interval.start = :param_1

>>> print Session().query(Interval).filter(Interval.contains(15))
SELECT interval.id AS interval_id, interval.start AS interval_start,
interval."end" AS interval_end
FROM interval
WHERE interval.start <= :start_1 AND interval."end" > :end_1



Defining Expression Behavior Distinct from Attribute Behavior

기본적으로 & 그리고 | 같은 비트 연산자는 ORM 에서 다른 함수의 인터페이스 없이 바로 적용할 수 있지만 abs같은 함수를 db 에서 정의하기 위해서는 sqlalchemy의 func 함수를 이용해야 한다.

from sqlalchemy import func

class Interval(object):
    # ...

    @hybrid_property
    def radius(self):
        return abs(self.length) / 2

    @radius.expression
    def radius(cls):
        return func.abs(cls.length) / 2

EX)
위와 마찬가지로 Print 함수를 통해 ORM 을 raw query로 보면

>>> print Session().query(Interval).filter(Interval.radius > 5)
SELECT interval.id AS interval_id, interval.start AS interval_start,
    interval."end" AS interval_end
FROM interval
WHERE abs(interval."end" - interval.start) / :abs_1 > :param_1


Defining Setters

하이브리드 property들 또한 setter method 를 사용할 수 있다.

class Interval(object):
    # ...

    @hybrid_property
    def length(self):
        return self.end - self.start

    @length.setter
    def length(self, value):
        self.end = self.start + value

EX)  코드 예시
>>> i1 = Interval(5, 10)
>>> i1.length
5
>>> i1.length = 12
>>> i1.end
17


Allowing Bulk ORM Update

sqlalchmey version 1.2  부터 hybrid properties 의 bulk update가 지원됩니다.

Update_expression 이라는 장식자를 이용해서 사용할 수 있습니다.

class Interval(object):
    # ...

    @hybrid_property
    def length(self):
        return self.end - self.start

    @length.setter
    def length(self, value):
        self.end = self.start + value

    @length.update_expression
    def length(cls, value):
        return [
            (cls.end, cls.start + value)
        ]


EX) 코드 예시

session.query(Interval).update(
    {Interval.length: 25}, synchronize_session='fetch)

UPDATE interval SET end=start + :value



Working with Relationships 

한 테이블의 column data로 만들었을 때와 relationship table에 column으로 만들 때와 큰 차이는 없다.

Relationship table에서 hybrid attributes는 “join-dependent”와 “correlated subquery” 두 가지 형태가 있다.

join-dependent 방식은 조인해서 어떤 1:N 관계에서 N 개 중 한 개의 인스턴스에 대해 다룰 때 쓰이고,

Correlated subquery 같은 경우는 subquery를 이용해서 aggregate 같은 연산을 하는데 사용할 수 있다.


join-dependent Relationship Hybrid

아래와 같이 두 relationship table을 선언했을 때,

from sqlalchemy import Column, Integer, ForeignKey, Numeric, String
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.hybrid import hybrid_property

Base = declarative_base()

class SavingsAccount(Base):
    __tablename__ = 'account'
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('user.id'), nullable=False)
    balance = Column(Numeric(15, 5))

class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)

    accounts = relationship("SavingsAccount", backref="owner")

    @hybrid_property
    def balance(self):
        if self.accounts:
            return self.accounts[0].balance
        else:
            return None

    @balance.setter
    def balance(self, value):
        if not self.accounts:
            account = Account(owner=self)
        else:
            account = self.accounts[0]
        account.balance = value

    @balance.expression
    def balance(cls):
        return SavingsAccount.balance

Ex) 

>>> from sqlalchemy import or_
>>> print (Session().query(User, User.balance).outerjoin(User.accounts).
...         filter(or_(User.balance < 5000, User.balance == None)))
SELECT "user".id AS user_id, "user".name AS user_name,
account.balance AS account_balance
FROM "user" LEFT OUTER JOIN account ON "user".id = account.user_id
WHERE account.balance <  :balance_1 OR account.balance IS NULL


Correlated Subquery Relationship Hybrid

from sqlalchemy import Column, Integer, ForeignKey, Numeric, String
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy import select, func

Base = declarative_base()

class SavingsAccount(Base):
    __tablename__ = 'account'
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('user.id'), nullable=False)
    balance = Column(Numeric(15, 5))

class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)

    accounts = relationship("SavingsAccount", backref="owner")

    @hybrid_property
    def balance(self):
        return sum(acc.balance for acc in self.accounts)

    @balance.expression
    def balance(cls):
        return select([func.sum(SavingsAccount.balance)]).\
                where(SavingsAccount.user_id==cls.id).\
                label('total_balance')

Ex)
>>> print s.query(User).filter(User.balance > 400)
SELECT "user".id AS user_id, "user".name AS user_name
FROM "user"
WHERE (SELECT sum(account.balance) AS sum_1
FROM account
WHERE account.user_id = "user".id) > :param_1


documentation 을 보고 혼자 간단하게 내용을 정리해보면서 적어본 내용입니다.

감사합니다.









안녕하세요.


몇 일 전 통계를 내는 페이지를 만들다보니 여러 테이블의 row를 mysql 에서 가져올 일이 있었습니다.


뭐 이 밖에도 여러 테이블의 row를 카운트 하는 경우가 종종 있습니다.



저는 flask 프레임워크를 사용하고 sqlalchemy를 사용하고 있기 때문에 


sqlalchemy에서 개선한 경험을 이야기 할 예정입니다.




1) ORM 이용 -> 조금 느림.


맨 처음에  sqlalchemy 의 ORM 을 사용했습니다.


예를 들면, 유저 수(user) , 주문1(order1), 주문2(order2) 라고 했을 때


len(User.query.all())

len(Order1.query.all())

len(Order2.query.all())



이렇게 했을 때 모든 필드를 다 가져 오기 때문에 row 가 많다면 속도가 느려질 수 있습니다.


+ 자신이 쓴 ORM이 어떻게 sql 로 바뀌는지 알고 싶다면


print(orm) 을 하거나, sqlalchemy 에서는 SQLALCHEMY_ECHO = True 값을 설정해주면 확인할 수 있습니다.


이를 개선하기 위해 raw query를 이용했습니다.




2) Raw query 이용


 sqlalchemy 에서는


db.engine.execute('sql') 를 이용하였습니다.


위에 orm을 


db.engine.execute('select count(id) from user;')

db.engine.execute('select count(id) from order1;')

db.engine.execute('select count(id) from order2;')


로 바꿔서 성능을 orm보다 개선했습니다.


여기서 다시 한 번 subquery를 이용했을 때 와 성능을 비교해보면




3) subquery


subquery를 이용하면 하나의 sql 에 여러 테이블의 원하는 row의 data를 가져올 수 있습니다.


db.engine.execute('select (select count(id) from user), '

                  '(select count(id) from order1), '

                  '(select count(id) from order2), '

                  )


Raw sql을 이용하면 하나의 단점은 객체 타입이 ResultProxy 이기 때문에 데이터를 파싱해야 되는 단점이 있지만, 전체적으로 성능은 좋아지기 때문에 


이 점은 무시해도 될 것 같습니다.

 

결과적으로 subquery를 이용해서 보다 성능면에서 개선할 수 있었습니다.


2번보다 3번이 조금 빠른 이유는 mysql과의 통신 비용이 줄어들었기 때문일 거라고 생각합니다. 



이삿짐을 여러번 나눠서 옮기는 것보다 한 번에 많이 들어서 옮기는 것이 조금 더 빠르겠죠? (왔다갔다하는 시간을 줄일 수 있다는 관점에서)




제가 잘 못 알고 있는 부분이 있거나 충고해주실 부분이 있다면 


말씀해주시면 정말 감사하겠습니다.





  1. 꼴뚜기 2017.12.28 18:21 신고

    열심히하시는 모습이 멋지십니다

  2. 2018.01.06 17:42

    비밀댓글입니다

  3. 2018.01.06 17:42

    비밀댓글입니다

  4. 박대성 2018.04.17 17:17 신고

    len(User.query.all()) 을 하게 되면 쿼리에서 가져온 리스트에 대해 사이즈를 세게 되서 느릴거 같은데요..

    내부에 이미 User.query.count() 메소드가 있지 않나요?
    이렇게 해도 속도가 느릴까요???

    • jordan_bae jordan17 2018.04.20 19:37 신고

      그러게요ㅠㅠㅠ 제가 당연히 있을 count() 메써드를 안쌋네요! 하지만 저 부분과 관계없이 raw query가 더 빠르기는 합니다만 상황에 따라서 편의성과 성능 중에 선택하셔서 쓰면 될 것 같습니다!

## 문제

Q. 다음 요구사항을 만족하는 RESTful API 를 파이썬과 Flask를 이용해서 만드세요.

정수를 포함하는 배열을 입력으로 받아서, 최대값을 리턴합니다.
요청 메시지와 응답 메시지는 모두 JSON 포맷입니다.

[요청 메시지 샘플]
{
“numbers”: [3,8,2,3,9,5,4]
}

[응답 메시지 샘플]
{
“result”: 9
}



해당 문제에 대해서 Flask를 이용해서 해결하도록 하겠습니다.


먼저 필요한 라이브러리로는 


Flask==0.12.2

flask-json

requests

nose


requests와 nose 라이브러리는 테스트를 위하여 설치합니다.


간단한 프로젝트이기 때문에 hello.py 하나의 파이썬 파일로 구성하고, 알고리즘 부분은 core.py , 


테스트 부분은 tests/test.py로 구성하였습니다.


import time
from flask import Flask
from core import maxheap_element, check_value_type, merge_sort, maxheap_element2
from flask import request
from flask_json import FlaskJSON, JsonError, json_response
app= Flask(__name__)
json = FlaskJSON(app)

@app.route('/heapsort',methods=['GET','POST'])
def index():
start_time = time.time()

# json으로 요청이 오기 때문에 request.get_json메서드를 이용하여 data를 받은 후

# data는 json type이기 때문에 원하는 input인 data['numbers']를 얻습니다.

# try - except 구문으로 input이 json이 아닌경우 json의 key가 numbers가 아닌 경우 JSONERROR를 발생시켜 예외 처리를

# 해줍니다.


data = request.get_json(force=True)

try:
value = data['numbers']

except (KeyError, TypeError, ValueError):
raise JsonError(description='Invalid value.')

check_value_type(value) #value가 int element를 가진 list인지 체크

result = maxheap_element(value) # heap_sort를 이용해서 처리

execution_time=time.time() - start_time # 실행시간 체크

sort = 'heap sort'

return json_response(result=result,time=execution_time,sort=sort)

if __name__ == '__main__':
app.run(debug=True)

저는 여러가지 알고리즘을 적용해서 길이가 1만인 리스트를 요청으로 보내서


걸리는 시간을 체크해봤습니다.


테스트는 JUPYTER NOTEBOOK에서 Requests 라이브러리를 이용해서 테스트 하였습니다.


속도는 heapsort > quick sort > merge sort 순이었습니다.









+ Recent posts