API 레퍼런스

Sophonz Python SDK의 공개 API — init, init_sdk, shutdown, force_flush, 트레이스 속성, 예외 기록, SophonzOptions, Flask·FastAPI·Django 헬퍼, 다시 내보내는 OpenTelemetry API.

핵심 API는 sophonz.opentelemetry에서 내보냅니다. 프레임워크 헬퍼는 sophonz.opentelemetry.flask, sophonz.opentelemetry.fastapi, sophonz.opentelemetry.django에 있으며, 각각 사용할 때만 프레임워크를 import합니다. API의 어떤 함수도 텔레메트리 실패 때문에 예외를 던지지 않습니다.

수명 주기

`init(**options) -> bool`

권장 진입점입니다. 전달하지 않은 경우 Sophonz 기본값 console_capture=True와 experimental_exception_capture=True를 적용하고, 나머지는 그대로 init_sdk()에 넘깁니다. 설치된 모든 계측을 로드합니다.

import os
 
from sophonz.opentelemetry import init
 
init(service="checkout-api", api_key=os.getenv("SOPHONZ_API_KEY"))

SDK가 시작되면 True를 반환합니다. API 키, OTEL_EXPORTER_OTLP_HEADERS, service.key 리소스 속성이 모두 없거나, 이 프로세스에서 SDK가 이미 실행 중이면 경고를 남기고 False를 반환합니다.

`init_sdk(**options) -> bool`

init_sdk(
    *,
    service=None,
    api_key=None,
    console_capture=None,
    experimental_exception_capture=None,
    advanced_network_capture=None,
    beta_mode=None,
    disable_tracing=None,
    disable_logs=None,
    disable_metrics=None,
    detect_resources=None,
    stop_on_termination_signals=None,
    disable_startup_logs=None,
    instrumentations=None,
    additional_instrumentations=None,
    **options,
) -> bool

옵션별 기본값(인자, 환경 변수, 내장 기본값 순)만 적용해 SDK를 시작합니다. None은 설정하지 않은 것으로 봅니다. **options는 나머지 설정 옵션을 받습니다: service_version, service_namespace, deployment_environment, error_log_capture, promote_baggage_keys, sampler, 엔드포인트·프로토콜·insecure 옵션, debug, debug_payload, log_level. 모든 인자는 키워드 전용입니다. 반환값은 init()과 같습니다.

`shutdown(timeout_millis: int = 5000) -> None`

트레이서, 로거, 미터 프로바이더를 flush하고 종료하며, SDK가 교체한 시그널 핸들러와 예외 훅을 복원하고 전송을 멈춥니다. 이후 기록한 텔레메트리는 버려집니다. 여러 번 호출해도, init() 전에 호출해도 안전합니다. gunicorn의 worker_exit와 스크립트 끝에서 사용하세요.

`force_flush(timeout_millis: int = 5000) -> bool`

종료하지 않고 버퍼에 쌓인 것을 모두 내보냅니다. 프로바이더가 실패했거나 시간을 초과하면 False를 반환합니다.

`is_initialized() -> bool`

이 프로세스에서 SDK가 시작되었는지 여부입니다. init() 뒤에 fork된 프로세스는 True를 물려받습니다.

트레이스 속성

`set_trace_attributes(attributes: Mapping[str, AttributeValue]) -> None`

이 프로세스에서 현재 요청의 모든 스팬에 속성을 설정합니다: 같은 로컬 루트 스팬 아래 이미 열려 있는 스팬(서버 스팬 포함)과 이후 그 아래에서 시작하는 모든 스팬. 스팬에 이미 있는 속성은 덮어쓰지 않고, 값이 None인 항목은 건너뜁니다. 기록 중인 스팬 밖에서는 아무것도 하지 않습니다.

from sophonz.opentelemetry import set_trace_attributes
 
set_trace_attributes({"enduser.id": user.id, "tenant.id": user.tenant_id})

예외

`record_exception(exc, attributes=None, span=None, mechanism="generic", escaped=False) -> None`

파라미터	타입	설명
`exc`	`BaseException`	기록할 예외.
`attributes`	`Mapping[str, AttributeValue] \| None`	`exception` 이벤트에 추가됩니다.
`span`	`Span \| None`	기록할 스팬. 기본값은 현재 스팬입니다.
`mechanism`	`str`	`attributes`에 없으면 이벤트의 `exception.mechanism`으로 저장됩니다.
`escaped`	`bool`	OpenTelemetry의 `Span.record_exception`에 전달됩니다.

exception 이벤트를 추가하고 스팬 상태를 설명 <ExceptionType>: <message>와 함께 ERROR로 설정합니다. 기록 중인 스팬이 없으면 예외 타입 이름으로 스팬을 시작하고 attributes를 스팬 속성으로 붙인 뒤 끝내므로 예외가 사라지지 않습니다. 예외를 던지지 않습니다.

from sophonz.opentelemetry import record_exception
 
try:
    sync_inventory()
except Exception as exc:
    record_exception(exc, attributes={"job.name": "nightly-sync"})

디버깅

`enable_debug_payload_exporters() -> None`

실행 중인 트레이서·로거 프로바이더에 콘솔 익스포터를 추가해, 호출 이후 전송하는 모든 스팬과 로그 레코드를 stdout에도 출력합니다. init() 뒤에 호출하세요. debug_payload=True 또는 SOPHONZ_DEBUG_PAYLOAD=true는 시작 시점에 같은 일을 하며 메트릭도 출력합니다.

from sophonz.opentelemetry import enable_debug_payload_exporters, init
 
init(service="checkout-api")
enable_debug_payload_exporters()

옵션 객체

`SophonzOptions(**options)`

해석된 설정입니다. init_sdk()와 같은 키워드 전용 인자를 받고 같은 우선순위를 적용합니다. SDK가 내부에서 하나를 만들며, 특정 환경이 어떻게 해석되는지 보려면 직접 만들어 확인할 수 있습니다.

from sophonz.opentelemetry import SophonzOptions
 
options = SophonzOptions(service="checkout-api")
print(options.has_identity(), options.get_traces_endpoint())

멤버	반환값
`has_identity()`	API 키, `OTEL_EXPORTER_OTLP_HEADERS`, `service.key` 리소스 속성 중 하나가 있으면 `True`.
`get_traces_endpoint()`, `get_metrics_endpoint()`, `get_logs_endpoint()`	시그널별로 해석된 전송 URL.
`get_all_endpoints()`	위 세 URL의 목록.
`get_trace_headers()`, `get_metrics_headers()`, `get_logs_headers()`	전송 헤더: `OTEL_EXPORTER_OTLP_*_HEADERS` 값에 API 키를 `authorization`으로 더한 것, 없으면 `None`.
`get_trace_endpoint_credentials()`, `get_metrics_endpoint_credentials()`, `get_logs_endpoint_credentials()`	gRPC 채널 자격 증명, insecure이면 `None`. `grpc` extra가 필요합니다.
`apply_log_level()`	`opentelemetry`와 `sophonz` 로거를 해석된 `log_level`로 설정합니다.

모든 옵션은 해석된 값을 가진 속성으로도 있습니다(options.service, options.console_capture, options.instrumentations 등). SOPHONZ_PYTHON_SQLCOMMENTER는 options.sql_commenter입니다.

`version`

패키지 버전, "0.1.0".

Flask

from sophonz.opentelemetry.flask import ... — flask extra가 필요합니다.

`sophonz_flask_middleware(app, *, capture_baggage=True, capture_client_ip=True, capture_user_agent=True, attributes=None, instrument=True)`

before_request 훅을 등록해 서버 스팬에, 그리고 set_trace_attributes()를 통해 요청의 모든 스팬에 baggage.<key>, http.client.ip(request.remote_addr에서), http.user_agent, attributes를 붙입니다. attributes는 매핑 또는 Flask request를 받는 함수입니다. instrument=True이면 앱이 아직 계측되지 않았을 때 FlaskInstrumentor로 계측하고, 계측 패키지가 없으면 경고를 남깁니다. app을 반환합니다.

`setup_flask_error_handler(app)`

Flask의 got_request_exception 시그널을 구독해 요청 스팬에 예외를 기록합니다. 스팬을 Flask 계측이 소유하면 계측이 예외를 직접 기록하므로 exception.mechanism=flask만 추가합니다. app을 반환합니다.

FastAPI와 Starlette

from sophonz.opentelemetry.fastapi import ... — fastapi extra가 필요합니다.

`sophonz_fastapi_middleware(app, *, capture_baggage=True, capture_client_ip=True, capture_user_agent=True, attributes=None, instrument=True)`

SophonzASGIMiddleware를 가장 안쪽 사용자 미들웨어로 덧붙이므로, 등록 순서와 상관없이 서버 스팬 안에서 실행됩니다. instrument=True이면 앱이 아직 계측되지 않았을 때 FastAPIInstrumentor로 계측합니다. 앱이 요청을 받기 시작하기 전에 호출해야 하며, 그렇지 않으면 RuntimeError가 발생합니다. instrument=False로 두면 모든 Starlette 애플리케이션에서 동작합니다. app을 반환합니다.

`setup_fastapi_error_handler(app)`

SophonzExceptionMiddleware를 가장 안쪽 사용자 미들웨어로 덧붙입니다. 호출 시점 규칙은 위와 같습니다. app을 반환합니다.

`SophonzASGIMiddleware(app, capture_baggage=True, capture_client_ip=True, capture_user_agent=True, attributes=None)`

순수 ASGI 미들웨어입니다. http와 websocket 스코프에서는 Flask 헬퍼처럼 스팬을 보강하며, 클라이언트 주소는 scope["client"]에서 가져오고 attributes 함수는 ASGI scope를 받습니다. lifespan 스코프에서는 애플리케이션이 lifespan.shutdown.complete 또는 lifespan.shutdown.failed를 보내면 텔레메트리를 flush합니다.

`SophonzExceptionMiddleware(app)`

애플리케이션에서 빠져나간 예외를 기록한 뒤 다시 던지는 순수 ASGI 미들웨어입니다. HTTPException처럼 안쪽에서 응답으로 처리된 예외는 여기까지 오지 않습니다. 스팬을 FastAPI, Starlette, ASGI 계측이 소유하면 exception.mechanism=asgi만 추가합니다.

NOTE — 클래스를 직접 추가할 때

app.add_middleware(SophonzASGIMiddleware)는 이 미들웨어를 가장 바깥에 둡니다. 서버 스팬을 그보다 먼저 추가된 미들웨어가 만든다면, 보강이 그 스팬이 생기기 전에 실행됩니다. 가장 안쪽에 덧붙이는 두 헬퍼 함수를 쓰세요.

Django

from sophonz.opentelemetry.django import ... — django extra가 필요합니다.

`SophonzMiddleware`

MIDDLEWARE에 넣는 Django 미들웨어로, 동기와 비동기를 모두 지원합니다. REMOTE_ADDR과 HTTP_USER_AGENT를 사용해 Flask 헬퍼처럼 스팬을 보강합니다. process_exception은 Http404, PermissionDenied, BadRequest, SuspiciousOperation을 제외한 뷰 예외를 기록하고, None을 반환해 Django의 처리를 계속합니다.

선택 설정인 SOPHONZ_MIDDLEWARE로 구성하며, 미들웨어가 생성될 때 한 번 읽습니다.

SOPHONZ_MIDDLEWARE = {
    "capture_baggage": True,
    "capture_client_ip": True,
    "capture_user_agent": True,
    "attributes": "mysite.telemetry.request_attributes",  # dict, 호출 가능 객체, 또는 점 표기 경로
}

`instrument_django(**kwargs) -> None`

아직 계측되지 않았다면 kwargs를 instrument()에 넘겨 DjangoInstrumentor로 Django를 계측합니다. opentelemetry-instrumentation-django가 설치되어 있으면 init()이 이를 하므로, 계측을 로드하지 않고 SDK를 시작한 경우에만 직접 호출하세요. DJANGO_SETTINGS_MODULE을 설정한 뒤 호출하세요.

배포판

`sophonz.opentelemetry.distro.SophonzDistro`

opentelemetry-instrument가 로드하는 배포판으로, opentelemetry_distro 엔트리 포인트에 sophonz로 등록되어 있습니다. 환경 변수로 SDK를 시작하고, SOPHONZ_PYTHON_INSTRUMENTATIONS, SOPHONZ_PYTHON_SQLCOMMENTER, 고급 네트워크 캡처 설정을 각 계측에 적용합니다. 배포판이 여러 개 설치되어 있으면 OTEL_PYTHON_DISTRO=sophonz로 선택하세요.

OpenTelemetry API

sophonz.opentelemetry는 애플리케이션에서 흔히 쓰는 OpenTelemetry API 모듈을 다시 내보냅니다.

이름	실제 대상
`trace`	`opentelemetry.trace`
`metrics`	`opentelemetry.metrics`
`baggage`	`opentelemetry.baggage`
`context`	`opentelemetry.context`
`propagate`	`opentelemetry.propagate`
`SpanKind`, `Status`, `StatusCode`	`opentelemetry.trace`에서

from sophonz.opentelemetry import trace
 
tracer = trace.get_tracer("checkout")
 
with tracer.start_as_current_span("price-cart") as span:
    span.set_attribute("cart.size", len(items))

사용자 정의 스팬, baggage, 메트릭은 계측을 참고하세요.