HTTP v1 API를 사용하여 db연동 푸시 알림을 전송하는 샘플 코드

pip install firebase-admin mysql-connector-python

import mysql.connector
import firebase_admin
from firebase_admin import credentials
from firebase_admin import messaging

# Firebase Admin SDK 초기화
cred = credentials.Certificate('./path/to/your/serviceAccountKey.json')
firebase_admin.initialize_app(cred)

def get_device_tokens():
    # MySQL 데이터베이스 연결
    connection = mysql.connector.connect(
        host='YOUR_DB_HOST',
        user='YOUR_DB_USER',
        password='YOUR_DB_PASSWORD',
        database='YOUR_DATABASE_NAME'
    )
    
    cursor = connection.cursor()
    
    # 토큰을 가져오는 쿼리
    query = "SELECT token FROM device_tokens"  # device_tokens 테이블에서 token을 선택
    cursor.execute(query)

    tokens = [row[0] for row in cursor.fetchall()]  # 결과를 리스트로 변환

    cursor.close()
    connection.close()

    return tokens

def send_fcm_messages(device_tokens, title, body, image_url, company_name):
    # 푸시 알림 메시지 생성
    message = messaging.MulticastMessage(
        notification=messaging.Notification(
            title=title,
            body=body
        ),
        data={
            'image': image_url,  # 이미지 URL 
            'click_action': 'FLUTTER_NOTIFICATION_CLICK',  # 클릭 시 동작 정의
            'companyName': company_name  # 회사명 추가
        },
        tokens=device_tokens,  # 여러 개의 토큰을 리스트로 전달
    )

    # 메시지 전송
    try:
        response = messaging.send_multicast(message)
        print('Successfully sent messages:', response)
    except Exception as e:
        print('Error sending messages:', e)

# 예제 사용
if __name__ == "__main__":
    device_tokens = get_device_tokens()  # DB에서 토큰 리스트 가져오기
    image_url = 'https://example.com/path/to/your/image.jpg'  # 실제 이미지 URL로 대체
    company_name = '회사명'  # 실제 회사명으로 대체
    
    if device_tokens:  # 토큰이 있는 경우에만 메시지 전송
        send_fcm_messages(device_tokens, "안녕하세요", "FCM HTTP v1 푸시 알림 테스트입니다.", image_url, company_name)
    else:
        print("등록된 장치 토큰이 없습니다.")