한국도로공사(Korea Expressway Corporation, KEX) 공공데이터 OpenAPI를 Python에서 편하게 쓰기 위한 비공식 클라이언트 라이브러리입니다.
python-krex-api는 data.ex.co.kr와 data.go.kr에 흩어진 고속도로 교통량, 실시간 소통, 통행료, 영업소, 휴게소, 휴게소별 날씨, 기준정보 API를 한 인터페이스로 감싸고, 응답을 Pydantic 모델과 enum으로 변환합니다.
현재 진행 중인 작업은 CHANGELOG.md를 참고하세요.
| 필요 정보 | 문서 |
|---|---|
| 엔드포인트 상세 명세 | endpoints.md |
| 지원/live 검증 상태 | API_COVERAGE.md |
| 공통 코드표 · enum | codes.md |
| 에러 코드 매핑 | error-codes.md |
| 에이전트 구현 규칙 | SKILL.md, AGENTS.md |
| 구조적 의사결정 기록 | docs/decisions.md |
| 기여 가이드 | CONTRIBUTING.md |
| 변경 이력 | CHANGELOG.md |
- 네임스페이스형 클라이언트:
client.traffic.flow(),client.tollfee.between_tollgates()처럼 문서의 API 범주와 같은 구조로 호출합니다.
traffic.flow()는 공식 0405 실시간 소통 데이터를 한 번 조회한 뒤 노선·콘존·방향
필터와 페이지 분할을 로컬에서 적용합니다. 기본 페이지는 최대 1,000행이며
total_count는 필터 후 전체 건수입니다. 전체 수집에는
page = await client.traffic.flow_all()을 사용하세요. 인자 없이 한 응답의 모든 행을
Page[TrafficFlow]로 반환하며 페이지 크기를 추측할 필요가 없습니다.
flow()도 내부에서 flow_all()을 호출한 뒤 로컬 필터·페이지 처리를 적용합니다.
방향은 FlowDirection.START(S, 기점)/END(E, 종점)를 사용합니다.
기존 Direction.UP/DOWN/EAST/SOUTH와는 의미가 달라 flow() 필터에 사용할 수 없습니다.
updated_at은 KST로 취급하는 YYYYMMDDHHMM 문자열이며, 음수 속도는 None입니다.
같은 콘존의 여러 VDS 행은 그대로 유지하며 공식 vdsId는 item.vds_id로 제공합니다.
과거 응답에 이 필드가 없으면 None입니다. 상세 계약을 참고하세요.
- 두 포털 동시 지원:
data.ex.co.kr키(KEX_EX_API_KEY)와data.go.kr키(DATA_GO_KR_SERVICE_KEY)를 분리해 사용합니다. - 로컬
.env기본 로딩: 환경변수가 없으면 현재 작업 디렉터리부터 부모 디렉터리의.env를 찾아 키를 읽고, 복붙 과정에서 섞인 공백 문자를 제거합니다. - API 카탈로그: 구현된 API의 함수명, 데이터셋명, 포털, 엔드포인트, 서비스키 발급/활용신청 링크를
api_catalog()또는get_api_catalog()로 조회할 수 있습니다. - Python 타입 변환: 날짜, 숫자, Y/N 플래그, 코드값을
date,int,float,bool,StrEnum으로 변환합니다. - Pydantic 응답 모델: 공개 모델은 불변
BaseModel기반이라model_dump(),model_validate(),model_json_schema()를 외부 프로그램에서 바로 사용할 수 있습니다. - 명확한 예외 계층: 인증, 한도 초과, 파라미터 오류, 데이터 없음, 서버 오류, 파싱 오류, 네트워크 오류를 구분합니다.
- 본문 에러 코드 검사:
data.go.kr가 HTTP 200으로 반환하는 애플리케이션 에러도 놓치지 않습니다. - 네트워크 없는 테스트: fake session 기반으로 URL/쿼리/파싱/에러 매핑을 검증합니다.
export KEX_EX_API_KEY="data.ex.co.kr에서_발급받은_키"
export DATA_GO_KR_SERVICE_KEY="data.go.kr에서_발급받은_decoding_키"Windows PowerShell:
$env:KEX_EX_API_KEY="data.ex.co.kr에서_발급받은_키"
$env:DATA_GO_KR_SERVICE_KEY="data.go.kr에서_발급받은_decoding_키"또는 프로젝트 루트의 .env에 저장해도 됩니다. KrexClient()와 KrexClient.from_env()는 명시 인자가 없으면 환경변수를 먼저 보고, 없으면 가장 가까운 .env를 자동으로 읽습니다.
KEX_EX_API_KEY=data.ex.co.kr에서_발급받은_키
DATA_GO_KR_SERVICE_KEY=data.go.kr에서_발급받은_decoding_키웹 화면에서 키를 복사하며 줄바꿈, 탭, 앞뒤 공백이 섞여도 호출 전에 제거됩니다.
httpx에서 params=...로 전달하므로 data.go.kr 키는 Decoding 값을 권장합니다.
개발 중인 로컬 저장소:
pip install -e ".[dev]"PyPI 배포 후:
pip install python-krex-apiKrexClient는 native async 전용 클라이언트입니다. 모든 네트워크 namespace 메서드에 await를 사용하고 async with로 HTTP 세션을 정리합니다. 코드표와 로컬 기준정보 함수는 일반 함수로 유지합니다.
import streamlit as st
import asyncio
from krex import CarType, KrexClient
async def main() -> None:
async with KrexClient() as client:
# 실시간 소통
flows = await client.traffic.flow(route_no="0010")
for item in flows.items[:5]:
print(item.route_name, item.conzone_name, item.speed, item.congestion_level)
# 영업소간 통행료
fees = await client.tollfee.between_tollgates(
start_unit_code="101",
end_unit_code="105",
car_type=CarType.LIGHT,
)
print(fees.items[0].toll_fee)
# 휴게소 표준데이터(data.go.kr)
areas = await client.restarea.list_all(route_name="경부고속도로")
print(areas.items[0].name, areas.items[0].has_ev_charger)
# 노선별 휴게시설과 휴게소 주유소 가격(data.ex.co.kr)
facilities = await client.restarea.route_facilities(route_code="0010")
fuel_prices = await client.restarea.fuel_prices(
service_area_code=facilities.items[0].service_area_code
)
print(facilities.items[0].service_area_name, fuel_prices.items[0].gasoline_price)
# 휴게소별 날씨(data.ex.co.kr)
weather = await client.restarea.latest_weather(lookback_hours=72)
if weather.first:
print(weather.first.unit_name, weather.first.weather, weather.first.temperature)
asyncio.run(main())스크립트에서는 from_env()로 설정을 읽고 asyncio.run()으로 비동기 진입점을 실행합니다.
import asyncio
from krex import CarType, KrexClient
async def main() -> None:
async with KrexClient.from_env() as client:
flows = (await client.traffic.flow(route_no="0010"))
for item in flows.items[:5]:
print(item.route_name, item.conzone_name, item.speed, item.congestion_level)
fees = (await client.tollfee.between_tollgates(
start_unit_code="101",
end_unit_code="105",
car_type=CarType.LIGHT,
))
print(fees.items[0].toll_fee)
asyncio.run(main())환경변수 대신 명시적으로 키를 주입할 수도 있습니다. 테스트나 배치 작업에서는 이 방식이 더 읽기 쉽습니다.
from krex import KrexClient
client = KrexClient(
ex_api_key="data.ex.co.kr 키",
go_api_key="data.go.kr Decoding 키",
timeout=5.0,
max_retries=2,
)실제 운영 코드에서는 키를 코드에 직접 적지 말고 Secret Manager, .env, CI/CD secret 같은 외부 설정에서 읽어오세요. 로컬 개발에서는 .env가 기본값으로 로드되지만, 배포 환경에서는 Secret Manager나 배포 플랫폼의 secret을 권장합니다.
| 범주 | 메서드 | 원본 포털 | 반환 |
|---|---|---|---|
| 교통 | traffic.by_ic() |
data.ex.co.kr |
Page[TrafficByIc] |
| 교통 | traffic.by_route() |
data.ex.co.kr |
Page[dict] |
| 교통 | traffic.flow() |
data.ex.co.kr |
Page[TrafficFlow] |
| 교통 | traffic.flow_all() |
data.ex.co.kr |
Page[TrafficFlow] |
| 교통 | traffic.incident() |
data.ex.co.kr |
Page[Incident] |
| 교통 | traffic.vds_raw(), traffic.avc_raw() |
data.ex.co.kr |
Page[dict] |
| 통행료 | tollfee.between_tollgates() |
data.ex.co.kr |
Page[TollFee] |
| 통행료 | tollfee.tollgate_list() |
data.ex.co.kr |
Page[Tollgate] |
| 휴게소 | restarea.route_facilities() |
data.ex.co.kr |
Page[RestAreaRouteFacility] |
| 휴게소 | restarea.list_all() |
data.go.kr |
Page[RestArea] |
| 휴게소 | restarea.weather(), latest_weather() |
data.ex.co.kr |
Page[RestAreaWeather] |
| 휴게소 | restarea.fuel_prices() |
data.ex.co.kr |
Page[RestAreaFuelPrice] |
| 휴게소 | restarea.convenience_facilities() |
data.ex.co.kr |
Page[dict] |
| 휴게소 | restarea.food_price() |
data.ex.co.kr |
Page[FoodPrice] |
| 휴게소 | restarea.parking(), wifi(), restroom() |
data.ex.co.kr |
Page[dict] |
| 휴게소 | restarea.disabled_facility(), bus_transit() |
data.ex.co.kr |
Page[dict] |
| 시설 | facility.tollgate_info() |
data.go.kr |
Page[dict] |
| 시설 | facility.drowsy_shelter() |
data.ex.co.kr |
Page[dict] |
| 시설 | facility.shoulder_lane() |
data.go.kr |
Page[dict] |
| 행정 | admin.procurement_contracts() |
data.go.kr |
Page[dict] |
| 기준정보 | reference.api_catalog() |
내장 카탈로그 | tuple[ApiCatalogItem, ...] |
| 기준정보 | reference.routes() |
내장 코드표 | tuple[Route, ...] |
| 기준정보 | reference.common_codes() |
내장 코드표 | dict[str, dict[str, str]] |
상세 파라미터와 응답 필드는 endpoints.md를 기준으로 관리합니다.
구현된 API 목록은 라이브러리에서 바로 조회할 수 있습니다. dataset_name은 UI에 사람이 읽기 좋은 데이터셋명으로 표시하기 위한 값이고, service_key_url은 해당 포털의 인증키 발급 또는 활용신청 화면으로 연결됩니다.
from krex import get_api_catalog, get_api_catalog_item
for item in get_api_catalog(namespace="restarea"):
print(item.function, item.dataset_name, item.service_key_url)
weather = get_api_catalog_item("restarea.weather")
if weather:
print(weather.dataset_name)Streamlit 같은 외부 디버그 UI에서는 debug_call() 결과의 catalog 필드를 Debug Trace 탭에 그대로 표시할 수 있습니다.
import streamlit as st
from krex import KrexClient
import asyncio
async def main() -> None:
async with KrexClient.from_env() as client:
run = (await client.debug_call("restarea.weather", sdate="20210507", std_hour=12))
if run.catalog:
st.write("데이터셋", run.catalog["dataset_name"])
if run.catalog.get("service_key_url"):
st.link_button("서비스키 발급/활용신청", run.catalog["service_key_url"])
st.dataframe([run.catalog])
asyncio.run(main())이 저장소에는 같은 흐름을 바로 확인할 수 있는 예제 UI가 있습니다.
pip install -e ".[debug-ui]"
python -m streamlit run examples/streamlit_debug_ui.py --server.port 8504예제 UI는 API 선택 시 데이터셋명과 서비스키 받기 링크를 보여주고, 실행 시 Debug Trace 탭에서 DebugRun.catalog를 함께 표시합니다. 키 입력칸을 비워두면 KrexClient()와 동일하게 환경변수 또는 로컬 .env의 값을 기본으로 사용합니다.
영역별 구현/live 검증 상태는 API_COVERAGE.md를 기준으로 관리합니다. 검증되지 않은 포털 경로나 데이터셋은 무리해서 모델로 고정하지 않고 Page[dict]로 반환하며, 실제 응답 fixture가 쌓이면 Pydantic 모델로 승격합니다.
모든 목록형 API는 Page[T]를 반환합니다.
from krex import KrexClient
import asyncio
async def main() -> None:
async with KrexClient.from_env() as client:
page = (await client.traffic.flow(route_no="0010"))
page.items # tuple[TrafficFlow, ...]
page.first # TrafficFlow | None
len(page) # 현재 페이지 item 수
bool(page) # item이 있으면 True
page.page_no # int | None
page.num_of_rows # int | None
page.total_count # int | None
page.raw # 원본 응답 dict | None
asyncio.run(main())Page는 바로 순회할 수 있고, Page.items는 tuple입니다. 호출 이후 결과가 의도치 않게 바뀌는 일을 줄이기 위한 선택입니다.
for flow in page:
print(flow.conzone_name, flow.speed)공개 응답 모델은 Pydantic v2 모델입니다. 라이브러리 밖에서는 dict 변환, JSON schema 생성, 입력 검증을 별도 래퍼 없이 사용할 수 있습니다.
from krex import TrafficFlow
flow = page.first
if flow:
flow.model_dump()
TrafficFlow.model_json_schema()한국도로공사 data.ex.co.kr의 휴게소별 날씨 정보는 restarea.weather()로 특정 기준일/시각을 조회하고, restarea.latest_weather()로 최근 비어 있지 않은 시간대를 찾습니다.
import streamlit as st
from krex import KrexClient
import asyncio
async def main() -> None:
async with KrexClient.from_env() as client:
page = (await client.restarea.weather(sdate="20210507", std_hour=12))
latest = (await client.restarea.latest_weather(lookback_hours=72))
for row in latest.items[:3]:
print(row.unit_name, row.route_name, row.weather, row.temperature)
asyncio.run(main())-99, -99.000000 같은 한국도로공사 결측값은 RestAreaWeather의 typed 필드에서 None으로 정규화하고, 원문은 raw에 보존합니다.
코드값은 StrEnum 기반 enum으로 제공합니다. 문자열처럼 API 파라미터에 쓸 수 있으면서, 라벨과 선택지를 함께 제공합니다.
from krex import KrexClient
import asyncio
from krex import CarType, TCSType
async def main() -> None:
async with KrexClient.from_env() as client:
CarType.LIGHT.value # "1"
CarType.LIGHT.label # "1종"
CarType.from_label("1종")
CarType.choices() # (("1", "1종"), ...)
CarType.values() # ("1", "2", ...)
(await client.traffic.by_ic(
ex_div_code="00",
unit_code="101",
in_out="0",
time_unit="1",
tcs_type=TCSType.HIPASS,
car_type=CarType.LIGHT,
))
asyncio.run(main())외부 프로그램의 폼, CLI, OpenAPI schema, Pydantic validator에서는 choices()나 values()를 그대로 사용할 수 있습니다.
휴게소처럼 WGS84 위경도가 명확한 데이터는 모델의 lat/lon 필드에 바로 제공합니다. GeoJSON, WKT, 외부 GIS 전송 경계에서는 각 표준에 맞춰 (lon, lat) 순서로 직접 전달하세요.
from krex import KrexClient
import asyncio
async def main() -> None:
async with KrexClient.from_env() as client:
rest_area = (await client.restarea.list_all()).first
if rest_area and rest_area.lon is not None and rest_area.lat is not None:
geojson_position = (rest_area.lon, rest_area.lat)
asyncio.run(main())영업소처럼 원본 좌표계가 불명확한 데이터는 raw_coordinate도 함께 제공합니다.
from krex import KrexClient
import asyncio
async def main() -> None:
async with KrexClient.from_env() as client:
tollgate = (await client.tollfee.tollgate_list()).first
if tollgate and tollgate.raw_coordinate:
print(tollgate.raw_coordinate.x, tollgate.raw_coordinate.y, tollgate.raw_coordinate.system)
asyncio.run(main())휴게소 주소가 있는 모델은 provider가 준 원문 주소 문자열을 address에 제공합니다.
from krex import KrexClient
import asyncio
async def main() -> None:
async with KrexClient.from_env() as client:
facility = (await client.restarea.route_facilities()).first
if facility and facility.address:
print(facility.address)
asyncio.run(main())KEX 응답의 주소 문자열만으로는 10자리 법정동코드를 안전하게 확정하지 않습니다. 주소 정규화나 법정동코드가 필요하면 검증된 geocoder 또는 boundary lookup 결과를 별도로 결합하세요.
실제 포털 호출 없이 fake session을 주입할 수 있습니다.
from krex import KrexClient
class Session:
async def get(self, url, *, params, timeout):
...
client = KrexClient(ex_api_key="test-key", session=Session())새 엔드포인트를 추가할 때는 이 방식으로 쿼리 파라미터와 파싱 결과를 먼저 고정한 뒤, 필요하면 별도의 @pytest.mark.live 테스트를 추가합니다.
디버그 UI나 임시 확인 도구는 라이브러리를 직접 호출하되, Streamlit 같은 UI 의존성을 이 패키지에 넣지 않습니다. 대신 debug_call()로 실행 정보를 모으고 save_fixture()로 replay 가능한 JSON fixture를 저장합니다.
import asyncio
from krex import KrexClient, save_fixture
async def main() -> None:
async with KrexClient.from_env() as client:
run = (await client.debug_call("restarea.weather", sdate="20210507", std_hour=12))
print(run.catalog["dataset_name"]) # 한국도로공사_휴게소별 날씨
print(run.catalog["service_key_url"]) # 서비스키 발급/활용신청 링크
save_fixture(
base_dir="tests/fixtures",
function_name=run.function,
case_name="weather_normal",
description="휴게소 날씨 정상 응답",
input_data=run.input,
request_data=run.request,
response_data=run.response,
parsed_result=run.parsed,
processed_result=run.processed,
)
asyncio.run(main())Fixture 저장 전 key, serviceKey, Authorization, api_key, token 계열 필드는 <REDACTED>로 마스킹됩니다. 저장된 fixture는 tests/test_generated_fixtures.py가 외부 API 호출 없이 raw response를 다시 파싱해 회귀 테스트로 실행합니다.
현재 DebugRun.parsed와 DebugRun.processed는 같은 값입니다. 특정 엔드포인트에 별도 가공 단계가 생기면 tests/runners.py의 process 함수와 fixture의 processed 기대값을 함께 갱신하세요.
실제 서버 테스트는 기본 테스트에서 자동 실행되지 않습니다. 로컬 .env에 키를 저장하고 KEX_LIVE=1을 명시했을 때만 실행됩니다.
# .env
KEX_EX_API_KEY=발급받은_key
# PowerShell
$env:KEX_LIVE="1"
python -m pytest -m live -vv현재 live 테스트는 trafficIc와 trafficRoute를 소량 호출해 인증키, 실제 응답 shape, 빈 결과의 count=0 처리를 검증합니다.
import asyncio
from krex import KrexAuthError, KrexClient, KrexQuotaExceededError, KrexServerError
async def main() -> None:
async with KrexClient.from_env() as client:
try:
(await client.traffic.flow(route_no="0010"))
except KrexAuthError:
print("인증키를 확인하세요.")
except KrexQuotaExceededError:
print("API 요청 한도를 초과했습니다.")
except KrexServerError:
print("포털 장애 가능성이 있어 재시도 대상입니다.")
asyncio.run(main())예외 계층과 원본 코드 매핑은 error-codes.md에 정리되어 있습니다.
data.go.kr는 HTTP 200이어도response.header.resultCode가 실패일 수 있습니다.data.ex.co.kr인증키는key,data.go.kr인증키는serviceKey입니다.data.ex.co.kr는 HTTPS를 사용합니다. HTTP로 호출하면 리다이렉트될 수 있습니다.data.ex.co.kr응답은list대신 endpoint 이름(trafficIc등)을 top-level 배열 키로 사용할 수 있습니다.- 표준데이터 API는
_type이 아니라type=json을 쓰는 경우가 있습니다. - 영업소/노선/기관 코드는 선행 0이 의미 있으므로
int로 바꾸지 않습니다. - GeoJSON/GIS 경계용 좌표는 모델의
lon,lat값을 명시적으로(lon, lat)순서로 전달하세요. - 응답의
items.item,list,data는 단일dict또는list[dict]양쪽을 처리합니다. count=0은None이 아니라 정수0으로 보존해야 합니다.NO_DATA는 기본적으로KrexNotFoundError입니다. 빈 결과로 받고 싶으면KrexClient(strict_no_data=False)를 사용합니다.- 테스트 fixture의 숫자값은 실제 API처럼 문자열로 유지합니다. 그래야 변환 경계가 검증됩니다.
python -m compileall src/krex tests
python -m pytest
python -m pytest --cov=krex --cov-fail-under=90
python -m mypy src/krex
ruff check .기본 테스트는 실제 API를 호출하지 않아야 합니다. 실제 호출 테스트는 @pytest.mark.live로 분리하고 인증키가 없으면 skip하세요.
python-krex-api는 얇은 계층을 선호합니다.
| 모듈 | 책임 |
|---|---|
src/krex/client.py |
사용자용 KrexClient, 엔드포인트 네임스페이스, 모델 파싱 |
src/krex/catalog.py |
구현 API 카탈로그, 데이터셋명, 서비스키 발급/활용신청 링크 |
src/krex/_env.py |
로컬 .env 키 로딩 |
src/krex/_http.py |
HTTP 호출, retry, 포털별 envelope 정규화, 에러 매핑 |
src/krex/_convert.py |
문자열 기반 API 응답을 Python 타입으로 변환 |
src/krex/codes.py |
안정적인 코드값 enum과 라벨 |
src/krex/models.py |
public Pydantic 반환 모델 |
src/krex/exceptions.py |
예외 계층 |
src/krex/debug.py |
DebugRun, JSON 변환, 민감정보 마스킹, fixture 저장 |
새 기능을 넣을 때는 보통 src/krex/codes.py와 src/krex/models.py를 먼저 보강하고, src/krex/client.py에서 메서드를 연결한 뒤, tests/에서 fake 응답으로 쿼리와 변환을 잠급니다.
src/krex/
├── __init__.py
├── _env.py
├── catalog.py
├── client.py
├── _http.py
├── _convert.py
├── codes.py
├── exceptions.py
├── debug.py
├── models.py
└── py.typed
tests/
├── fixtures/
├── runners.py
├── utils.py
└── test_*.py
GPL-3.0-or-later. 자세한 조건은 LICENSE를 참고하세요. 이 라이선스는 이 저장소의 코드에만 적용됩니다.
원천 데이터의 저작권과 이용조건은 한국도로공사, 공공데이터포털, 각 데이터 제공기관 정책을 따릅니다. 이 프로젝트는 비공식 라이브러리이며 한국도로공사와 무관합니다. 실제 이용 전에는 각 제공기관의 최신 이용약관을 직접 확인하세요 — 본 안내는 법적 자문이 아니며 법적 효력을 보장하지 않습니다.
KrexClient와 모든 네트워크 서비스 메서드는 native async다. 조회와 디버그에는
await, 세션 관리에는 async with 또는 await client.aclose()를 사용한다.
동기 HTTP bridge, Async 접두사 클라이언트와 aio/adebug_call 별칭은 제거했다.
reference.routes, reference.common_codes, reference.api_catalog, 파싱과 코드표,
fixture 저장 등 로컬 유틸리티는 일반 함수다. 기존 서비스 인자·모델·선행 0 코드는 유지한다.
기존 TPS 변경의 기본값을 보존한다: max_rps=5.0, 버킷 capacity=1이다.
초기 버스트 없이 송신 사이에 최소 1/max_rps 간격을 둔다. 모든 서비스는 같은 버킷을
공유하고 각 재시도·리다이렉트·debug·latest_weather의 시간대 조회에도 토큰을 소비한다.
다른 라이브러리와 같은 구현의 AsyncTokenBucket을 rate_limiter=에 주입하면
여러 클라이언트가 요청 예산을 합산한다. 주입한 버킷이 max_rps보다 우선한다.
공통 버킷 클래스 자체의 기본 capacity는 max(1, max_rps)로 초기 burst를 허용하므로
Krex의 기본 간격을 유지하는 공유 버킷도 capacity=1을 명시한다.
버킷은 한 이벤트 루프에서 사용한다. 대기 취소는 토큰을 소비하지 않고 다음 대기자를 진행시킨다.
401/403/429는 즉시 실패한다. 5xx와 HTTPX 네트워크 오류는 기존 backoff를 유지하되 각 시도마다 새 토큰을 얻는다. 인자 검증 실패와 로컬 코드 조회는 토큰을 쓰지 않는다. HTTPX 리다이렉트는 계측한다. Digest/custom Auth처럼 내부 추가 송신이 가능한 인증 세션은 첫 전송 전에 거부한다. 인증 없는 세션과 HTTPX 기본 BasicAuth를 지원한다. 사용자 정의 transport 내부의 연결 재시도는 라이브러리 밖의 동작이다.
내부 HTTP 세션은 첫 요청에서 생성하고 클라이언트 종료 시 닫는다. session에 주입한
비동기 세션은 호출자가 닫는다.
내부 세션은 User-Agent: python-krex-api/<version>을 보낸다(data.ex.co.kr WAF가 2026-10-08부터
python-httpx/* 같은 라이브러리 기본 UA를 400 Request Blocked로 막는다). 주입한 세션은 그 헤더를 그대로 쓴다. 동기 get을 제공하는 세션은 거부한다.
디버그 기록은 ContextVar로 작업마다 격리하며 성공·실패·취소 뒤 복구한다.
원문으로 오류 분류·모델 파싱을 마친 후 예외와 진단 출력의 키를 마스킹한다.
사용 예제: docs/async-tps.md.
parse_rest_area_weather은 기존 HTTP client와 같은 파서의 공개 함수다.
사용법과 단일 행/장소 신뢰 경계를 따른다.