232. (python/파이썬) [AWS] [Lambda] 런타임 Python 3.13 - boto3 모듈 사용해 listSignalingChannels 신호 채널 목록 조회

---

[개발 환경 설정]

​

개발 툴 : Aws / Lambda / Runtime Python 3.13

​

개발 언어 : python

​

 

---

[소스 코드]

​

// --------------------------------------------------------------------------------------
[개발 및 테스트 환경]
// --------------------------------------------------------------------------------------

- 언어 : Python


- 개발 툴 : Aws / Lambda / Runtime Python 3.13


- 개발 기술 : AWS Lambda 이벤트 동작 함수


- 사전) 👉 AWS Lambda 설명 : 

  >> Aws Lambda 는 서버 리스 FaaS 솔루션으로, 함수의 인스턴스를 실행하여 이벤트를 처리할 수 있습니다.

  >> Aws Lambda 는 이벤트에 응답하여 코드를 실행 하고 해당 코드에 필요한 컴퓨팅 리소스를 자동으로 관리합니다.


- 사전) 👉 AWS WebRTC 간략 설명 : 

  >> WebRTC 란 웹, 애플리케이션, 디바이스 간 중간자 없이 오디오나 영상 미디어를 포착하고 실시간 스트림할 뿐 아니라, 임의의 데이터도 교환할 수 있도록 하는 기술입니다

  >> WebRTC 는 간단한 API 를 통해 웹 브라우저, 모바일 애플리케이션 및 커넥티드 디바이스 간에 실시간 통신을 활성화할 수 있습니다

  >> WebRTC 주요 용어 : 

    - SDP (Session Description Protocol) : 오디오/비디오 코덱, 해상도, 포트 등 스트리밍 정보를 담은 텍스트 포맷
    - Offer / Answer : 통신 연결을 협상하기 위한 SDP 메시지 (초기 연결 설정)
    - ICE (Interactive Connectivity Establishment) : NAT/P2P 환경에서도 연결 가능한 경로(IP, 포트 등)를 찾기 위한 기술
    - Candidate : 가능한 연결 경로 (IP + Port 조합)

  >> WebRTC SDP 오퍼 생성 (뷰어) 및 응답 (마스터) 스트리밍 플로우 : 

    [Viewer → Signaling Server] -- SDP Offer --> [Master] : 뷰어는 마스터로 스트리밍 오퍼 신호 보낸다
    [Master] -- SDP Answer --> [Viewer] : 마스터는 특정 뷰어의 오퍼 신호 확인 후 응답을 보낸다

    [Viewer] -- ICE Candidate --> [Master] : 스트리밍을 할 수 있는 경로 확인
    [Master] -- ICE Candidate --> [Viewer] : 스트리밍을 할 수 있는 경로 확인

    P2P 연결 성립 → 스트리밍 시작

// --------------------------------------------------------------------------------------






// --------------------------------------------------------------------------------------
[소스 코드]
// --------------------------------------------------------------------------------------

# ========================================================================
# [Aws] : [boto3 모듈] : IAM 계정 정보 사용해 WebRTC 신호 채널 목록 조회 - listSignalingChannels
# ========================================================================
# [참고] : API Gateway 와 연동 되어 Post 방식으로 Lambda 함수 호출 및 응답 값 반환 수행
# ========================================================================
"""
1. aws lambda python 3.13 런타임 환경 기반
2. boto3 모듈 기본 내장 AWS 사용 모듈
3. import ClientError : AWS SDK for Python 인 boto3 에서 발생할 수 있는 예외 중 하나로, AWS 서비스 호출 중 오류가 발생했을 때 사용됩니다
"""
# ========================================================================

import json
import os

import boto3
from datetime import datetime
from botocore.exceptions import ClientError
from decimal import Decimal

def lambda_handler(event, context): # Lambda 호출 시 동작 되는 메인 함수

    # [event , context 정보 디버깅 로그 출력]
    print(f"DLOG = event : {event} / context {context}")


    # [Return 반환 Json 변수 선언]
    response = {
        "statusCode" : 0,
        "headers" : {},
        "body" : ""
    }


    # [AWS IAM 계정 AccessKey, SecretKey 변수 선언]
    iamAccessKey = "AK..A6"
    iamSecretKey = "mP..5J"
    iamRegion = "ap-northeast-2"  # 서울 리전


    # [명시적 인증 정보로 세션 생성]
    session = boto3.Session(
        aws_access_key_id=iamAccessKey,
        aws_secret_access_key=iamSecretKey,
        region_name = iamRegion
    )


    # [AWS 클라이언트 생성]
    aws_client = session.client('kinesisvideo')

    
    try:

      # ---------------------------------------------
      # ✅ [주요 에러 정리]
      # ---------------------------------------------
      # AccessDeniedException : IAM 권한 부족
      # ---------------------------------------------
      # ClientLimitExceededException : Kinesis Video Streams의 API 호출 속도가 너무 빠름 / 지정된 API rate limit을 초과함
      # ---------------------------------------------
      # InvalidArgumentException : MaxResults가 허용 범위(1~10000)를 벗어남 / NextToken 형식 오류
      # ---------------------------------------------
      # ResourceNotFoundException : 특정 조건(ChannelNameCondition 등)이 맞지 않아 대상 없음
      # ---------------------------------------------
      # InternalServerError : AWS 서비스 내부 오류
      # ---------------------------------------------
      # ServiceUnavailable : AWS 장애 또는 일시적 중단
      # ---------------------------------------------
            
      channels = []
      next_token = None

      
      while True:
        # 요청 파라미터 구성
        params = {
            "MaxResults": 10
        }

        if next_token:
          params["NextToken"] = next_token


        # API 호출
        response = aws_client.list_signaling_channels(**params)

        
        # ✅ datetime → str 변환 : Object of type datetime is not JSON serializable
        for ch in response.get("ChannelInfoList", []):
          if "CreationTime" in ch and isinstance(ch["CreationTime"], datetime):
            ch["CreationTime"] = ch["CreationTime"].isoformat()


        # 결과 확인 및 배열에 추가
        channels.extend(response.get("ChannelInfoList", []))

        # nextToken 확인
        next_token = response.get("NextToken")
        if not next_token:
          break  # 더 이상 페이지 없음 → 종료


      # [리턴 변수 삽입] : ApiGateWay 응답 반환 설정 : Lambda 통합 요청 사용
      response["statusCode"] = 200 
      response["headers"] = {
        "Content-Type": "application/json",
        'Access-Control-Allow-Origin': '*',  # CORS 허용 (필요 시)
      }
      response["body"] = json.dumps({        
        "length": len(channels), # 배열 길이
        "channels": channels # 배열 데이터
      })
        
    except ClientError as e: # AWS 서비스 호출 중 오류 발생 처리
        error_code = e.response['Error']['Code']
        error_message = e.response['Error']['Message']

        # [리턴 변수 삽입]
        response["statusCode"] = 400
        response["headers"] = {
          "Content-Type": "application/json",
          'Access-Control-Allow-Origin': '*',  # CORS 허용 (필요 시)
        }
        response["body"] = json.dumps(
          {
            "exception" : "ClientError",
            "error_code" : error_code,
            "error_message" : error_message
          }
        )


    # [리턴 반환 수행]
    return response

# ========================================================================
"""
{ 
  "statusCode": 200,
  "headers": {
    "Content-Type": "application/json",
    "Access-Control-Allow-Origin": "*"
  },
  "body": "{\"length\": 1, \"channels\": [{\"ChannelName\": \"56565099\", \"ChannelARN\": \"arn:aws:kinesisvideo:ap-northeast-2:123456789012:channel/56565099/1753692294870\", \"ChannelType\": \"SINGLE_MASTER\", \"ChannelStatus\": \"ACTIVE\", \"CreationTime\": \"2025-07-28T08:44:54.870000+00:00\", \"SingleMasterConfiguration\": {\"MessageTtlSeconds\": 60}, \"Version\": \"49pTiIr\"}]}"
}
"""
# ========================================================================

// --------------------------------------------------------------------------------------






// --------------------------------------------------------------------------------------
[참고 사이트]
// --------------------------------------------------------------------------------------

[자바스크립트 AWS Kvs WebRtc 생성 된 신호 채널 ARN 리스트 목록 확인 - listSignalingChannels]

https://kkh0977.tistory.com/8063

https://blog.naver.com/kkh0977/223901315493?trackingCode=blog_bloghome_searchlist


[자바스크립트 AWS Kvs WebRtc 신호 채널 생성 수행 - createSignalingChannel]

https://kkh0977.tistory.com/8091

https://blog.naver.com/kkh0977/223913968669?trackingCode=blog_bloghome_searchlist


[자바스크립트 AWS Kvs WebRtc 신호 채널 삭제 수행 - deleteSignalingChannel]

https://kkh0977.tistory.com/8093

https://blog.naver.com/kkh0977/223914446536?trackingCode=blog_bloghome_searchlist


[Amazon API Gateway] Aws API Gateway 게이트웨이 설명 정리 - 중개 서버

https://blog.naver.com/kkh0977/223827753479


[Aws Lambda] Aws 사이트에서 생성 된 Lambda 람다 검증 함수 리스트 및 내용 소스 코드 확인 방법

https://blog.naver.com/kkh0977/223765198383


[AWS] Lambda 람다 함수 수행 errorType Sandbox.Timedout 에러 발생

https://blog.naver.com/kkh0977/223962778768

// --------------------------------------------------------------------------------------

 

---

[참고 사이트]

​

https://kkh0977.tistory.com/8063

396. (javaScript) 자바스크립트 AWS Kvs WebRtc 생성 된 신호 채널 ARN 리스트 목록 확인 - listSignalingChannels

---