핵심 문법을 묶는 미니 프로젝트
시리즈 마지막 편에서는 지금까지 배운 문법을 한 흐름으로 엮어봅니다. 목표는 "학생 학습 리포트를 모아 요약하고 전송하는 자동화 도구"입니다. 로직은 단순하지만, 컴프리헨션·데코레이터·제너레이터·컨텍스트 매니저·타입 힌트·비동기까지 모두 등장합니다. 처음부터 모든 것을 동시에 붙이려 하지 말고, Stage별로 끊어가며 필요할 때 해당 챕터(예: 14편 데코레이터, 16편 컨텍스트 매니저, 17편 타입 힌트, 18편 비동기, 19편 패키징)를 다시 열어보세요.
이번 글에서 새로 나오는 용어
- 파이프라인: 입력→처리→출력 단계를 순서대로 연결한 데이터 흐름
- 비동기 알림:
async함수로 HTTP 같은 외부 전송을 동시에 처리하는 방식
개념 정리
학습 메모
- 소요 시간: 90~120분(모듈별 실습 포함)
- 준비물: 02~19편 전체 실습 경험, Typer CLI 빌드, pytest 기본
- 학습 목표: 로더→요약→알림→CLI 전체 파이프라인을 연결하고 테스트 아이디어까지 기록하기
- JSONL(JSON Lines)은 줄마다 하나의 JSON 객체가 있는 텍스트 형식입니다.
- 제너레이터는 파일을 한 줄씩 처리해 메모리를 절약합니다.
- TypedDict는 요약 데이터 구조를 명시해 협업을 돕습니다.
- Protocol은 비동기 알림 인터페이스를 코드로 표현하는 계약입니다.
- Stage 제목 옆 괄호에 참조할 챕터 번호를 붙여 두었으니, 모르는 개념이 보이면 그대로 찾아가면 됩니다.
코드로 이해하기
프로젝트 개요
- 입력: JSON 라인 파일(
reports.jsonl)에 저장된 학생별 진도 데이터 - 처리: 제너레이터로 한 줄씩 읽어 필터링, 컴프리헨션으로 통계 생성, 데코레이터로 로깅
- 출력: 비동기 HTTP API로 요약본 전송
구조는 다음과 같습니다.
studybot/
__init__.py
loader.py
summary.py
notifier.py
cli.py
모듈 사이 데이터 타입뿐 아니라 학습 챕터 번호를 나란히 적어두면, 다시 복습해야 할 지점을 빠르게 찾을 수 있습니다.
단계별 체크포인트
Stage 0. JSONL 샘플 데이터 만들기 (Core)
reports.jsonl 파일은 한 줄에 한 명의 학습 기록이 들어갑니다.
{"student": "민지", "module": "functions", "score": 95, "done": true}
{"student": "준호", "module": "async", "score": 81, "done": false}
- 입력 재료: 5~10줄만 있어도 충분합니다. 06편 파일 IO에서 다룬
Path.write_text로 생성해도 됩니다. - 체크포인트: 빈 줄이나 잘못된 JSON이 있다면 Stage 1에서
json.loads가 즉시 실패하므로 지금 정리합니다.
Stage 1. 데이터 로더 (제너레이터 + 컨텍스트 매니저, Core)
# loader.py
from collections.abc import Iterator
from pathlib import Path
def load_reports(path: Path) -> Iterator[dict]:
with path.open("r", encoding="utf-8") as f:
for line in f:
if line.strip():
yield json.loads(line)
with문으로 파일을 열고 자동으로 닫습니다.- 제너레이터 함수가 한 줄씩 파싱해 메모리를 절약합니다.
입력 → 출력
- 입력: JSONL 파일 경로
- 출력:
Iterator[dict]
입력 라인 1 → {'student': '민지', 'module': 'functions', ...}
입력 라인 2 → {'student': '준호', 'module': 'async', ...}
문제가 생기면 16편(컨텍스트 매니저)과 06편(파일 IO)을 복습해 자원 닫기를 다시 확인하세요.
Stage 2. 통계 요약 (컴프리헨션 + 타입 힌트, Core)
# summary.py
from typing import Iterable, TypedDict
class Summary(TypedDict):
student: str
avg_score: float
completed: list[str]
def summarize(records: Iterable[dict]) -> list[Summary]:
summaries = []
for student, items in group_by_student(records).items():
scores = [item["score"] for item in items]
completed = [item["module"] for item in items if item["done"]]
summaries.append(
{
"student": student,
"avg_score": sum(scores) / len(scores),
"completed": completed,
}
)
return summaries
def group_by_student(records: Iterable[dict]) -> dict[str, list[dict]]:
grouped: dict[str, list[dict]] = {}
for record in records:
grouped.setdefault(record["student"], []).append(record)
return grouped
- 리스트 컴프리헨션으로 조건부 항목을 추출합니다.
- TypedDict로 결과 구조를 명시합니다.
입력 → 출력
- 입력: Stage 1에서 넘겨준
Iterator[dict] - 출력:
list[Summary]
[
{
"student": "민지",
"avg_score": 94.5,
"completed": ["functions", "decorators"],
},
...
]
컴프리헨션이 헷갈리면 13편을, 데코레이터·타입 힌트가 낯설면 14·17편을 다시 열어본 뒤 이 코드와 비교하세요.
Stage 2 확장: 데코레이터 기반 로깅 (Optional)
# utils.py
def log_action(message: str):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
logging.info(message)
return func(*args, **kwargs)
return wrapper
return decorator
# summary.py
from .utils import log_action
@log_action("요약 생성")
def summarize(...):
...
데코레이터를 통해 공통 로그를 삽입합니다.
이 부분은 14편에서 연습한 패턴 그대로입니다. Stage 2에서 계산을 시작하기 전에 로그를 남기면 CLI와 테스트 모두에서 같은 메시지를 확인할 수 있습니다.
Stage 3. 비동기 알림 모듈 (Core → Plus)
# notifier.py
from typing import Protocol
class Notifier(Protocol):
async def send(self, payload: dict) -> None: ...
class WebhookNotifier:
def __init__(self, endpoint: str):
self.endpoint = endpoint
async def send(self, payload: dict) -> None:
async with httpx.AsyncClient(timeout=5) as client:
await client.post(self.endpoint, json=payload)
async def dispatch_all(notifier: Notifier, summaries):
tasks = [asyncio.create_task(notifier.send(summary)) for summary in summaries]
await asyncio.gather(*tasks)
- Protocol로 인터페이스를 문서화합니다.
async with은 비동기 컨텍스트 매니저로 HTTP 세션을 안전하게 닫습니다.
입력 → 출력
- 입력: Stage 2의
list[Summary], 웹훅 URL, 비동기 이벤트 루프 - 출력: 성공 시
None, 실패 시 예외(로그와 함께)
입력 Summary 1 → POST /webhook (payload 1)
입력 Summary 2 → POST /webhook (payload 2)
이 단계가 낯설면 18편(비동기), 12편(로깅), 16편(컨텍스트 매니저)을 다시 확인해 async with httpx.AsyncClient가 무엇을 보장하는지 복습하세요.
Stage 4. CLI 연결부 (Core)
# cli.py
from pathlib import Path
from .loader import load_reports
from .summary import summarize
from .notifier import WebhookNotifier, dispatch_all
app = typer.Typer(help="학습 리포트 요약 도구")
@app.command()
def run(path: str, webhook: str):
records = load_reports(Path(path))
summaries = summarize(records)
notifier = WebhookNotifier(webhook)
asyncio.run(dispatch_all(notifier, summaries))
입력 → 출력
- 입력: 터미널 명령
uv run studybot run reports.jsonl https://hooks.slack.com/... - 출력: 성공 로그, Slack 메시지 n건
$ uv run studybot run reports.jsonl $WEBHOOK
INFO 요약 생성 (Stage 2)
INFO Slack 전송 완료 (Stage 3)
CLI는 12편 Typer 구조와 15편 이터레이터·제너레이터 개념, 19편 패키징 흐름을 모두 재사용합니다. pyproject.toml에 [project.scripts] studybot = "studybot.cli:app"을 추가해 패키징까지 완료하세요.
테스트 아이디어 (Optional)
loader는 임시 파일을 만들어 한 줄씩 읽히는지 확인합니다.summarize는 고정 입력에 대해 평균과 완료 목록을 검증합니다.dispatch_all은httpx.MockTransport나 가짜 Notifier로 호출 횟수를 검사합니다.
왜 중요할까
- 시리즈 전반의 문법을 한 프로젝트에 녹여야 실전 감각이 완성됩니다.
- 제너레이터, 타입 힌트, 비동기, CLI가 서로 어떻게 연결되는지 체감할 수 있습니다.
- 테스트 아이디어를 함께 적어 두면 유지보수 계획까지 세울 수 있습니다.
- Stage 1
4를 차례로 밟으면 1219편의 학습 포인트를 순서대로 복습하게 되어, 필요한 챕터를 다시 열어보기 쉬운 구조가 됩니다. - 각 Stage를 끝낼 때마다 "무엇을 입력·출력했는가"를 메모하면 복습 루프가 훨씬 가벼워집니다.
실습
- 따라 하기:
load_reports,summarize,dispatch_all까지 최소 파이프라인을 연결하고 로컬 JSONL 샘플을 만들어 CLI로 실행합니다. - 확장하기:
log_action데코레이터를 추가해 단계별 로그를 출력하고typer.Option으로 출력 파일 경로 같은 설정을 받아 봅니다. - 디버깅: 알림 엔드포인트를 잘못 넣어 실패를 만든 뒤 예외 로그와 테스트 더블을 이용해 문제를 찾고 수정합니다.
- 완료 기준: CLI 실행 로그와 테스트 메모가 남아 있으며 추후 확장 아이디어를 백로그에 적어둘 수 있을 때입니다.
마무리
이 미니 프로젝트는 순수 Python만으로도 꽤 다양한 요구사항을 처리할 수 있음을 보여줍니다. 막히는 단계가 있다면 Stage와 괄호 속 챕터 번호를 보고 해당 글로 돌아가 복습하세요. 12편에서 만든 Typer CLI와 19편의 배포 체크리스트까지 다시 쓰이므로, 전편 내용이 실제로 어떻게 이어지는지 자연스럽게 복기할 수 있습니다. 다음과 같은 방향으로 확장해보세요.
pydantic같은 데이터 밸리데이션 도구 도입FastAPI또는Django백엔드와 연동rich로 CLI 출력을 꾸미고schedule로 주기 실행
시리즈를 따라오며 다진 기초는 어떤 프레임워크를 접하더라도 튼튼한 기반이 됩니다. 필요할 때마다 0111편의 기초 문법과 1219편의 심화 문법을 다시 펼쳐 이 캡스톤을 반복 확장해 보세요. 한 번에 모든 기능을 넣으려 하기보다, Stage별로 결과를 눈으로 확인하면서 다음 단계로 넘어가면 부담이 훨씬 줄어듭니다.
💬 댓글
이 글에 대한 의견을 남겨주세요