Python용 Azure Communication Chat Package 클라이언트 라이브러리 - 버전 1.2.0

이 패키지에는 채팅용 Azure Communication Services Python SDK가 포함되어 있습니다. Azure Communication Services 대한 자세한 내용은 여기를 참조하세요.

소스 코드 | 패키지(Pypi) | 패키지(Conda) | API 참조 설명서 | 제품 설명서

고지 사항

Python 2.7에 대한 Azure SDK Python 패키지 지원은 2022년 1월 1일에 종료되었습니다. 자세한 내용과 질문은 다음을 참조하세요. https://github.com/Azure/azure-sdk-for-python/issues/20691

시작

필수 구성 요소

• 이 패키지를 사용하려면 Python 3.7 이상이 필요합니다.
• 배포된 Communication Services 리소스. Azure Portal 또는 Azure PowerShell 사용하여 설정할 수 있습니다.

패키지 설치

Azure Communication Service Chat SDK를 설치합니다.

pip install --pre azure-communication-chat

사용자 액세스 토큰

사용자 액세스 토큰을 사용하면 Azure Communication Services에 직접 인증되는 클라이언트 애플리케이션을 빌드할 수 있습니다. azure.communication.identity 모듈을 사용하여 이러한 토큰을 생성한 다음, 이를 사용하여 Communication Services SDK를 초기화할 수 있습니다. azure.communication.identity 사용 예제:

pip install azure-communication-identity

from azure.communication.identity import CommunicationIdentityClient
identity_client = CommunicationIdentityClient.from_connection_string("<connection string of your Communication service>")
user = identity_client.create_user()
tokenresponse = identity_client.get_token(user, scopes=["chat"])
token = tokenresponse.token

user 이 토큰을 사용하여 만들 때 해당 사용자를 새 채팅 스레드의 참가자로 추가해야 하므로 위에서 만든 가 나중에 사용됩니다. 만들기 요청의 초기자가 채팅 스레드의 참가자 목록에 있어야 하기 때문입니다.

채팅 클라이언트 만들기

이렇게 하면 채팅 스레드를 생성, 가져오기, 나열 또는 삭제할 수 있습니다.

from azure.communication.chat import ChatClient, CommunicationTokenCredential

# Your unique Azure Communication service endpoint
endpoint = "https://<RESOURCE_NAME>.communcationservices.azure.com"
chat_client = ChatClient(endpoint, CommunicationTokenCredential(token))

채팅 스레드 클라이언트 만들기

ChatThreadClient를 사용하면 메시지 보내기, 메시지 가져오기, 채팅 스레드 토픽 업데이트, 채팅 스레드에 참가자 추가 등 채팅 스레드와 관련된 작업을 수행할 수 있습니다.

ChatClient를 사용하여 새 채팅 스레드를 만들어 가져올 수 있습니다.

create_chat_thread_result = chat_client.create_chat_thread(topic)
chat_thread_client = chat_client.get_chat_thread_client(create_chat_thread_result.chat_thread.id)

또한 클라이언트는 요청을 반복할 수 있도록 지시할 수도 있습니다. 즉, 클라이언트가 동일한 Idempotency-Token 사용하여 요청을 여러 번 수행하고 서버가 요청을 여러 번 실행하지 않고 적절한 응답을 다시 가져오는 경우입니다. Idempotency-Token 값은 요청의 모든 시간 동안 전역적으로 고유한 클라이언트 생성 식별자를 나타내는 불투명 문자열입니다.

create_chat_thread_result = chat_client.create_chat_thread(
    topic,
    thread_participants=thread_participants,
    idempotency_token=idempotency_token
)
chat_thread_client = chat_client.get_chat_thread_client(create_chat_thread_result.chat_thread.id)

또는 이전에 채팅 스레드를 만들었으며 해당 thread_id 있는 경우 다음을 통해 만들 수 있습니다.

chat_thread_client = chat_client.get_chat_thread_client(thread_id) # thread_id is the id of an existing chat thread

주요 개념

채팅 대화는 채팅 스레드로 표시됩니다. 스레드의 각 사용자를 스레드 참가자라고 합니다. 스레드 참가자는 1:1 채팅에서 비공개로 채팅하거나 1:N 그룹 채팅에서 서로 채팅할 수 있습니다. 또한 사용자는 다른 사용자가 입력하는 시기와 메시지를 읽은 시간에 대한 거의 실시간 업데이트를 받습니다.

클래스를 ChatClient 초기화한 후에는 다음 채팅 작업을 수행할 수 있습니다.

스레드 만들기, 가져오기, 업데이트 및 삭제

스레드에서 CRD(Create-Read-Delete) 작업 수행

create_chat_thread(topic, **kwargs)
list_chat_threads(**kwargs)
delete_chat_thread(thread_id, **kwargs)

클래스를 ChatThreadClient 초기화한 후에는 다음 채팅 작업을 수행할 수 있습니다.

스레드 업데이트

스레드 토픽에서 업데이트 작업 수행

update_topic(topic, **kwargs)

채팅 스레드 속성 가져오기

get_properties(**kwargs)

메시지 보내기, 가져오기, 업데이트 및 삭제

메시지에 대해 CRUD(Create-Read-Update-Delete) 작업 수행

send_message(content, **kwargs)
get_message(message_id, **kwargs)
list_messages(**kwargs)
update_message(message_id, content, **kwargs)
delete_message(message_id, **kwargs)

참가자 가져오기, 추가 및 제거

스레드 참가자에 대해 CRD(Create-Read-Delete) 작업 수행

list_participants(**kwargs)
add_participants(thread_participants, **kwargs)
remove_participant(participant_identifier, **kwargs)

입력 알림 보내기

서비스에 입력 알림 알림

send_typing_notification(**kwargs)

읽음 확인 보내기 및 받기

메시지가 읽혀지고 읽기 메시지 목록을 가져올 수 있음을 서비스에 알립니다.

send_read_receipt(message_id, **kwargs)
list_read_receipts(**kwargs)

예제

다음 섹션에서는 다음을 포함하여 가장 일반적인 작업 중 일부를 다루는 몇 가지 코드 조각을 제공합니다.

• 스레드 작업
• 메시지 작업
• 스레드 참가자 작업
• 이벤트 작업

스레드 작업

스레드 만들기

create_chat_thread 메서드를 사용하여 채팅 스레드를 만듭니다.

• 를 사용하여 topic스레드 토픽을 제공합니다.
• 선택 사항인 를 사용하여 thread_participants스레드에 추가할 목록을 ChatParticipant 제공합니다.
  • user, 필수, 사용자 액세스 토큰에서 CommunicationIdentityClient.create_user()에서 만든 입니다 CommunicationUserIdentifier .
  • 선택 사항인 display_name은 스레드 참가자의 표시 이름입니다.
  • 선택 사항인 share_history_time은 채팅 기록이 참가자와 공유된 시간입니다.
• 필요에 따라 를 사용하여 idempotency_token요청에 대한 고유 식별자를 지정합니다.

CreateChatThreadResult는 스레드를 만들 때 반환되는 결과이며, 생성된 채팅 스레드의 id를 페치하는 데 사용할 수 있습니다. 이 id는 get_chat_thread_client 메서드를 사용하여 ChatThreadClient 개체를 페치하는 데 사용할 수 있습니다. ChatThreadClient는 이 채팅 스레드에 대해 다른 채팅 작업을 수행하는 데 사용할 수 있습니다.

# Without idempotency_token and thread_participants
topic = "test topic"
create_chat_thread_result = chat_client.create_chat_thread(topic)
chat_thread_client = chat_client.get_chat_thread_client(create_chat_thread_result.chat_thread.id)

# With idempotency_token and thread_participants
from azure.communication.identity import CommunicationIdentityClient
from azure.communication.chat import ChatParticipant, ChatClient, CommunicationTokenCredential
import uuid
from datetime import datetime

# create an user
identity_client = CommunicationIdentityClient.from_connection_string('<connection_string>')
user = identity_client.create_user()

# user access tokens
tokenresponse = identity_client.get_token(user, scopes=["chat"])
token = tokenresponse.token

## OR pass existing user
# from azure.communication.chat import CommunicationUserIdentifier
# user_id = 'some_user_id'
# user = CommunicationUserIdentifier(user_id)

# create the chat_client
endpoint = "https://<RESOURCE_NAME>.communcationservices.azure.com"
chat_client = ChatClient(endpoint, CommunicationTokenCredential(token))

# modify function to implement customer logic
def get_unique_identifier_for_request(**kwargs):
    res = uuid.uuid4()
    return res

topic = "test topic"
thread_participants = [ChatParticipant(
    identifier=user,
    display_name='name',
    share_history_time=datetime.utcnow()
)]

# obtains idempotency_token using some customer logic
idempotency_token = get_unique_identifier_for_request()

create_chat_thread_result = chat_client.create_chat_thread(
    topic,
    thread_participants=thread_participants,
    idempotency_token=idempotency_token)
thread_id = create_chat_thread_result.chat_thread.id

# fetch ChatThreadClient
chat_thread_client = chat_client.get_chat_thread_client(create_chat_thread_result.chat_thread.id)

# Additionally, you can also check if all participants were successfully added or not
# and subsequently retry adding the failed participants again
def decide_to_retry(error, **kwargs):
    """
    Insert some custom logic to decide if retry is applicable based on error
    """
    return True

retry = [thread_participant for thread_participant, error in create_chat_thread_result.errors if decide_to_retry(error)]
if retry:
    chat_thread_client.add_participants(retry)

스레드 가져오기

get_properties 메서드를 통해 서비스에서 ChatThreadProperties를 검색합니다. thread_id는 스레드의 고유 ID입니다.

chat_thread_properties = chat_thread_client.get_properties()

채팅 스레드 나열

Use list_chat_threads 메서드는 생성된 채팅 스레드 목록을 검색합니다.

• 사용 results_per_page, 선택 사항, 페이지당 반환할 최대 메시지 수입니다.
• 사용 start_time, 선택 사항, 범위 쿼리가 있는 시작 시간입니다.

[ChatThreadItem]의 반복기는 스레드 목록에서 반환된 응답입니다.

from azure.communication.chat import ChatClient, CommunicationTokenCredential
from datetime import datetime, timedelta

token = "<token>"
endpoint = "https://<RESOURCE_NAME>.communcationservices.azure.com"
chat_client = ChatClient(endpoint, CommunicationTokenCredential(token))
start_time = datetime.utcnow() - timedelta(days=2)

chat_threads = chat_client.list_chat_threads(results_per_page=5, start_time=start_time)
for chat_thread_item_page in chat_threads.by_page():
    for chat_thread_item in chat_thread_item_page:
        print("thread id:", chat_thread_item.id)

스레드 토픽 업데이트

메서드를 사용하여 update_topic 스레드의 속성을 업데이트합니다. topic 는 스레드 토픽의 변경 사항을 설명하는 데 사용됩니다.

• 를 사용하여 topic 스레드에 새 토픽을 제공합니다.

topic = "new topic"
chat_thread_client.update_topic(topic=topic)

chat_thread = chat_thread_client.get_properties(thread_id)

assert chat_thread.topic == topic

스레드 삭제

메서드를 사용하여 delete_chat_thread 스레드 thread_id 를 삭제합니다. 는 스레드의 고유 ID입니다.

• 필수 thread_id를 사용하여 스레드의 고유 ID를 지정합니다.

chat_client.delete_chat_thread(thread_id=thread_id)

메시지 작업

메시지 보내기

메서드를 사용하여 send_message 로 식별된 스레드에 메시지를 보냅니다 thread_id.

• , 필수를 사용하여 content채팅 메시지 콘텐츠를 제공합니다.
• 선택 사항인 를 사용하여 chat_message_type채팅 메시지 유형을 제공합니다. 가능한 값은 다음과 같습니다. ChatMessageType.TEXT, ChatMessageType.HTML, 'text', 'html'; 지정 ChatMessageType.TEXT 하지 않으면 가 설정됩니다.
• ,선택적 을 사용하여 sender_display_name보낸 사람의 표시 이름을 지정합니다. 지정하지 않으면 빈 이름이 설정됩니다.

SendChatMessageResult는 메시지 전송 후 반환된 응답이며, 메시지의 고유 ID인 id를 포함합니다.

from azure.communication.chat import ChatMessageType

topic = "test topic"
create_chat_thread_result = chat_client.create_chat_thread(topic)
thread_id = create_chat_thread_result.chat_thread.id
chat_thread_client = chat_client.get_chat_thread_client(create_chat_thread_result.chat_thread.id)

content='hello world'
sender_display_name='sender name'
chat_message_type = ChatMessageType.TEXT

# without specifying sender_display_name and chat_message_type
send_message_result = chat_thread_client.send_message(content)
send_message_result_id = send_message_result.id
print("Message sent: id: ", send_message_result_id)

# specifying sender_display_name and chat_message_type
send_message_result_w_type = chat_thread_client.send_message(
            content,
            sender_display_name=sender_display_name,
            chat_message_type=chat_message_type # equivalent to chat_message_type = 'text'
)
send_message_result_w_type_id = send_message_result_w_type.id
print("Message sent: id: ", send_message_result_w_type_id)

메시지 가져오기

Use get_message 메서드는 서비스에서 message_id 메시지를 검색합니다. 는 메시지의 고유 ID입니다.

• 기존 메시지 ChatMessage 의 메시지 ID를 지정하려면 ,필수를 사용합니다message_id. 메시지 가져오기에서 반환된 응답이고, 메시지의 고유 ID인 ID를 포함하고, 다른 필드는 azure.communication.chat.ChatMessage를 참조하세요.

chat_message = chat_thread_client.get_message(message_id=send_message_result_id)
print("get_chat_message succeeded, message id:", chat_message.id, "content: ", chat_message.content)

메시지 나열

Use list_messages 메서드는 서비스에서 메시지를 검색합니다.

• 사용 results_per_page, 선택 사항, 페이지당 반환할 최대 메시지 수입니다.
• 사용 start_time, 선택 사항, 범위 쿼리가 있는 시작 시간입니다.

[ChatMessage]의 반복기는 메시지 목록에서 반환된 응답입니다.

from datetime import datetime, timedelta

start_time = datetime.utcnow() - timedelta(days=1)

chat_messages = chat_thread_client.list_messages(results_per_page=1, start_time=start_time)
for chat_message_page in chat_messages.by_page():
    for chat_message in chat_message_page:
        print("ChatMessage: Id=", chat_message.id, "; Content=", chat_message.content.message)

메시지 업데이트

threadId 및 messageId로 식별되는 메시지를 업데이트하는 데 사용합니다 update_message .

• 필수 message_id는 메시지의 고유 ID입니다.
• 필요에 따라 content를 사용하여 업데이트할 메시지 콘텐츠를 지정합니다. 지정하지 않으면 비어 있는 것으로 할당됩니다.

content = "updated message content"
chat_thread_client.update_message(send_message_result_id, content=content)

chat_message = chat_thread_client.get_message(message_id=send_message_result_id)

assert chat_message.content.message == content

메시지 삭제

를 사용하여 delete_message 메시지를 삭제합니다.

• 사용 message_id, 필수는 메시지의 고유 ID입니다.

chat_thread_client.delete_message(message_id=send_message_result_id)

스레드 참가자 작업

스레드 참가자 나열

list_participants를 사용하여 스레드의 참가자를 검색합니다.

• results_per_page(선택 사항)를 사용합니다. 페이지당 반환될 최대 참가자 수입니다.
• skip(선택 사항)을 사용하여 응답에서 지정된 위치까지 참가자를 건너뜁니다.

[ChatParticipant]의 반복기는 참가자 목록에서 반환된 응답입니다.

chat_participants = chat_thread_client.list_participants(results_per_page=5, skip=5)
for chat_participant_page in chat_participants.by_page():
    for chat_participant in chat_participant_page:
        print("ChatParticipant: ", chat_participant)

스레드 참가자 추가

메서드를 사용하여 add_participants 스레드 참가자를 스레드에 추가합니다.

• 를 사용하여 thread_participants스레드에 추가할 를 ChatParticipant 나열합니다.
  • user, 필수, CommunicationUserIdentifier 사용자 액세스 토큰에서 CommunicationIdentityClient.create_user()로 만든 입니다.
  • 선택 사항인 display_name은 스레드 참가자의 표시 이름입니다.
  • 선택 사항인 share_history_time은 채팅 기록이 참가자와 공유된 시간입니다.

list(tuple(ChatParticipant, ChatError))가 반환됩니다. 참가자가 성공적으로 추가되면 빈 목록이 필요합니다. 참가자를 추가하는 동안 오류가 발생한 경우 발생한 오류와 함께 실패한 참가자로 목록이 채워집니다.

from azure.communication.identity import CommunicationIdentityClient
from azure.communication.chat import ChatParticipant
from datetime import datetime

# create 2 users
identity_client = CommunicationIdentityClient.from_connection_string('<connection_string>')
new_users = [identity_client.create_user() for i in range(2)]

# # conversely, you can also add an existing user to a chat thread; provided the user_id is known
# from azure.communication.chat import CommunicationUserIdentifier
#
# user_id = 'some user id'
# user_display_name = "Wilma Flinstone"
# new_user = CommunicationUserIdentifier(user_id)
# participant = ChatParticipant(
#     identifier=new_user,
#     display_name=user_display_name,
#     share_history_time=datetime.utcnow())

participants = []
for _user in new_users:
  chat_participant = ChatParticipant(
    identifier=_user,
    display_name='Fred Flinstone',
    share_history_time=datetime.utcnow()
  )
  participants.append(chat_participant)

response = chat_thread_client.add_participants(thread_participants=participants)

def decide_to_retry(error, **kwargs):
    """
    Insert some custom logic to decide if retry is applicable based on error
    """
    return True

# verify if all users has been successfully added or not
# in case of partial failures, you can retry to add all the failed participants
retry = [p for p, e in response if decide_to_retry(e)]
if retry:
    chat_thread_client.add_participants(retry)

스레드 참가자 제거

remove_participant 메서드를 사용하여 threadId로 식별되는 스레드에서 스레드 참가자를 제거합니다. identifier 는 에서 CommunicationUserIdentifier CommunicationIdentityClient.create_user()에서 만든 입니다. azure-communication-identity

및 가 이 채팅 스레드에 추가되었습니다.

• 를 사용하여 identifier 만든 을 지정합니다 CommunicationUserIdentifier .

chat_thread_client.remove_participant(identifier=new_user)

# # conversely you can also do the following; provided the user_id is known
# from azure.communication.chat import CommunicationUserIdentifier
#
# user_id = 'some user id'
# chat_thread_client.remove_participant(identifier=CommunicationUserIdentifier(new_user))

이벤트 작업

입력 알림 보내기

메서드를 사용하여 send_typing_notification 사용자를 대신하여 스레드에 입력 알림 이벤트를 게시합니다.

chat_thread_client.send_typing_notification()

읽음 확인 보내기

메서드를 사용하여 send_read_receipt 사용자를 대신하여 스레드에 읽기 확인 이벤트를 게시합니다.

• 를 사용하여 message_id 읽음 확인을 보낼 메시지의 ID를 지정합니다.

content='hello world'
send_message_result = chat_thread_client.send_message(content)
send_message_result_id = send_message_result.id
chat_thread_client.send_read_receipt(message_id=send_message_result_id)

읽음 확인 나열

Use list_read_receipts 메서드는 스레드에 대한 읽기 확인을 검색합니다.

• 사용 results_per_page, 선택 사항, 페이지당 반환할 최대 읽기 확인 수입니다.
• (선택 사항)을 사용하여 skip응답에서 지정된 위치까지 읽은 영수증을 건너뜁니다.

[ChatMessageReadReceipt]의 반복기는 읽음 확인 목록에서 반환된 응답입니다.

read_receipts = chat_thread_client.list_read_receipts(results_per_page=5, skip=5)

for read_receipt_page in read_receipts.by_page():
    for read_receipt in read_receipt_page:
        print(read_receipt)
        print(read_receipt.sender)
        print(read_receipt.chat_message_id)
        print(read_receipt.read_on)

샘플 코드

Azure Communication Chat 클라이언트 라이브러리를 사용하는 일반적인 시나리오 작업을 보여 주는 코드 샘플입니다. 샘플의 비동기 버전(와 함께 _async추가된 python 샘플 파일)은 비동기 작업을 표시합니다. 샘플 코드를 실행하기 전에 필수 구성 요소를 참조하세요.

리소스를 만든 다음, 일부 환경 변수를 설정합니다.

set AZURE_COMMUNICATION_SERVICE_ENDPOINT="https://<RESOURCE_NAME>.communcationservices.azure.com"
set COMMUNICATION_SAMPLES_CONNECTION_STRING="<connection string of your Communication service>"

pip install azure-communication-identity

python samples\chat_client_sample.py
python samples\chat_client_sample_async.py
python samples\chat_thread_client_sample.py
python samples\chat_thread_client_sample_async.py

문제 해결

문제가 발생했나요? 이 섹션에는 수행할 작업에 대한 세부 정보가 포함되어야 합니다.

다음 단계

적절한 예제 테스트에 대한 링크와 함께 더 많은 샘플 코드가 여기에 나와야 합니다.

참여

버그가 발생하거나 제안이 있는 경우 프로젝트의 문제 섹션에 문제를 제출하세요.