[sqlalchemy] Postgresql Query ORM 변환

SQL Alchemy

어제 회사에서 코드를 보다가 뭔가 개선점을 발견하고 한참을 작업해서 코드를 수정했는데 기록을 남겨두고자 포스팅한다.

AS-IS

해당 코드는 이런 역할을 한다. 검색에서 광고로 사용할 키워드를 등록해두면 사용자가 검색 질의를 하면 해당 키워드가 현재 사용중인 광고에 포함되는 키워드인지 확인해보는 것이다.

각 광고는 여러개의 키워드를 가질 수 있으면 광고와 키워드 간의 관계는 m:n의 관계이며 각 광고는 서비스의 시작일과 종료일을 가진다.

이렇게 키워드를 추출하는 이유는 DB의 부하를 줄이기 위해서 키워드 리스트만 따로 추출해서 레디스에 들고 있다가 질의가 들어오면 해당 키워드로 등록된 광고가 있는지 확인하고 있으면 정보를 가져오고 없으면 빈 값을 보내기 위해서이다. 즉, 디비의 부하를 줄이기 위한 조치이다.

기존에는 사용중이 키워드의 전체 리스트를 가져와서 결과에서 루프를 돌리며 키워드를 추출하고 시작일 또는 만료일이 가장 먼저 도래하는 날짜를 찾았다. 날짜를 찾는 이유는 레디스의 TTL을 구워서 해당 시간이 되면 레디스를 날리고 다시 레디스를 생성하기 위해서이다.

여기서는 다음과 같이 SQL문을 작성한다.

select
    keyword,
    show_start_timestamp,
    show_end_timestamp
from
    tb_ads
        join tb_ads_keyword on tb_ads.ad_no = tb_ads_keyword.ad_no
where
    tb_ads.is_deleted = 'F'
and
    tb_ads_keyword.is_deleted = 'F'
and
    tb_ads.is_published = 'T'
and
    (tb_ads.show_start_timestamp >= now() or tb_ads.show_end_timestamp >= now())

이렇게 하면 사용중인 키워드 리스트와 각 키워드의 사용 시작일과 종료일을 알 수 있다. 해당 쿼리를 sqlalchemy를 통해서 statement로 작성하면 아래 코드와 같이 된다.

stmt = select(models.AdKeyword.keyword, models.Ad.show_start_timestamp, models.Ad.show_end_timestamp).join(
    models.Ad,
    models.Ad.ad_no == models.AdKeyword.ad_no
).filter(
    models.AdKeyword.is_deleted == 'F',
    models.Ad.is_deleted == 'F',
    models.Ad.is_published == 'T',
    ((models.Ad.show_start_timestamp >= func.now()) | (models.Ad.show_end_timestamp >= func.now()))
)

이렇게 코드를 작성해서 쿼리를 돌리고 나면 결과가 리스트로 나올테니 for문을 통해서 결과를 만들면 된다. 다음 코드가 해당 역할을 한다.

result = await db.execute(stmt)
data = []
for row in result.fetchall():
    data.append(row[0])
    ....

 

728x90

TO-BE

근데 생각해보니 이 정도 작업은 그냥 쿼리에서 하면 될거 같다. 그래서 다음과 같이 쿼리를 작성했다.

select
    string_agg(distinct keyword, ','),
    min(case when show_start_timestamp < now() then null else show_start_timestamp end),
    min(show_end_timestamp)
from
    tb_ads_keyword
        inner join tb_ads on tb_ads.ad_no = tb_ads_keyword.ad_no
where
    tb_ads.is_deleted = 'F'
and
    tb_ads_keyword.is_deleted = 'F'
and
    tb_ads.is_published = 'T'
and
    (tb_ads.show_start_timestamp >= now() or tb_ads.show_end_timestamp >= now())

이렇게 구현하면 keyword 리스트는 DB에서 ,으로 이어지는 문자열로 찾아주고 show_start_timestamp와 show_end_timestamp의 각 최소값을 가져오게 된다.

이 코드를 sqlalchemy 코드를 변환하면 다음과 같다.

stmt = select(
    func.string_agg(distinct(models.AdKeyword.keyword), literal_column("','")),
    func.min(case(
        [
            (models.Ad.show_start_timestamp <= func.now(), None),
        ],
        else_=models.Ad.show_start_timestamp
    )),
    func.min(models.Ad.show_end_timestamp),
).join(
    models.Ad,
    models.Ad.ad_no == models.AdKeyword.ad_no
).filter(
    models.AdKeyword.is_deleted == 'F',
    models.Ad.is_deleted == 'F',
    models.Ad.is_published == 'T',
    ((models.Ad.show_start_timestamp >= func.now()) | (models.Ad.show_end_timestamp >= func.now()))
)

이렇게 해서 결과를 얻으면 python에서는 루프를 돌릴 필요없이 바로 결과를 도출할 수 있다. 다음의 코드를 참조하면 된다.

result = await db.execute(stmt)
row = result.fetchone()
data = row[0].split(",")
min_datetime = row[2]
if row[1] is not None:
    min_datetime = min(row[1], row[2])

row[2]는 show_end_timestamp 값인데 이 값은 null인 값이 올 수 없기 때문에 show_start_timestamp를 case문으로 null로 처리한 부분에 대한 예외처리만 해주면 된다.

python 코드 자체는 성능상의 큰 차이가 없을 것이고 데이터 자체가 많지 않다면 굳이 이렇게 할 필요없이 위의 코드만으로도 충분히 잘 작동할 것이다.

하지만 코드를 늘 발전하고 더 나은 코드를 작성하기 위해서 노력해야하므로 이렇게 최적화를 진행하는 것은 늘 옳다고 생각한다.

반응형