기본 콘텐츠로 건너뛰기

JSON Array Hook: Control Lists

thumbnail

`json.loads()`가 리스트를 그냥 리스트로만 만들던 시대가 끝난다: PR #146441의 `array_hook` 실전 용도

meta_description: CPython PR #146441은 JSON 디코더에 array_hook 파라미터를 추가해, JSON 배열을 파싱할 때 리스트 대신 원하는 타입으로 변환할 수 있게 했다. tuple/커스텀 리스트/검증 래퍼 등으로 일괄 변환하는 방법과, object_hook/object_pairs_hook과의 차이, 운영 코드에서 실수하기 쉬운 포인트(성능/메모리/예외 처리)를 정리한다. meta_keywords: python, json, loads, JSONDecoder, array_hook, object_hook, object_pairs_hook, decoder, parsing, tuple, list, validation, performance, CPython, practical, 운영, 변환 meta_robots: index,follow

JSON을 오래 만져본 사람은 다 안다.

  • object_hook는 있다(딕셔너리 후처리)
  • object_pairs_hook도 있다(순서 보존/중복 키 처리)

근데 이상하게도 “배열”에는 훅이 없었다.

그래서 우리는 늘 한 번 더 손을 댔다.

  • loads()로 일단 파싱하고
  • 리스트를 전부 걷어내며(재귀)
  • tuple로 바꾸거나, 커스텀 컨테이너로 감싸거나, 값 검증을 한다

PR #146441(gh-146440)은 그 빈 칸을 메운다.

JSON 디코더에 array_hook가 들어간다.

이제 “배열이 파싱되는 순간” 리스트를 다른 타입으로 바꾸는 길이 열린다.

1) 뭐가 추가됐나: load/loads와 JSONDecoder에 array_hook

diff를 보면 핵심은 단순하다.

  • json.load() / json.loads()에 array_hook= 인자가 추가된다
  • json.JSONDecoder(..., array_hook=...)가 가능해진다
  • 파이썬 구현과 C 구현 모두에 반영된다

즉, “문서만”이 아니라 실제 파서 경로에서 지원한다.

이제 기본 사용 형태는 이런 식이 된다.

import json

data = json.loads('[1, 2, 3]', array_hook=tuple)
assert data == (1, 2, 3)

이게 왜 유용하냐면, 후처리 재귀를 안 돌려도 되는 순간들이 있다.

2) object_hook와 뭐가 다르나(헷갈리기 쉬운 포인트)

정리하면 이렇다.

  • object_hook: JSON object {...} → dict가 만들어진 다음 후처리
  • object_pairs_hook: {...}의 key/value 쌍 리스트를 받아서 직접 컨테이너를 결정
  • array_hook: JSON array [...] → list가 만들어진 다음 후처리

즉, array_hook는 “배열 전용 object_hook”이다.

그래서 다음이 가능해진다.

  • “모든 배열은 tuple로” 같은 정책
  • “모든 배열은 FrozenList처럼 수정 불가능하게” 같은 정책
  • “배열 길이가 N 넘으면 예외” 같은 안전장치

3) 실전 용도 1: tuple로 고정해서 ‘의도치 않은 변경’을 막기

운영에서 JSON을 받은 뒤에, 아래 두 문제를 자주 본다.

1) 파싱한 데이터 구조를 여기저기 넘기다가 누군가가 리스트를 append/pop 한다 2) 그 변경이 로그/캐시/서명 검증 결과를 조용히 깨뜨린다

이때 배열을 tuple로 고정하면, 실수가 즉시 터진다(좋은 의미로).

import json

payload = json.loads(body, array_hook=tuple)
# payload 안의 모든 배열은 tuple이 됨

이 패턴은 특히 “외부 입력(JSON)”을 내부 도메인 모델로 빨리 굳히고 싶은 서비스에서 유용하다.

주의

배열을 tuple로 만들면, 이후 코드가 list를 기대하던 곳에서 깨질 수 있다.

  • 이건 array_hook의 단점이 아니라
  • “원래 애매하게 흘러가던 계약”이 드러나는 것이다

따라서 도입은 점진적으로(특정 endpoint/특정 파서부터) 하는 게 안전하다.

4) 실전 용도 2: 배열 크기/중첩 깊이 제한(DoS/폭주 방어)

JSON은 “구조”로 공격한다.

  • 지나치게 깊은 중첩
  • 지나치게 큰 배열

이게 Python 레벨에서 메모리와 시간을 잡아먹는다.

array_hook는 여기서 유용한 “마지막 방어선”이 된다.

import json

MAX_LEN = 10_000

def limit_array(a):
    if len(a) > MAX_LEN:
        raise ValueError(f"array too long: {len(a)}")
    return a

obj = json.loads(body, array_hook=limit_array)

이건 완벽한 보안 솔루션은 아니다. 하지만 “후처리로 순회하면서 크기를 재는 비용” 없이, 배열이 만들어질 때마다 바로 잘라낼 수 있다.

5) 실전 용도 3: 리스트 대신 ‘검증 래퍼’를 붙여서 문제 위치를 남기기

개인적으로 제일 마음에 드는 케이스는 이거다.

JSON 검증에서 가장 짜증나는 건 “어디 배열에서 터졌는지” 정보가 약하다는 점이다.

array_hook로 리스트를 감싸면, 나중에 에러가 났을 때 그 배열의 컨텍스트를 붙일 수 있다.

import json

class TaggedList(list):
    __slots__ = ("tag",)
    def __init__(self, it=(), *, tag=None):
        super().__init__(it)
        self.tag = tag


def wrap(a):
    # 여기서는 tag를 못 박기 어렵지만,
    # 최소한 타입을 바꿔서 "여기는 외부 입력"임을 표시할 수 있다.
    return TaggedList(a, tag="external-json")

obj = json.loads(body, array_hook=wrap)

이런 식으로 “외부 입력”을 내부 데이터와 구분해두면,

  • 후속 처리에서 검증을 빼먹었을 때
  • 타입으로라도 경고가 된다

(실무에서는 TaggedList까지는 과한데, 최소한 tuple 변환만 해도 효과가 있다.)

6) 성능/메모리 관점: array_hook는 ‘0원’이 아니다

훅이 들어오면 호출 비용이 생긴다.

그래도 대부분의 서비스에서 병목은 JSON 파싱 자체가 아니라 네트워크/IO/DB인 경우가 많고, array_hook는 “후처리 재귀”를 줄여서 오히려 이득이 될 수 있다.

하지만 아래 케이스는 조심하자.

  • 초당 수천~수만 req에서 큰 JSON을 파싱한다
  • 배열이 엄청나게 많은 데이터(작은 배열이 수천 개)

이럴 땐 훅 호출이 누적된다.

실무 팁:

  • array_hook=tuple처럼 가벼운 변환이 1순위다(여기서 무거운 일을 하지 말기)
  • “검증”은 훅에서 끝내지 말고, 훅은 신호만 남기고(길이/타입), 본 검증은 별도 단계로 빼라
  • 정말 빡센 트래픽이면, payload를 전부 강제 변환하는 대신 특정 엔드포인트/특정 요청군에만 켠다

내가 쓰는 계측 포인트(운영용)

array_hook를 넣으면, 체감상 느려졌는지 판단이 어려울 때가 있다. 이때는 훅 자체가 아니라 배열 개수와 총 원소 수를 찍어보면 답이 빨리 나온다.

import json

class C:
    arrays = 0
    elems = 0


def count_array(a):
    C.arrays += 1
    C.elems += len(a)
    return a  # 변환은 하지 않고, 카운트만

json.loads(body, array_hook=count_array)
print(C.arrays, C.elems)

이 값이 “생각보다 큰” 서비스라면, array_hook를 켜는 순간 훅 호출이 폭증한다. 그때는 변환 정책을 최소화하거나, 특정 데이터에만 적용하는 편이 낫다.

7) 언제 써야 하나(내 결론)

array_hook는 “있으면 좋은 옵션” 수준이 아니라, JSON을 도메인 모델로 굳히는 서비스에선 꽤 큰 도구다.

  • 외부 입력을 불변(immutable)로 만들고 싶다 → array_hook=tuple
  • 폭주 방어를 하고 싶다 → 길이 제한 훅
  • 파싱 직후 정책을 강제하고 싶다 → array_hook + object_hook 조합

반대로, 데이터 과학/ETL 쪽에서 “리스트를 바로 numpy로” 같은 걸 기대하면 오히려 실망할 수 있다. (그건 대개 파서 단이 아니라 변환/검증 파이프라인의 문제다.)

Keywords

python,json,decoder,array_hook,loads,load,JSONDecoder,object_hook,object_pairs_hook,tuple,list,parsing,validation,performance,DoS,limits,immutability,hook,pr

References

  • CPython PR #146441
    • https://github.com/python/cpython/pull/146441
  • Diff
    • https://github.com/python/cpython/pull/146441.diff

이미지 크레딧/라이선스

  • Macro laptop coding (Unsplash).jpg — Marc Mueller / CC0
    • https://commons.wikimedia.org/wiki/File:Macro_laptop_coding_(Unsplash).jpg
    • http://creativecommons.org/publicdomain/zero/1.0/deed.en
라벨:

댓글

댓글 쓰기

이 블로그의 인기 게시물

Django에서 트랜잭션 관리하기

Django에서 트랜잭션 관리하기 안녕하세요! 오늘은 Django에서 데이터베이스 트랜잭션을 효과적으로 관리하는 방법에 대해 알아보겠습니다. 1. 트랜잭션의 중요성 트랜잭션은 데이터베이스의 일관성과 무결성을 보장하는 중요한 개념입니다. Django에서는 여러 가지 방법으로 트랜잭션을 관리할 수 있습니다. 1.1 기본 개념 원자성(Atomicity) : 트랜잭션은 모두 실행되거나 모두 실행되지 않아야 합니다. 일관성(Consistency) : 트랜잭션 전후로 데이터베이스의 일관성이 유지되어야 합니다. 격리성(Isolation) : 동시에 실행되는 트랜잭션들이 서로 영향을 주지 않아야 합니다. 지속성(Durability) : 완료된 트랜잭션의 결과는 영구적으로 저장되어야 합니다. 2. Django의 트랜잭션 관리 2.1 기본 설정 # settings.py DATABASES = { 'default': { 'ENGINE': 'django.db.backends.postgresql', 'NAME': 'mydatabase', 'USER': 'myuser', 'PASSWORD': 'mypassword', 'HOST': 'localhost', 'PORT': '5432', 'ATOMIC_REQUESTS': True, # 모든 뷰를 트랜잭션으로 래핑 } } 2.2 데코레이터 사용 from django.db import transaction @transaction.atomic def create_order(user, items): order = Order.objects.create(user=...
댓글 쓰기
자세한 내용 보기

Python에서 asyncio 완전 정복 (await, async, gather 등)

어휴, 요즘 파이썬으로 비동기 프로그래밍 하는 재미에 푹 빠졌어요! 특히 asyncio 는 정말 마법 같더라고요. 처음엔 좀 낯설었는데, 익숙해지니까 속도 향상이 눈에 띄게 느껴져서 완전 반해버렸습니다. 이 글에선 제가 asyncio 를 배우면서 깨달은 점들을 풀어놓을게요. 혹시 비동기 프로그래밍이 뭔지 잘 모르시겠다면, 간단히 말해 여러 작업을 동시에 처리해서 프로그램 속도를 엄청나게 높이는 기술이라고 생각하시면 돼요. 마치 여러 요리사가 동시에 음식을 만들어서 손님에게 빨리 제공하는 것과 비슷하죠! 일단 async 와 await 라는 녀석들이 핵심인데요, async 는 함수 앞에 붙여서 "얘는 비동기 함수야!"라고 선언하는 거예요. 그리고 await 는 다른 비동기 함수가 끝날 때까지 기다리라고 지시하는 역할을 하죠. 예를 들어, 네트워크에서 데이터를 가져오는 함수가 있다면, await 를 사용해서 데이터가 다 가져올 때까지 기다렸다가 다음 작업을 진행할 수 있어요. 그 동안 다른 작업을 처리할 수 있으니, 마치 멀티태스킹을 하는 것처럼 느껴져요. 신기하지 않나요? 그리고 asyncio.gather 는 여러 비동기 함수를 동시에 실행하고 결과를 모아주는 아주 유용한 친구입니다. 제가 웹사이트 여러 개에서 데이터를 동시에 가져와야 할 때 정말 요긴하게 썼어요. 하나씩 순서대로 가져오는 것보다 훨씬 빠르더라고요! 마치 여러 개의 탭을 동시에 열어놓고 작업하는 것과 같다고 생각하시면 될 것 같아요. 실제로 제가 썼던 코드를 보여드릴게요. 세 개의 웹사이트에서 데이터를 가져오는 예제인데요. (아래 코드 삽입) 이 코드를 보시면, fetch_data 함수가 각 웹사이트에서 데이터를 가져오는 역할을 하고, asyncio.gather 가 이 함수들을 동시에 실행하도록 도와주는 것을 볼 수 있을 거예요. asyncio.sleep(2) 는 네트워크 지연을 시뮬레이션하기 위해 넣...
댓글 쓰기
자세한 내용 보기

AWS S3 + CloudFront로 정적 파일 서빙 완전 가이드

AWS S3 + CloudFront로 정적 파일 서빙 완전 가이드 안녕하세요! 오늘은 AWS S3와 CloudFront를 사용하여 정적 파일을 효율적으로 서빙하는 방법에 대해 알아보겠습니다. 왜 S3와 CloudFront를 사용할까요? 높은 가용성 : AWS의 글로벌 인프라를 활용 빠른 전송 속도 : CloudFront의 CDN 기능으로 전 세계 사용자에게 빠른 전송 비용 효율성 : 사용한 만큼만 지불 보안 : AWS의 보안 기능 활용 확장성 : 트래픽 증가에 자동 대응 1. S3 버킷 설정 1.1 버킷 생성 및 설정 import boto3 def create_s3_bucket(): s3 = boto3.client('s3') # 버킷 생성 bucket_name = 'your-static-files-bucket' s3.create_bucket( Bucket=bucket_name, CreateBucketConfiguration={ 'LocationConstraint': 'ap-northeast-2' } ) # 버킷 정책 설정 bucket_policy = { "Version": "2012-10-17", "Statement": [ { "Sid": "PublicReadGetObject", "Effect": "Allow", "Principal": "*", "Action": "s3:GetObje...
댓글 쓰기
자세한 내용 보기