Python 루프백 REST 점검: 성공·연결 실패를 CSV로 기록

REST API를 연결하기 전에 HTTP 응답과 저장 경로부터 분리해 확인하는 실습입니다. 외부 공개 API를 호출하는 예제가 아닙니다. 내 컴퓨터의 127.0.0.1에 작은 서버를 켜서 성공 한 행을 만들고, 그 서버를 종료한 뒤 같은 주소에서 연결 실패 한 행을 남깁니다.

첨부 ZIP의 loopback_server.py와 health_check.py는 아래 코드와 같습니다. 평균 점수 계산 코드나 외부 계정 정보는 들어 있지 않습니다. 기본 표의 성공·실패 값은 독자가 대조할 기대값입니다. 별도로 수행한 Python 검산의 환경과 결과는 아래 ‘분리된 Python 환경에서 확인한 8가지 분기’에 구분해 적었습니다. Microsoft Excel 실행 검증은 포함하지 않았습니다.

준비 환경과 점검 범위

Python 3.12 환경과 터미널 두 개를 준비합니다. 코드는 표준 라이브러리만 사용합니다. Excel은 만들어진 CSV를 읽고 필터링하는 데 쓰며, 네트워크 요청에는 관여하지 않습니다.

Python 3.12의 http.server는 HTTP 서버 클래스를 제공하지만 운영용 서버로 권장되지 않습니다. Python HTTP 서버 문서를 근거로 이 코드를 인터넷 공개 서버에 사용하지 않습니다. 공식 문서 확인일은 2026-09-12입니다.

이 서버는 폴더를 웹에 보여 주는 python -m http.server와 다릅니다. 파일 목록이나 현재 디렉터리를 읽지 않고, /health 요청에 고정 바이트 OK\n만 보냅니다. GET 응답을 점검하는 작은 HTTP 실습이며 REST의 인증·업무 데이터·쓰기 동작 전체를 구현하지 않습니다.

1. 파일을 노출하지 않는 로컬 서버 켜기

ZIP을 새 연습 폴더에 풀고 첫 터미널에서 다음 명령을 실행합니다. LOCAL_ONLY 안내가 나온 터미널은 다음 성공 요청이 끝날 때까지 열어 둡니다.

python3 loopback_server.py --port 8000

첨부의 서버 코드는 다음과 같습니다. 호스트를 외부 인터페이스로 바꾸지 않고, 이미 실행 중인 다른 프로그램을 종료해 포트를 확보하지 않습니다.

import argparse
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

class Handler(BaseHTTPRequestHandler):
    timeout = 5

    def do_GET(self):
        ok = self.path == "/health"
        payload = b"OK\n" if ok else b"NOT_FOUND\n"
        self.send_response(200 if ok else 404)
        self.send_header("Content-Type", "text/plain; charset=utf-8")
        self.send_header("Content-Length", str(len(payload)))
        self.send_header("Connection", "close")
        self.end_headers()
        self.wfile.write(payload)

    def log_message(self, format, *args):
        pass

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--port", type=int, default=8000)
    args = parser.parse_args()
    if not 1024 <= args.port <= 65535:
        parser.error("Use a local practice port from 1024 to 65535.")
    with ThreadingHTTPServer(("127.0.0.1", args.port), Handler) as server:
        print(f"LOCAL_ONLY http://127.0.0.1:{args.port}/health", flush=True)
        try:
            server.serve_forever()
        except KeyboardInterrupt:
            pass

if __name__ == "__main__":
    main()

Address already in use라면 해당 서버가 시작된 것이 아닙니다. 예를 들어 사용하지 않는 8766으로 바꾸고, 뒤의 점검 명령에도 --port 8766을 지정합니다. 다른 서비스가 8000번에서 응답한다는 이유만으로 이 실습 서버가 정상이라고 간주하지 않습니다.

2. 응답과 기록 실패를 나누는 점검기

health_check.py는 사용자에게 임의 URL을 받지 않습니다. 포트만 바꾸며 호스트와 경로는 코드에서 고정합니다. 프록시 자동 탐지와 리디렉션을 끄므로 로컬 점검이 다른 주소로 따라가는 상황을 줄입니다. 인증 정보도 보내지 않습니다.

urllib.request는 요청 처리기의 프록시·리디렉션 동작을 구성할 수 있고, timeout은 연결 같은 블로킹 작업에 적용됩니다. Python urllib.request 문서를 참고합니다. 여기서 timeout=3은 전체 프로그램이 반드시 3초 안에 끝난다는 뜻이 아닙니다. 파일 접근과 여러 I/O 단계의 총시간 상한으로 설명하지 않습니다.

import argparse
import csv
import http.client
import io
import json
import os
import time
import urllib.error
import urllib.request
from datetime import datetime, timezone
from pathlib import Path

FIELDS = ["checked_at_utc", "host", "http_status", "elapsed_ms",
          "sample_bytes", "result", "error_type"]
LIMIT = 1_000_000

class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None

def probe(port):
    row = dict.fromkeys(FIELDS, "")
    row.update(checked_at_utc=datetime.now(timezone.utc).isoformat(),
               host=f"127.0.0.1:{port}", sample_bytes=0, result="FAIL")
    started = time.perf_counter()
    opener = urllib.request.build_opener(
        urllib.request.ProxyHandler({}), NoRedirect()
    )
    request = urllib.request.Request(
        f"http://127.0.0.1:{port}/health",
        headers={"User-Agent": "loopback-practice/2.0"},
        method="GET",
    )
    try:
        with opener.open(request, timeout=3) as response:
            row["http_status"] = response.status
            body = response.read(4)
            row["sample_bytes"] = len(body)
            if response.status == 200 and body == b"OK\n":
                row["result"] = "PASS"
            else:
                row["error_type"] = "UnexpectedResponse"
    except urllib.error.HTTPError as exc:
        row["http_status"] = exc.code
        row["error_type"] = "HTTPError"
        exc.close()
    except urllib.error.URLError:
        row["error_type"] = "URLError"
    except (OSError, http.client.HTTPException) as exc:
        row["error_type"] = type(exc).__name__
    row["elapsed_ms"] = round((time.perf_counter() - started) * 1000, 1)
    return row

def append_row(output, row):
    if output.suffix.lower() != ".csv" or output.is_symlink():
        raise ValueError("Use a regular practice CSV path.")
    lock = Path(str(output) + ".lock")
    # Cooperative exclusion; never remove a lock this invocation did not create.
    with lock.open("x", encoding="ascii") as handle:
        handle.write(str(os.getpid()) + "\n")
    try:
        with output.open("a+b") as handle:
            handle.seek(0)
            previous = handle.read(LIMIT + 1)
            if len(previous) > LIMIT:
                raise ValueError("Practice CSV size limit exceeded.")
            if previous:
                if not previous.endswith(b"\n"):
                    raise ValueError("Incomplete existing CSV; preserve it for review.")
                records = list(csv.reader(
                    io.StringIO(previous.decode("utf-8-sig"), newline=""),
                    strict=True,
                ))
                if not records or records[0] != FIELDS:
                    raise ValueError("Existing CSV header differs; use a NEW filename.")
                if any(len(record) != len(FIELDS) for record in records[1:]):
                    raise ValueError("Existing CSV row width differs; stop.")
                if len(records) > 1000:
                    raise ValueError("Practice row limit exceeded.")
            buffer = io.StringIO(newline="")
            writer = csv.DictWriter(buffer, fieldnames=FIELDS, lineterminator="\n")
            if not previous:
                writer.writeheader()
            writer.writerow(row)
            encoding = "utf-8" if previous else "utf-8-sig"
            handle.write(buffer.getvalue().encode(encoding))
            handle.flush()
            os.fsync(handle.fileno())
    finally:
        lock.unlink()

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--port", type=int, default=8000)
    parser.add_argument("--out", default="api_health.csv")
    args = parser.parse_args()
    if not 1024 <= args.port <= 65535:
        parser.error("Use a local practice port from 1024 to 65535.")
    row = probe(args.port)
    append_row(Path(args.out), row)
    print(json.dumps(row, ensure_ascii=False))
    # A recorded connection failure is a valid diagnostic row, not a write error.

if __name__ == "__main__":
    try:
        main()
    except (OSError, ValueError, csv.Error) as exc:
        print("STOP:", type(exc).__name__,
              "Preserve the CSV/lock and inspect the reported inputs; do not blind-retry.")
        raise SystemExit(1) from None

응답은 최대 네 바이트만 읽습니다. 200과 정확한 OK\n이 함께 맞을 때만 PASS로 기록하므로, 다른 프로그램의 200 응답이나 더 긴 본문을 그대로 성공으로 취급하지 않습니다. 읽은 본문 자체는 CSV에 저장하지 않습니다.

CSV는 기존 내용을 덮어쓰는 대신 한 행을 추가합니다. 같은 코드끼리 기록이 겹치지 않도록 .lock 파일을 사용하고, 기존 파일의 헤더·열 개수·마지막 줄바꿈을 먼저 확인합니다. 이는 간단한 협력 잠금이지 다른 편집기까지 통제하는 저장 시스템은 아닙니다. 실행 중에는 Excel에서 해당 CSV를 편집하거나 저장하지 않습니다.

3. 켜진 서버와 꺼진 서버를 같은 CSV로 비교하기

다음 절차는 api_health.csv가 아직 없는 새 연습 폴더를 기준으로 합니다. 이미 파일이 있다면 보존하고, 두 요청에서 동일한 새 이름을 --out에 지정합니다.

1. 첫 터미널의 서버가 LOCAL_ONLY를 출력하고 대기 중인지 확인합니다.

2. 두 번째 터미널에서 아래 점검 명령을 한 번 실행합니다.

3. 첫 터미널에서 이번에 직접 켠 서버에만 Ctrl+C를 입력하고 종료를 기다립니다.

4. 같은 포트와 같은 CSV 이름으로 점검 명령을 한 번 더 실행합니다.

5. 이제 CSV를 Excel에서 열고 헤더를 제외한 두 행을 비교합니다.

python3 health_check.py --port 8000 --out api_health.csv

아래 표는 두 요청의 기대 판정입니다. 시각과 경과 시간은 독자가 실행할 때 측정하며, 첨부에는 측정하지 않은 시각이나 경과 시간을 넣지 않았습니다.

서버를 끈 뒤 다른 프로그램이 같은 포트를 차지했거나 OS가 다른 종류의 오류를 반환하면 기대 행과 다를 수 있습니다. 그 차이를 성공처럼 고쳐 적지 않습니다. 서버 터미널의 종료 상태, 선택한 포트, 실제 오류 종류를 함께 남깁니다.

분리된 Python 환경에서 확인한 8가지 분기

2026년 10월 2일 검산에서는 본문과 동일한 서버·점검기 코드를 Linux의 Python 3.12.14에서 실행했습니다. 서버는 127.0.0.1과 자동 배정된 사용 가능한 포트만 이용했고, 검사가 끝날 때마다 종료했습니다. 외부 API 접속, Excel 가져오기, 실제 서비스 가동률은 이 검산에 포함하지 않았습니다.

8개 보조 테스트가 모두 통과했습니다. 보조 테스트 스크립트나 이번 실행 로그가 기존 ZIP에 추가된 것은 아닙니다. 표는 위 코드의 제한된 분기를 확인한 기록이며, 다른 운영체제의 오류 이름이나 실제 API의 동작이 같다는 보장은 아닙니다. 기본 실습은 앞의 두 터미널 절차로 같은 PASS·FAIL을 다시 확인할 수 있습니다.

기록된 FAIL은 이번 실습에서는 필요한 관찰 결과입니다. CSV 저장 자체가 실패하면 프로그램은 STOP을 출력하고 비정상 종료합니다. “프로그램 종료 코드가 정상”과 “원격 연결이 성공”을 같은 의미로 사용하지 않도록 결과 열을 따로 읽습니다.

Excel에서 두 행을 읽는 방법

csv.DictWriter는 지정한 필드 순서로 CSV를 쓰며, 텍스트 파일로 사용할 때 newline을 명시해 줄바꿈 처리를 조정합니다. Python CSV 문서를 참고했습니다. 이 코드도 CSV 직렬화와 파일 인코딩을 분리하고 첫 기록에만 UTF-8 BOM을 붙입니다.

Excel의 텍스트/CSV 가져오기로 UTF-8과 쉼표 구분을 선택합니다. 열이 일곱 개이고 result가 여섯 번째 열인지 확인합니다. 현재 예제에서는 헤더를 제외한 행이 두 개이며, 필터로 PASS 한 행과 FAIL 한 행을 각각 볼 수 있어야 합니다.

표로 관리하려면 범위 A1:G3을 선택하고 머리글 포함 표로 만듭니다. 상태 개수를 집계할 때는 result의 개수를 셉니다. 수치가 없는 실패 행의 http_status 개수를 세면 실패를 빠뜨릴 수 있습니다. 피벗 테이블을 직접 만들 경우 행은 host, 열은 result, 값은 result의 개수로 설정합니다. 첨부에는 미리 생성된 피벗이나 XLSX는 없습니다.

시간은 성공과 실패의 원인을 설명하는 보조값입니다. 이 두 요청만으로 서비스 가동률이나 네트워크 성능을 일반화하지 않습니다. 이 예제의 주소는 숫자 루프백이므로 DNS 상태를 검사한 결과도 아닙니다.

실패에 따라 고칠 위치

잠금 파일이 있다고 즉시 지우지 않습니다. 기록 중인 프로세스가 없다는 사실과 CSV 끝부분이 완전하다는 점을 확인한 뒤 수동 복구해야 합니다. 확신이 없으면 기존 파일을 보존하고 별도의 새 실습 파일로 시작합니다. 장애 뒤 자동으로 같은 요청을 재실행하는 로직은 넣지 않았습니다.

CSV가 디스크 부족 등으로 일부만 기록될 수 있으므로 STOP 이후의 파일은 완료 보고서로 사용하지 않습니다. 끝이 잘린 파일에 헤더나 정답 행을 임의로 붙여 성공으로 만드는 대신, 원본을 남기고 원인을 해결한 뒤 새 파일에서 두 상태를 다시 관찰합니다.

운영 코드로 옮기기 전에 남는 한계

이 점검기의 PASS는 이번 로컬 HTTP 응답 조건이 맞았다는 뜻입니다. API 인증, TLS, 실제 응답 스키마, 업무 데이터의 최신성은 검사하지 않습니다. 프로덕션 REST API의 성공 판정에는 해당 서비스의 공식 계약과 별도 입력 검증이 필요합니다.

첨부의 파일 크기·행 수 제한은 작은 실습을 위한 값입니다. 대량 로그 수집, 여러 프로세스의 안정적인 기록, 원격 호출 주기와 재시도는 구현하지 않았습니다. 요청이 끝나면 이번 로컬 서버를 종료하고, 업무용 서버나 다른 프로그램은 그대로 둡니다.

로그를 분석하는 다음 단계는 WebSocket 수신 로그의 누락·중복 감사와 구분할 수 있습니다. HTTP 응답 여부와 이미 받은 이벤트의 완전성은 서로 다른 문제입니다. 결과 파일을 메일로 보내기 전에는 첨부 메일 미리보기 검산처럼 전송 전후 확인을 별도 단계로 설계합니다.

그림은 설명용이며 실제 실행 결과 화면이 아닙니다. 첨부 정답 기준과 독자가 실행해 얻은 CSV를 구별해 보관하십시오.

작성자 소개

최지원

엑셀·VBA·파이썬·AI 업무 자동화의 사용 절차와 실습 자료를 정리합니다.

WP 게시물 작성자의 글 보기