API 구현
서비스 클라이언트
HTTP 파이프라인
클라이언트 라이브러리는 일반적으로 하나 이상의 HTTP 요청을 래핑하므로 표준 네트워크 기능을 지원하는 것이 중요합니다. HTTP 파이프라인은 azure-coreHTTP 기반 Azure 서비스에 대한 연결을 제공하는 데 도움이 되는 라이브러리 의 구성 요소입니다 .
✅ HTTP 파이프라인을 사용하여 서비스 REST 엔드포인트에 요청을 보냅니다.
☑️ HTTP 파이프라인에 다음 정책을 포함해야 합니다:
고유 요청 ID( azure.core.pipeline.policies.RequestIdPolicy)
헤더( azure.core.pipeline.policies.HeadersPolicy)
원격 측정( azure.core.pipeline.policies.UserAgentPolicy)
프록시( azure.core.pipeline.policies.ProxyPolicy)
콘텐츠 디코딩( azure.core.pipeline.policies.ContentDecodePolicy)
재시도( azure.core.pipeline.policies.RetryPolicy및 azure.core.pipeline.policies.AsyncRetryPolicy)
자격 증명(예: BearerTokenCredentialPolicy, AzureKeyCredentialPolicy등)
분산 추적( azure.core.pipeline.policies.DistributedTracingPolicy)
로깅( azure.core.pipeline.policies.NetworkTraceLoggingPolicy)
from azure.core.pipeline import Pipeline
from azure.core.pipeline.policies import (
BearerTokenCredentialPolicy,
ContentDecodePolicy,
DistributedTracingPolicy,
HeadersPolicy,
HttpLoggingPolicy,
NetworkTraceLoggingPolicy,
UserAgentPolicy,
)
class ExampleClient(object):
...
def _create_pipeline(self, credential, base_url=None, **kwargs):
transport = kwargs.get('transport') or RequestsTransport(**kwargs)
try:
policies = kwargs['policies']
except KeyError:
scope = base_url.strip("/") + "/.default"
if hasattr(credential, "get_token"):
credential_policy = BearerTokenCredentialPolicy(credential, scope)
else:
raise ValueError(
"Please provide an instance from azure-identity or a class that implement the 'get_token protocol"
)
policies = [
HeadersPolicy(**kwargs),
UserAgentPolicy(**kwargs),
ContentDecodePolicy(**kwargs),
RetryPolicy(**kwargs),
credential_policy,
HttpLoggingPolicy(**kwargs),
DistributedTracingPolicy(**kwargs),
NetworkTraceLoggingPolicy(**kwargs)
]
return Pipeline(transport, policies)
맞춤형 정책
일부 서비스의 경우 사용자 지정 정책을 구현해야 할 수 있습니다. 예를 들어 사용자 지정 정책은 재시도, 요청 서명 또는 기타 전문 인증 기술 중에 보조 엔드포인트로 폴백을 구현할 수 있습니다.
☑️ 가능하면 정책 구현을 Azure-core로 사용해야 합니다.
✅ Azure SDK Architecture Board에서 제안된 정책을 검토합니다. 필요에 맞게 수정/매개변수를 지정할 수 있는 기존 정책이 이미 있을 수 있습니다.
✅ HTTPPolicy/Async에서 파생된 DOHTTP 정책(네트워크 호출이 필요한 경우) 또는 SANIOHTTP 정책(없는 경우)
✅ 사용자 지정 정책의 스레드 안전성을 보장합니다. 따라서 정책 인스턴스 자체가 아니라 요청별 또는 연결별 기록 데이터를 컨텍스트에 보관해야 합니다.
✅ 패키지에 있는 사용자 정의 정책을 문서화합니다. 이 문서에는 라이브러리 사용자가 정책을 사용하는 방법이 명시되어 있습니다.
서비스 방법
매개변수 검증
⛔️ 기본 제공 유형(예: str 등) 이외의 매개 변수 값 유형의 유효성을 검사하는 데 isinstance를 사용하면 안됩니다.
지원 유형
✅ 모델 유형에 대해 repr을(를) 구현합니다. 표현에는 유형 이름과 모든 주요 속성(모델 인스턴스를 식별하는 데 도움이 되는 속성)이 포함되어야 합니다.
✅ 1024자 뒤에 <repr>의 출력을 잘라냅니다.
확장 가능한 열거
SDK에 정의된 Enum은 대소문자를 구분하지 않는 문자열과 상호 교환할 수 있어야 합니다. 이는 aure-core에 정의된 CaseInsensitiveEnumMeta 클래스를 사용하여 수행됩니다.
from enum import Enum
from six import with_metaclass
from azure.core import CaseInsensitiveEnumMeta
class MyCustomEnum(with_metaclass(CaseInsensitiveEnumMeta, str, Enum)):
FOO = 'foo'
BAR = 'bar'
SDK 기능 구현
Logging
✅ Pythons 표준 로깅 모듈을 사용하세요
✅ 라이브러리에 대해 명명된 로거를 제공합니다.
패키지의 로거가 모듈 이름을 사용해야 합니다. 라이브러리는 하위 로거를 추가로 제공할 수 있습니다. 하위 로거가 제공된 경우 문서화합니다.
예를 들어:
패키지 이름: azure-someservice
모듈 이름: azure.someservice
로거 이름: azure.someservice
하위 로거: azure.someservice.achild
사용자는 이러한 이름 지정 규칙을 사용하여 모든 Azure 라이브러리, 특정 클라이언트 라이브러리 또는 클라이언트 라이브러리의 하위 집합에 대한 기록을 사용할 수 있습니다.
✅ 응용 프로그램이 복구할 가능성이 낮은 오류(예: 메모리 부족)에 대해 오류 로깅 수준을 사용해주세요
✅ 함수가 의도한 작업을 수행하지 못할 경우 WARNING 로깅 수준 level을 사용해주세요 이 기능은 또한 예외를 제기해야 합니다.
자체 복구 이벤트(예: 요청이 자동으로 재시도되는 경우)는 포함하지 않습니다.
✅ 함수가 정상적으로 작동하는 경우 INFO 로깅 수준을 사용해주세요
✅ 자세한 문제 해결 시나리오에 DEBUG 로깅 수준을 사용해주세요
DEBUG 로깅 수준은 개발자 또는 시스템 관리자가 특정 장애를 진단하기 위한 것입니다.
⛔️ DEBUG 이외의 로그 수준에서는 중요한 정보를 보내지 마십시오. 예를 들어 헤더를 기록할 때 계정 키를 수정하거나 제거합니다.
✅ 나가는 요청에 대한 요청 줄, 응답 줄 및 헤더를 INFO 메시지로 기록해주세요
✅ 서비스 호출이 취소된 경우 INFO 메시지를 기록해주세요
✅ 로그 예외가 WARNING 수준 메시지로 전달되었습니다. 로그 수준이 DEBUG로 설정된 경우 메시지에 스택 추적 정보를 추가합니다.
logging.Logger.isEnabledFor.를 호출하여 주어진 로거에 대한 로깅 수준을 결정할 수 있습니다
분산 추적
✅ 각 라이브러리 메서드 호출에 대해 새 추적 범위를 만듭니다. 가장 쉬운 방법은 azure.core.tracing에서 분산 추적 데코레이터를 추가하는 것입니다.
✅ 나가는 각 네트워크 호출에 대해 새 범위를 만듭니다. HTTP 파이프라인을 사용할 경우 새 범위가 작성됩니다.
✅ 나가는 각 서비스 요청에 추적 컨텍스트를 전파합니다.
원격 측정
클라이언트 라이브러리 사용 원격 측정은 고객이 서비스를 호출하는 데 사용하는 SDK 언어, 클라이언트 라이브러리 버전 및 언어/플랫폼 정보를 모니터링하기 위해 서비스 팀이 사용합니다. 클라이언트는 클라이언트 응용 프로그램의 이름 및 버전을 나타내는 추가 정보를 추가할 수 있습니다.
클라이언트가 zure-core에서 UserAgentPolicy를 사용하지 않을 경우의 고려 사항
✅ 라이브러리의 소비자가 application_id 매개 변수를 서비스 클라이언트 생성자에게 전달하여 응용 프로그램 ID를 설정할 수 있습니다. 이를 통해 사용자는 앱에 대한 서비스 간 원격 측정 기능을 얻을 수 있습니다.
✅ 응용 프로그램 ID의 길이가 24자 이하여야 합니다. 애플리케이션 ID가 짧을수록 서비스 팀은 진단 정보를 사용자 에이전트의 "플랫폼 정보" 섹션에 포함시킬 수 있으며, 소비자가 자신의 애플리케이션에 대한 원격 측정 정보를 얻을 수 있습니다.
테스트
✅ 테스트 프레임워크로 pytest를 사용해주세요
☑️ 비동기 코드 테스트에는 pytest-asyncio를 사용해야 합니다.
✅ 실시간 서비스에 대해 시나리오 테스트를 실행해주세요. 시나리오 테스트에는 Python Azure-DevTools 패키지를 사용하는 것이 좋습니다.
✅ Azure 가입 없이 오프라인에서 테스트를 실행할 수 있는 기록을 제공합해주세요.
✅ 동일한 구독에서 동시 테스트 실행을 지원해주세요
✅ 각 검사 사례를 다른 검사와 독립적으로 만들어 주세요
코드 분석 및 스타일 도구
✅ 암호에 pylint를 사용하세요. 리포지토리의 루트에 있는 pylintrc 파일을 사용합니다.
✅ flake8-doc 문자열을 사용하여 의사 의견을 확인해주세요
✅ 코드를 포맷하려면 검정색을 사용해주세요
☑️ MyPy를 사용하여 라이브러리의 공용 영역을 정적으로 확인해주세요
테스트 등 비배송 코드를 확인할 필요가 없습니다.
Azure Core 활용
zure-core 패키지는 클라이언트 라이브러리에 대한 공통 기능을 제공합니다. 설명서 및 사용 예는 azure/azure-sdk-for-python 저장소에서 찾을 수 있습니다.
HTTP 파이프라인
HTTP 파이프라인은 여러 정책으로 래핑되는 HTTP 전송입니다. 각 정책은 요청 또는 응답을 수정할 수 있는 제어 지점입니다. 클라이언트 라이브러리가 Azure 서비스와 상호 작용하는 방식을 표준화하기 위한 기본 정책 세트가 제공됩니다.
파이프라인의 Python 구현에 대한 자세한 내용은 설명서를 참조해주세요
프로토콜
설계 지침에서 요구하는 프로토콜 중 상당수는 기본 구현이 aure-core로 되어 있습니다.
LROPoller
T = TypeVar("T")
class LROPoller(Protocol):
def result(self, timeout=None) -> T:
""" Retrieve the final result of the long running operation.
:param timeout: How long to wait for operation to complete (in seconds). If not specified, there is no timeout.
:raises TimeoutException: If the operation has not completed before it timed out.
"""
...
def wait(self, timeout=None) -> None:
""" Wait for the operation to complete.
:param timeout: How long to wait for operation to complete (in seconds). If not specified, there is no timeout.
"""
def done(self) -> boolean:
""" Check if long running operation has completed.
"""
def add_done_callback(self, func) -> None:
""" Register callback to be invoked when operation completes.
:param func: Callable that will be called with the eventual result ('T') of the operation.
"""
...
azure.core.polling.LROPollerLROPoller프로토콜을 구현합니다 .
ItemPaged
T = TypeVar("T")
class ByPagePaged(Protocol, Iterable[Iterable[T]]):
continuation_token: "str"
class ItemPaged(Protocol, Iterable[T]):
continuation_token: "str"
def by_page(self) -> ByPagePaged[T] ...
aure.core.ItemPage는 ItemPage 프로토콜을 구현합니다.
자세한 내용은 ItemPaged 프로토콜을 참조하십시오.
DiagnosticsResponseHook
class ResponseHook(Protocol):
__call__(self, headers, deserialized_response): -> None ...
Python 언어 및 코드 스타일
✅ 본 문서에서 명시적으로 재정의하지 않는 한 PEP8의 일반 지침을 따라주세요
⛔️ 다른 언어에서 코딩 패러다임을 "차용" 하지 마세요
예를 들어, Java 커뮤니티에서 Reactive 프로그래밍이 아무리 일반적이더라도 대부분의 Python 개발자들에게는 아직 생소합니다.
✅ 동일한 서비스에 대해 다른 라이브러리보다 다른 Python 구성 요소와의 일관성을 선호합니다.
개발자가 여러 언어로 동일한 서비스를 사용하는 것보다 동일한 언어를 사용하여 여러 라이브러리를 사용할 가능성이 높습니다.
오류 처리
✅ 새 예외를 캐치 및 제기할 때 오류의 원래 원인을 포함하려면 예외 체인을 사용해주세요
# Yes:
try:
# do something
something()
except:
# __context__ will be set correctly
raise MyOwnErrorWithNoContext()
# No:
success = True
try:
# do something
something()
except:
success = False
if not success:
# __context__ is lost...
raise MyOwnErrorWithNoContext()
명명 규칙
✅ 변수, 함수 및 메서드 이름에 snake_case를 사용해주세요:
# Yes:
service_client = ServiceClient()
service_client.list_things()
def do_something():
...
# No:
serviceClient = ServiceClient()
service_client.listThings()
def DoSomething():
...
✅ type은 Pascal case을 사용해주세요 :
# Yes:
class ThisIsCorrect(object):
pass
# No:
class this_is_not_correct(object):
pass
# No:
class camelCasedTypeName(object):
pass
✅ 상수는 대문자를 사용합니다 :
# Yes:
MAX_SIZE = 4711
# No:
max_size = 4711
# No:
MaxSize = 4711
✅ 모듈 이름은 snake_case을 사용해주세요.
메서드 서명
⛔️ 정적 메서드(정적 메서드)를 사용하지 마세요. 대신 모듈 레벨 기능을 선호합니다.
정적 메서드는 드물고 일반적으로 다른 라이브러리에 의해 강제됩니다.
⛔️ 간단한 getter 및 setter 기능을 사용하지 마세요. 대신 속성을 사용해주세요.
# Yes
class GoodThing(object):
@property
def something(self):
""" Example of a good read-only property."""
return self._something
# No
class BadThing(object):
def get_something(self):
""" Example of a bad 'getter' style method."""
return self._something
⚠️ 5개 이상의 위치 매개변수가 필요한 메서드를 사용하면 안 됩니다. 키워드 전용 인수 또는 **kwargs를 사용하여 선택적/플래그 매개 변수를 수락할 수 있습니다.
위치 매개변수와 선택적 매개변수에 대한 일반적인 지침은 TODO: 삽입 링크를 참조해주세요
✅ Python 3만 지원하면 되는 모듈의 선택적 인수 또는 자주 사용되지 않는 인수에 키워드 전용 인수를 사용합니다.
# Yes
def foo(a, b, *, c, d=None):
# Note that I can even have required keyword-only arguments...
...
✅ 명확한 순서가 없는 인수에는 키워드 전용 인수를 사용 하지 마세요 .
# Yes - `source` and `dest` have logical order, `recurse` and `overwrite` do not.
def copy(source, dest, *, recurse=False, overwrite=False) ...
# No
def copy(source, dest, recurse=False, overwrite=False) ...
✅ 필요한 위치 매개변수가 세 개 이상인 메서드를 호출할 때 매개 변수 이름을 지정합니다.
def foo(a, b, c):
pass
def bar(d, e):
pass
# Yes:
foo(a=1, b=2, c=3)
bar(1, 2)
bar(e=3, d=4)
# No:
foo(1, 2, 3)
✅ 함수를 호출할 때 선택적 매개변수에 대한 매개변수 이름을 지정합니다.
def foo(a, b=1, c=None):
pass
# Yes:
foo(1, b=2, c=3)
# No:
foo(1, 2, 3)
Public vs “private”
✅ 이름이 공용 API의 일부가 아님을 나타내려면 선행 밑줄을 사용합니다. 공용 API가 아닌 API는 안정성이 보장되지 않습니다.
⛔️ 상속 계층에서 이름이 충돌할 가능성이 있는 경우가 아니면 선행 이중 밑줄 접두사 메서드 이름을 사용하지 마세요. 이름 충돌은 드물어요
✅ 모듈의 all 속성에 공개 메서드 및 유형을 추가해주세요
✅ 내부 모듈에는 선행 밑줄을 사용해주세요. 모듈이 내부 모듈의 하위 모듈인 경우 선행 밑줄을 생략할 수 있습니다.
# Yes:
azure.exampleservice._some_internal_module
# Yes - some_internal_module is still considered internal since it is a submodule of an internal module:
azure.exampleservice._internal.some_internal_module
# No - some_internal_module is considered public:
azure.exampleservice.some_internal_module
Types (or not)
✅ 명시적 유형 검사보다 구조 하위 유형 및 프로토콜을 선호합니다.
✅ 사용자 지정 매핑 유형을 제공하기 위한 추상 컬렉션 기본 클래스 collections.abc(또는 Python 2.7용 컬렉션)에서 파생됩니다.
✅ 공개적으로 문서화된 클래스 및 기능에 대한 유형 힌트 PEP484를 제공합니다.
Python 2.7 호환 코드에 대한 지침은 Python 2.7 및 2.7-3.x straddling 코드에 대해 제안된 구문을 참조하세요. Python 3 관련 코드(예: 비동기 클라이언트)에는 이 작업을 수행하지 마세요.
스레딩
✅ 사용자가 제공한 콜백에 대해 스레드 선호도를 유지하도록 명시적으로 문서화되지 않은 경우.
✅ 메서드(함수/클래스)가 스레드 세이프라는 사실을 설명서에 명시적으로 포함해주세요
예: asyncio.loop.call_soon_threadsafe,queue
☑️ 호출자가 병렬 처리를 위해 자체 스레드 또는 프로세스 관리를 정의하는 대신 실행자 인스턴스 전달을 허용해야 합니다.
스레드가 호출자에게 노출되지 않는 경우 사용자가 직접 스레드 관리를 수행할 수 있습니다. 예를 들어 LRPoller 구현에서는 백그라운드 폴러 스레드를 사용합니다.
