無伺服器架構與訊息代理：大學應用系統最佳化完整實踐

# AWS Lambda 學生註冊處理函式
# 處理學生註冊請求並發送確認通知

import json
import boto3
import uuid
from datetime import datetime
from typing import Dict, Any, Optional
from botocore.exceptions import ClientError

# 初始化 AWS 服務客戶端
# Lambda 執行環境會自動使用 IAM 角色的認證
dynamodb = boto3.resource('dynamodb')
sqs = boto3.client('sqs')
sns = boto3.client('sns')

# 資料表和佇列配置
STUDENT_TABLE = 'university_students'
NOTIFICATION_QUEUE_URL = 'https://sqs.ap-northeast-1.amazonaws.com/123456789/notifications'

def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    """
    Lambda 主處理函式

    處理來自 API Gateway 的學生註冊請求，
    將資料儲存到 DynamoDB 並發送通知訊息到 SQS

    Args:
        event: API Gateway 傳入的事件物件
        context: Lambda 執行環境上下文

    Returns:
        包含狀態碼和回應內容的字典
    """
    try:
        # 解析請求內容
        # API Gateway 會將請求主體包裝在 body 欄位中
        if 'body' not in event:
            return create_response(400, {'error': '缺少請求內容'})

        # 處理 body 可能是字串或已解析的字典
        body = event['body']
        if isinstance(body, str):
            body = json.loads(body)

        # 驗證必要欄位
        validation_result = validate_registration_data(body)
        if validation_result is not None:
            return create_response(400, {'error': validation_result})

        # 建立學生記錄
        student_record = create_student_record(body)

        # 儲存到 DynamoDB
        save_result = save_student_to_database(student_record)
        if not save_result['success']:
            return create_response(500, {'error': save_result['message']})

        # 發送註冊確認通知到 SQS
        notification_result = send_registration_notification(student_record)
        if not notification_result['success']:
            # 記錄通知失敗但不影響註冊成功
            print(f"通知發送失敗: {notification_result['message']}")

        # 回傳成功回應
        return create_response(200, {
            'message': '註冊成功',
            'student_id': student_record['student_id'],
            'registration_time': student_record['registration_time']
        })

    except json.JSONDecodeError:
        return create_response(400, {'error': '無效的 JSON 格式'})
    except Exception as e:
        # 記錄錯誤以便追蹤
        print(f"處理錯誤: {str(e)}")
        return create_response(500, {'error': '內部伺服器錯誤'})

def validate_registration_data(data: Dict[str, Any]) -> Optional[str]:
    """
    驗證註冊資料

    檢查必要欄位是否存在且格式正確

    Args:
        data: 註冊資料字典

    Returns:
        如果驗證失敗則回傳錯誤訊息，否則回傳 None
    """
    required_fields = ['name', 'email', 'department', 'program']

    # 檢查必要欄位
    for field in required_fields:
        if field not in data or not data[field]:
            return f'缺少必要欄位: {field}'

    # 驗證 email 格式
    email = data['email']
    if '@' not in email or '.' not in email:
        return '無效的 email 格式'

    # 驗證 email 是否為學校網域
    if not email.endswith('.edu.tw'):
        return 'email 必須使用學校網域 (.edu.tw)'

    return None

def create_student_record(data: Dict[str, Any]) -> Dict[str, Any]:
    """
    建立學生記錄

    根據輸入資料建立完整的學生記錄，
    包含自動生成的 ID 和時間戳記

    Args:
        data: 原始註冊資料

    Returns:
        完整的學生記錄字典
    """
    # 生成唯一的學生 ID
    # 使用 UUID 確保全域唯一性
    student_id = f"STU-{datetime.now().strftime('%Y%m')}-{uuid.uuid4().hex[:8].upper()}"

    return {
        'student_id': student_id,
        'name': data['name'],
        'email': data['email'],
        'department': data['department'],
        'program': data['program'],
        'phone': data.get('phone', ''),
        'address': data.get('address', ''),
        'registration_time': datetime.utcnow().isoformat(),
        'status': 'pending_verification',
        'academic_year': datetime.now().year
    }

def save_student_to_database(student: Dict[str, Any]) -> Dict[str, Any]:
    """
    儲存學生記錄到 DynamoDB

    Args:
        student: 學生記錄字典

    Returns:
        包含操作結果的字典
    """
    try:
        table = dynamodb.Table(STUDENT_TABLE)

        # 使用條件表達式確保 email 唯一性
        # 如果 email 已存在則操作失敗
        table.put_item(
            Item=student,
            ConditionExpression='attribute_not_exists(email)'
        )

        return {'success': True, 'message': '儲存成功'}

    except ClientError as e:
        error_code = e.response['Error']['Code']
        if error_code == 'ConditionalCheckFailedException':
            return {'success': False, 'message': '此 email 已被註冊'}
        else:
            return {'success': False, 'message': str(e)}

def send_registration_notification(student: Dict[str, Any]) -> Dict[str, Any]:
    """
    發送註冊確認通知到 SQS

    將通知訊息放入佇列，由專門的通知處理服務處理

    Args:
        student: 學生記錄字典

    Returns:
        包含操作結果的字典
    """
    try:
        # 建立通知訊息
        message = {
            'type': 'registration_confirmation',
            'recipient': student['email'],
            'student_id': student['student_id'],
            'student_name': student['name'],
            'template': 'welcome_email',
            'timestamp': datetime.utcnow().isoformat()
        }

        # 發送訊息到 SQS
        response = sqs.send_message(
            QueueUrl=NOTIFICATION_QUEUE_URL,
            MessageBody=json.dumps(message),
            MessageGroupId='registration',
            MessageDeduplicationId=student['student_id']
        )

        return {
            'success': True,
            'message_id': response['MessageId']
        }

    except Exception as e:
        return {'success': False, 'message': str(e)}

def create_response(status_code: int, body: Dict[str, Any]) -> Dict[str, Any]:
    """
    建立 API Gateway 回應格式

    Args:
        status_code: HTTP 狀態碼
        body: 回應內容

    Returns:
        API Gateway 格式的回應字典
    """
    return {
        'statusCode': status_code,
        'headers': {
            'Content-Type': 'application/json',
            'Access-Control-Allow-Origin': '*',
            'Access-Control-Allow-Headers': 'Content-Type',
            'Access-Control-Allow-Methods': 'POST, OPTIONS'
        },
        'body': json.dumps(body, ensure_ascii=False)
    }

// course_service.proto
// 課程管理服務定義

syntax = "proto3";

package university.course;

// 課程管理服務
service CourseService {
    // 取得課程詳細資訊
    rpc GetCourse(GetCourseRequest) returns (CourseResponse) {}

    // 搜尋課程
    rpc SearchCourses(SearchCoursesRequest) returns (SearchCoursesResponse) {}

    // 取得課程列表（伺服器串流）
    rpc ListCourses(ListCoursesRequest) returns (stream CourseResponse) {}
}

// 取得課程請求
message GetCourseRequest {
    string course_id = 1;
    bool include_schedule = 2;
    bool include_instructor = 3;
}

// 搜尋課程請求
message SearchCoursesRequest {
    string keyword = 1;
    string department = 2;
    int32 credits = 3;
    int32 page = 4;
    int32 page_size = 5;
}

// 搜尋課程回應
message SearchCoursesResponse {
    repeated CourseResponse courses = 1;
    int32 total_count = 2;
    int32 page = 3;
}

// 列出課程請求
message ListCoursesRequest {
    string department = 1;
    string semester = 2;
}

// 課程回應
message CourseResponse {
    string course_id = 1;
    string title = 2;
    string description = 3;
    int32 credits = 4;
    string department = 5;
    int32 capacity = 6;
    int32 enrolled = 7;
    repeated Schedule schedules = 8;
    Instructor instructor = 9;
}

// 課程時間表
message Schedule {
    string day_of_week = 1;
    string start_time = 2;
    string end_time = 3;
    string location = 4;
}

// 授課教師
message Instructor {
    string id = 1;
    string name = 2;
    string email = 3;
    string office = 4;
}

# gRPC 課程管理服務實作
# 提供高效能的課程資訊查詢服務

import grpc
from concurrent import futures
from typing import Iterator
import time

# 由 protoc 生成的模組
import course_service_pb2
import course_service_pb2_grpc

class CourseServicer(course_service_pb2_grpc.CourseServiceServicer):
    """
    課程管理服務實作

    實作 Protocol Buffers 定義的所有 RPC 方法
    """

    def __init__(self):
        """初始化服務，設定模擬資料"""
        # 模擬課程資料庫
        self.courses = {
            'CS101': {
                'course_id': 'CS101',
                'title': '程式設計導論',
                'description': '學習程式設計的基本概念，包含變數、控制流程、函式和資料結構。',
                'credits': 3,
                'department': '資訊工程學系',
                'capacity': 60,
                'enrolled': 45,
                'schedules': [
                    {'day_of_week': '一', 'start_time': '09:00', 'end_time': '12:00', 'location': '工程大樓 101'}
                ],
                'instructor': {
                    'id': 'T001',
                    'name': '王教授',
                    'email': 'wang@university.edu.tw',
                    'office': '資工系館 301'
                }
            },
            'CS201': {
                'course_id': 'CS201',
                'title': '資料結構',
                'description': '深入學習各種資料結構的實作與應用。',
                'credits': 3,
                'department': '資訊工程學系',
                'capacity': 50,
                'enrolled': 48,
                'schedules': [
                    {'day_of_week': '三', 'start_time': '14:00', 'end_time': '17:00', 'location': '工程大樓 201'}
                ],
                'instructor': {
                    'id': 'T002',
                    'name': '李教授',
                    'email': 'lee@university.edu.tw',
                    'office': '資工系館 302'
                }
            }
        }

    def GetCourse(
        self,
        request: course_service_pb2.GetCourseRequest,
        context: grpc.ServicerContext
    ) -> course_service_pb2.CourseResponse:
        """
        取得單一課程詳細資訊

        Args:
            request: 包含課程 ID 的請求
            context: gRPC 服務上下文

        Returns:
            課程詳細資訊
        """
        course_id = request.course_id

        # 檢查課程是否存在
        if course_id not in self.courses:
            context.set_code(grpc.StatusCode.NOT_FOUND)
            context.set_details(f'找不到課程: {course_id}')
            return course_service_pb2.CourseResponse()

        # 取得課程資料
        course_data = self.courses[course_id]

        # 建立回應
        response = self._create_course_response(
            course_data,
            include_schedule=request.include_schedule,
            include_instructor=request.include_instructor
        )

        return response

    def SearchCourses(
        self,
        request: course_service_pb2.SearchCoursesRequest,
        context: grpc.ServicerContext
    ) -> course_service_pb2.SearchCoursesResponse:
        """
        搜尋課程

        支援關鍵字、學系、學分等條件搜尋

        Args:
            request: 搜尋條件
            context: gRPC 服務上下文

        Returns:
            符合條件的課程列表
        """
        # 過濾課程
        results = []
        for course_data in self.courses.values():
            # 關鍵字搜尋
            if request.keyword:
                keyword = request.keyword.lower()
                if (keyword not in course_data['title'].lower() and
                    keyword not in course_data['description'].lower()):
                    continue

            # 學系過濾
            if request.department and course_data['department'] != request.department:
                continue

            results.append(course_data)

        # 分頁處理
        page = request.page if request.page > 0 else 1
        page_size = request.page_size if request.page_size > 0 else 10
        start_idx = (page - 1) * page_size
        end_idx = start_idx + page_size

        paginated_results = results[start_idx:end_idx]

        # 建立回應
        response = course_service_pb2.SearchCoursesResponse(
            total_count=len(results),
            page=page
        )

        for course_data in paginated_results:
            course_response = self._create_course_response(
                course_data,
                include_schedule=True,
                include_instructor=True
            )
            response.courses.append(course_response)

        return response

    def ListCourses(
        self,
        request: course_service_pb2.ListCoursesRequest,
        context: grpc.ServicerContext
    ) -> Iterator[course_service_pb2.CourseResponse]:
        """
        串流列出課程

        使用伺服器串流逐一回傳課程資料

        Args:
            request: 列出條件
            context: gRPC 服務上下文

        Yields:
            課程回應物件
        """
        for course_data in self.courses.values():
            # 學系過濾
            if request.department and course_data['department'] != request.department:
                continue

            # 建立並回傳課程回應
            response = self._create_course_response(
                course_data,
                include_schedule=True,
                include_instructor=True
            )

            yield response
            time.sleep(0.1)  # 模擬處理時間

    def _create_course_response(
        self,
        course_data: dict,
        include_schedule: bool = False,
        include_instructor: bool = False
    ) -> course_service_pb2.CourseResponse:
        """建立課程回應物件"""
        response = course_service_pb2.CourseResponse(
            course_id=course_data['course_id'],
            title=course_data['title'],
            description=course_data['description'],
            credits=course_data['credits'],
            department=course_data['department'],
            capacity=course_data['capacity'],
            enrolled=course_data['enrolled']
        )

        # 加入時間表
        if include_schedule and 'schedules' in course_data:
            for schedule_data in course_data['schedules']:
                schedule = course_service_pb2.Schedule(
                    day_of_week=schedule_data['day_of_week'],
                    start_time=schedule_data['start_time'],
                    end_time=schedule_data['end_time'],
                    location=schedule_data['location']
                )
                response.schedules.append(schedule)

        # 加入教師資訊
        if include_instructor and 'instructor' in course_data:
            instructor_data = course_data['instructor']
            response.instructor.CopyFrom(course_service_pb2.Instructor(
                id=instructor_data['id'],
                name=instructor_data['name'],
                email=instructor_data['email'],
                office=instructor_data['office']
            ))

        return response

def serve():
    """啟動 gRPC 伺服器"""
    # 建立 gRPC 伺服器
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))

    # 註冊服務
    course_service_pb2_grpc.add_CourseServiceServicer_to_server(
        CourseServicer(),
        server
    )

    # 綁定連接埠
    server.add_insecure_port('[::]:50051')

    # 啟動伺服器
    server.start()
    print('課程管理 gRPC 伺服器已啟動，監聽連接埠 50051')

    try:
        server.wait_for_termination()
    except KeyboardInterrupt:
        server.stop(0)

if __name__ == '__main__':
    serve()

# Redis 訊息系統
# 實作發布/訂閱和任務佇列功能

import redis
import json
import time
import threading
from typing import Callable, Dict, Any, Optional
from datetime import datetime
import uuid

class RedisMessageBroker:
    """
    Redis 訊息代理

    提供發布/訂閱和佇列功能
    """

    def __init__(self, host: str = 'localhost', port: int = 6379, db: int = 0):
        """
        初始化訊息代理

        Args:
            host: Redis 主機
            port: Redis 連接埠
            db: Redis 資料庫編號
        """
        # 建立 Redis 連線池
        self.pool = redis.ConnectionPool(
            host=host,
            port=port,
            db=db,
            decode_responses=True
        )

        # 主連線
        self.redis = redis.Redis(connection_pool=self.pool)

        # 訂閱者執行緒
        self.subscriber_threads = []
        self.running = True

    def publish(self, channel: str, message_type: str, payload: Dict[str, Any]) -> str:
        """
        發布訊息到頻道

        Args:
            channel: 頻道名稱
            message_type: 訊息類型
            payload: 訊息內容

        Returns:
            訊息 ID
        """
        # 建立訊息
        message_id = str(uuid.uuid4())
        message = {
            'id': message_id,
            'type': message_type,
            'payload': payload,
            'timestamp': datetime.utcnow().isoformat()
        }

        # 發布到頻道
        self.redis.publish(channel, json.dumps(message))

        return message_id

    def subscribe(
        self,
        channels: list,
        handler: Callable[[str, Dict], None]
    ) -> None:
        """
        訂閱頻道

        在背景執行緒中監聽訊息

        Args:
            channels: 要訂閱的頻道列表
            handler: 訊息處理函式
        """
        def subscriber_worker():
            # 為訂閱者建立獨立連線
            pubsub = self.redis.pubsub()
            pubsub.subscribe(*channels)

            for message in pubsub.listen():
                if not self.running:
                    break

                # 只處理實際訊息
                if message['type'] == 'message':
                    try:
                        data = json.loads(message['data'])
                        handler(message['channel'], data)
                    except Exception as e:
                        print(f"處理訊息錯誤: {e}")

        # 啟動訂閱者執行緒
        thread = threading.Thread(target=subscriber_worker, daemon=True)
        thread.start()
        self.subscriber_threads.append(thread)

    def enqueue(self, queue_name: str, task_type: str, task_data: Dict[str, Any]) -> str:
        """
        將任務加入佇列

        Args:
            queue_name: 佇列名稱
            task_type: 任務類型
            task_data: 任務資料

        Returns:
            任務 ID
        """
        # 建立任務
        task_id = str(uuid.uuid4())
        task = {
            'id': task_id,
            'type': task_type,
            'data': task_data,
            'timestamp': datetime.utcnow().isoformat(),
            'retry_count': 0
        }

        # 加入佇列
        self.redis.rpush(queue_name, json.dumps(task))

        return task_id

    def dequeue(self, queue_name: str, timeout: int = 0) -> Optional[Dict]:
        """
        從佇列取出任務

        Args:
            queue_name: 佇列名稱
            timeout: 等待超時秒數

        Returns:
            任務字典或 None
        """
        if timeout > 0:
            # 阻塞式等待
            result = self.redis.blpop(queue_name, timeout)
            if result:
                return json.loads(result[1])
        else:
            # 非阻塞式
            result = self.redis.lpop(queue_name)
            if result:
                return json.loads(result)

        return None

    def get_queue_length(self, queue_name: str) -> int:
        """取得佇列長度"""
        return self.redis.llen(queue_name)

    def close(self):
        """關閉訊息代理"""
        self.running = False
        for thread in self.subscriber_threads:
            thread.join(timeout=1)

class TaskWorker:
    """
    任務工作者

    從佇列中取出任務並處理
    """

    def __init__(
        self,
        broker: RedisMessageBroker,
        queue_name: str,
        max_retries: int = 3
    ):
        """
        初始化工作者

        Args:
            broker: 訊息代理實例
            queue_name: 監聽的佇列名稱
            max_retries: 最大重試次數
        """
        self.broker = broker
        self.queue_name = queue_name
        self.max_retries = max_retries
        self.handlers = {}
        self.running = False

    def register_handler(self, task_type: str, handler: Callable[[Dict], bool]):
        """
        註冊任務處理器

        Args:
            task_type: 任務類型
            handler: 處理函式，回傳 True 表示成功
        """
        self.handlers[task_type] = handler

    def start(self):
        """啟動工作者"""
        self.running = True
        print(f"任務工作者已啟動，監聽佇列: {self.queue_name}")

        while self.running:
            # 從佇列取出任務
            task = self.broker.dequeue(self.queue_name, timeout=5)

            if task is None:
                continue

            # 處理任務
            task_type = task.get('type')
            task_id = task.get('id')

            print(f"處理任務 {task_id} (類型: {task_type})")

            if task_type not in self.handlers:
                print(f"未知的任務類型: {task_type}")
                continue

            try:
                # 呼叫處理器
                success = self.handlers[task_type](task['data'])

                if success:
                    print(f"任務 {task_id} 處理成功")
                else:
                    self._handle_failure(task)

            except Exception as e:
                print(f"任務 {task_id} 處理錯誤: {e}")
                self._handle_failure(task)

    def _handle_failure(self, task: Dict):
        """處理失敗的任務"""
        retry_count = task.get('retry_count', 0)

        if retry_count < self.max_retries:
            # 重新加入佇列
            task['retry_count'] = retry_count + 1
            self.broker.redis.rpush(
                self.queue_name,
                json.dumps(task)
            )
            print(f"任務 {task['id']} 將重試 (第 {task['retry_count']} 次)")
        else:
            # 移至死信佇列
            self.broker.redis.rpush(
                f"{self.queue_name}:dead",
                json.dumps(task)
            )
            print(f"任務 {task['id']} 已移至死信佇列")

    def stop(self):
        """停止工作者"""
        self.running = False

# 使用範例
if __name__ == '__main__':
    # 建立訊息代理
    broker = RedisMessageBroker()

    # 發布/訂閱範例
    def notification_handler(channel: str, message: Dict):
        print(f"收到通知 [{channel}]: {message['payload']}")

    # 訂閱通知頻道
    broker.subscribe(['announcements'], notification_handler)

    # 發布公告
    broker.publish(
        'announcements',
        'system_announcement',
        {
            'title': '系統維護通知',
            'content': '系統將於週六凌晨進行例行維護'
        }
    )

    # 任務佇列範例
    worker = TaskWorker(broker, 'email_tasks')

    def send_email_handler(data: Dict) -> bool:
        print(f"發送郵件至 {data.get('recipient')}")
        return True

    worker.register_handler('send_email', send_email_handler)

    # 加入郵件任務
    task_id = broker.enqueue(
        'email_tasks',
        'send_email',
        {
            'recipient': 'student@university.edu.tw',
            'subject': '選課結果通知'
        }
    )
    print(f"已加入任務: {task_id}")

    # 在背景啟動工作者
    import threading
    worker_thread = threading.Thread(target=worker.start, daemon=True)
    worker_thread.start()

    time.sleep(3)
    worker.stop()
    broker.close()