FastAPI — 타입이 곧 계약이다 · 목록을 통째로 쌓지 않는 JSON Lines 내보내기 · 实验

목록을 통째로 쌓지 않는 JSON Lines 내보내기

목표

지연 생성·줄 경계·공개 필드·출력 상한을 HTTP 스트림과 연결합니다.

왜 중요한가

관리자가 모든 주문을 내려받자 서버 메모리가 급증했다. export 함수가 JSON을 만들기 전에 모든 행을 리스트에 모았기 때문이다. 줄 단위 출력으로 바꿨지만 본문 속 개행이 실제 줄 경계를 깨고 내부 원가도 그대로 나갔다. 스트리밍은 반환 타입 하나를 바꾸는 일이 아니라 지연 평가와 표현 계약을 함께 정하는 일이다.

단계

1. /root/work/fa-jsonl-export-lab/service.py에서 validate_row(row)는 dict이며 id가 bool 제외 양의 int, name이 비어 있지 않은 str일 때 row를 반환합니다. 나머지는 ValueError입니다. 추가 내부 필드는 허용합니다.

처음 한 번 준비하세요. 기존 파일은 덮어쓰지 않습니다.

mkdir -p /root/work/fa-jsonl-export-labtest -e /root/work/fa-jsonl-export-lab/service.py || cp /opt/fixtures/ten_labs/fa-jsonl-export-lab/service.py /root/work/fa-jsonl-export-lab/service.pycd /root/work/fa-jsonl-export-lab

2. /root/work/fa-jsonl-export-lab/service.py에서 project(row)는 validate_row 후 id와 name만 가진 새 dict를 반환합니다. 원본 내부 필드는 그대로 보존합니다.

3. /root/work/fa-jsonl-export-lab/service.py에서 encode_line(row)는 project 결과를 ensure_ascii=False, separators=(',',':'), sort_keys=True로 JSON 인코딩하고 마지막에 '
' 한 개를 붙인 str입니다. name 안 개행은 JSON 이스케이프여야 합니다.

4. /root/work/fa-jsonl-export-lab/service.py에서 validate_max(value)는 bool 제외 int 1~1000만 그대로 반환하고 그 외 ValueError입니다.

5. /root/work/fa-jsonl-export-lab/service.py에서 take_rows(rows, maximum)는 islice 등으로 최대 maximum개만 지연 반환하는 iterator입니다. 호출 시 maximum을 검증하며 한 번 next하면 입력을 한 번만 소비합니다.

6. /root/work/fa-jsonl-export-lab/service.py에서 json_lines(rows, maximum=100)는 take_rows에서 받은 행마다 encode_line을 yield합니다. 전부 합친 문자열이나 리스트를 반환하지 않습니다.

7. /root/work/fa-jsonl-export-lab/service.py에서 decode_lines(text)는 splitlines의 각 비어 있지 않은 줄을 json.loads 후 validate_row하고 리스트로 반환합니다. 빈 문자열은 [], 비어 있는 중간 줄은 ValueError입니다.

8. /root/work/fa-jsonl-export-lab/service.py에서 create_app(rows)는 GET /export에서 json_lines(rows, 100)을 application/x-ndjson StreamingResponse로 반환합니다. rows는 다시 순회 가능한 리스트입니다. 내부 필드가 없고 각 행의 내용과 순서를 보존해야 합니다.

참고

인터넷과 패키지 설치 없이 기존 lab-dev 환경에서 수행합니다.
각 단계는 45초 채점 예산 안에서 실행됩니다. 실제 sleep이나 네트워크 호출을 추가하지 마세요.
채점은 제출 모듈을 새로 불러오고 독립 입력과 임시 DB로 검사합니다. 예상값을 상수로 반환하는 대신 계약을 구현하세요.
[FastAPI 공식 문서](https://fastapi.tiangolo.com/) · [pytest 공식 문서](https://docs.pytest.org/en/stable/) · [Python sqlite3](https://docs.python.org/3/library/sqlite3.html)
한계: TestClient는 응답을 버퍼링하므로 네트워크 첫 바이트 지연이나 메모리 상한 전체를 증명하지 않는다. 지연 평가 여부는 별도의 계수 생성기로 검사한다. 스트림 시작 이후 잘못된 행을 만나면 정상 오류 JSON으로 상태를 바꾸기 어렵다. 실서비스는 사전 검증·행별 오류 형식·중단 정책 중 무엇을 택할지 정해야 한다.

8个步骤

행 계약을 검증한다
공개 행만 만든다
줄 경계를 보존하며 인코딩한다
출력 개수 상한을 검증한다
필요한 행만 소비한다
행을 지연 직렬화한다
내려받은 줄을 다시 검증한다
HTTP 다운로드를 완성한다