[ 글의 목적: django에서 sqlite 의 json field를 위한 contain look up 구현하기 ]

Django Sqlite JSON

Django ORM 에서 JSONField (또는 Arrary) 는 psql 과 궁합이 좋다. 그래서 django + psql 궁합을 많이 볼 수 있다. 기본 제공하는 file base sqlite3 DBMS는 sqlite 에 대한 JSONField 는 많은 부분이 미흡하다. 결론만 살펴보자면, JSONField 를 사용하는 경우 Django ORM 에서 contains 라는 look up 을 제공하지 않는다.

1. Sqlite3 JSON

• Django 3.1 JSONField update 를 보면 PostgreSQL, MySQL, SQLite 대상으로 JSONField 를 제공한다. 이전에는 PSQL 만 해당 field를 제공했다.

• django는 JSON 데이터를 데이터베이스에 저장할 때 "자동으로 시리얼라이즈" 하고, 데이터베이스에서 데이터를 검색할 때는 "자동으로 디시리얼라이즈" 하여 Python 객체로 변환해주고, JSON 데이터 유효성 검사를 수행하여 데이터가 유효한 JSON 형식인지 확인도 해준다.

• 근데 sqlite 대상으로 contains 라는 look up 을 제공하지 않는다는 것이다.. 근본적으로 sqlite 내부에 JSON function 만 있지 실제로는 TEXT field를 사용해서 생기는 일이다.

그래서 test DB, local DB 등으로 sqlite3 를 사용할때 JSONField 의 contains 가 작동을 안한다!! 3.2 버전의 JSON Django 내부코드 에서 확인 가능하다.

2. sqlite 에서 contains 구현하기

• 특히 비용을 아끼는 CI 환경 + test only DBMS 로 사용하는 경우, 이 부분이 굉장히 골치아프다는 것이다. 그도 그럴것이 text로 처리된 친구 대상으로 "어떻게 contains 를 체크하면 좋을까,," 라는 애매한 부분이 분명하게 존재한다.

• 왜냐? "[1, 2, 3, 4, 5]" 문자열인 친구 대상으로 [2, 3] 중 하나가 포함되어 있는지를 어떻게 simple 하게 할 수 있을까,, 반복된 OR 구문이 답이다..

1) django 의 get_lookup & as_sql 오버라이딩

• django에서 Lookup API reference 를 보면 아래와 같이 특정 model field 에 대한 lookup 을 "등록할 수 있다"

DateField.register_lookup(YearExact)
User._meta.get_field("date_joined").register_lookup(MonthExact)

• 그리고 How to write custom lookups 문서를 보면 from django.db.models import Lookup 를 통해 as_sql method 를 오버라이딩해서 나만의 커스텀 lookup 연산을 만들 수 도, 기존 연산을 커스텀할 수 도 있다.

• 즉 "author"."name" <> 'Jack' SQL 연산을 직접 이를 통해 구현하면 아래와 같을 것이다.

from django.db.models import Lookup


class NotEqual(Lookup):
    lookup_name = "ne"

    def as_sql(self, compiler, connection):
        lhs, lhs_params = self.process_lhs(compiler, connection)
        rhs, rhs_params = self.process_rhs(compiler, connection)
        params = lhs_params + rhs_params
        return "%s <> %s" % (lhs, rhs), params

특정Field.register_lookup(NotEqual)

• 사실 기존 연산 자체를 오버라이딩하는 것은 사이드 이펙트가 아주 기대된다. 특정 연산을 강제 오버라이딩 해버리면, 특히 like를 위한 contains, 어떤 판도라의 상자가 나를 기다리고 있을지 모른다!!

구현 전략

• 그래서 아예 custom field 를 만들고, 해당 field에 custom lookup 을 등록하되, sqlite 일 때만 다른 look up을 타게 만들어야 한다.

class CustomContains(Lookup):
    """sqlite3 를 위한 List 만큼의 Like query"""

    lookup_name = "customcontains"

    def as_sql(self, compiler: Any, connection: Any) -> Any:
    	...
        

class CustomJSONField(models.JSONField):
    """local DBMS, sqlite3 를 위한 `CustomJSONField` 오버라이딩 구현"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def get_lookup(self, lookup_name: str) -> Any | None:
        if lookup_name == "contains" and (
            settings.DEBUG == True or settings.TESTING == True
        ):
        	# 기존 contains 는 살려두고 특정 경우만 분기 쳐서 작동하게
            lookup_name = "customcontains"
        return super().get_lookup(lookup_name)

CustomJSONField.register_lookup(CustomContains)

• 이렇게 세팅해두고, 기존 JSONField 를 CustomJSONField 로, 필요한 경우, 사용하게 해야한다.

• 나의 경우 CustomJSONField 에 특정 pk 를 Array 형태로 저장해야 했고, 해당 모델의 pk 가 "포함되어 있는지" 를 체크해야 했었다.

• 물론 "왜 외래키로 안뺏지,,?" 가 당연하게 나오는 물음이다. 하지만 기존 구현 사항을 지키되, DBMS 에서 퍼포먼스 영향을 최대한 주지 않을 선택을 할 필요가 있었다. 애초에 해당 값이 채워지는 경우는 1만분의 1건 정도로 아주 아주 적었다.

queryset.filter(
	Q(branch=user.branch) | Q(accessible_branch__contains=[user.branch.id])
)

• 내가 구현해야 할 부분은 위였고, 해당 field로 연산이 들어가는 경우도 거의 많지 않았을 것이라, 이렇게 구현하게 되었다! 아 psql arrary field 가 매우 그립다.

2) sqlite 일때만 lookup field 바꾸기 구현

• 사실 sqlite3 에서는 JSON field opertation 이 존재 (JSON1 확장에 대한 문서) 한다. 그래서 key value 의 json 형태라면 JSON_EXTRACT DBMS 함수 활용 가능하다.

• 근데 array 형태라면, 또 바로 활용은 불가능하다. 한 단계가 더 필요하다!!

class CustomContains(Lookup):
    """sqlite3 를 위한 List 만큼의 Like query"""

    lookup_name = "customcontains"

    def as_sql(self, compiler: Any, connection: Any) -> Any:
        if connection.vendor == "sqlite":
            lhs, lhs_params = self.process_lhs(compiler, connection)
            rhs, rhs_params = self.process_rhs(compiler, connection)
            rhs_params = rhs_params[0]  # django에 의해 dubble list 로 만들어져 버림

            # Array[int] 형태의 rhs_params를 문자열 형태로 변환
            rhs_params = [str(param) for param in rhs_params]

            # JSON 배열 내 모든 요소에 대해 LIKE 연산 적용을 위한 SQL 조건 구성
            conditions = []
            for i, param in enumerate(rhs_params):
                condition = "JSON_EXTRACT(%s, '$[%d]') LIKE '%%%s%%'" % (lhs, i, param)
                conditions.append(condition)
            sql = " OR ".join(conditions)
            return sql, lhs_params
        else:
            raise NotImplementedError("This backend does not support this lookup.")

• 사실 위로 만들어지는 SQL 은 다음과 같다.

SELECT * FROM 테이블명 WHERE
JSON_EXTRACT(json_필드명, '$[0]') LIKE '%검색어1%' OR
JSON_EXTRACT(json_필드명, '$[1]') LIKE '%검색어2%' OR
JSON_EXTRACT(json_필드명, '$[2]') LIKE '%검색어3%';

• 그렇다,, [23, 9, 21, 98, 10] 이 피연산자 대상 필드 값고 이를 [32] 로 contains 연산을 하고 싶으면 올바르지 않은 연산 형태다. 이래서 contains를 제공 안하는 듯 ㅋㅋ 즉 for loop * 2 가 필요하다.

• 게다가 피연산자의 배열 길이를 over 해버린다면 (즉 3번째 요소가 없는데 $[2] 로 접근하면) DB SQL error 로 터진다. dammmmn..

process_lhs

• process_lhs 메서드는 쿼리의 왼쪽 부분(주로 데이터베이스 필드)을 처리하고, 필요한 SQL 표현과 파라미터를 준비 한다. 이 경우 "json_필드명" 이 될것 이다.

process_rhs

• process_rhs 메서드는 쿼리의 오른쪽 부분(주로 조건 값)을 처리하며, 이를 SQL 쿼리에 적합한 형태로 변환하고 파라미터를 준비 한다. 이 경우, list 형태로 저장된 pk, [1, 2, 3 ...] 가 해당한다.

3) Like query로 접근

• 어렵게 접근할 필요 없다. 어짜피 test only DBMS 이고, 실제 환경에서는 사용하지 않을 것이다. 하지만 테스트 결과와 조건은 명백하게 같아야 한다. 그래도 퍼포먼스에 대한 고민은 할 필요가 없기때문에, 다중 LIKE query 로 접근하면 된다.

class CustomContains(Lookup):
    """sqlite3 를 위한 List 만큼의 Like query"""

    lookup_name = "customcontains"

    def as_sql(self, compiler: Any, connection: Any) -> Any:
        if connection.vendor == "sqlite":
            lhs, lhs_params = self.process_lhs(compiler, connection)
            rhs, rhs_params = self.process_rhs(compiler, connection)
            rhs_params = rhs_params[0]  # django에 의해 dubble list 로 만들어져 버림

            # 검색어를 '%검색어%' 형태로 변환
            search_terms = ["'%%%s%%'" % param for param in rhs_params]

            # JSON 배열의 각 요소와 모든 검색어를 비교하는 SQL 조건문 생성
            conditions = []
            for term in search_terms:
                condition = "%s LIKE %s" % (lhs, term)
                conditions.append(condition)

            sql = " OR ".join(conditions)
            return sql, lhs_params
        else:
            raise NotImplementedError("This backend does not support this lookup.")

• 그러면 아래와 같은 SQL 형태가 만들어진다. (어짜피 sqlite는 json이 text라는 점!)

SELECT * FROM 테이블명 WHERE
json_필드명 LIKE '%검색어1%' OR
json_필드명 LIKE '%검색어1%' OR
json_필드명 LIKE '%검색어1%';

• git action을 통해 sqlite를 기반으로 한 공짜 CI 로 이득보는 나에게 아주 아주 좋은 솔루션이었다. (게다가 프로덕션 환경에서는 psql 을 쓰고 있으니 icontains 같은 꼼수도 불가능...)

• 만약 스테이징 DMBS 서버가 있거나 개발 DMBS 서버가 있다면,, CI 로 DMBS - psql 를 구성하시고 굳이 이런 미련한 오버라이딩은 안하는게 건강에 좀 좋지 않을까 생각한다.