API

API

Bases: BaseRouterMixin

モダンな Lambda 用 API フレームワーク

Source code in lambapi/core.py

class API(BaseRouterMixin):
    """モダンな Lambda 用 API フレームワーク"""

    def __init__(self, event: Dict[str, Any], context: Any, root_path: str = ""):
        self.event = event
        self.context = context
        self.root_path = self._validate_root_path(root_path)
        self.routes: List[Route] = []
        # 高速ルート検索のための最適化構造
        self._exact_routes: Dict[str, Dict[str, Route]] = {}  # method -> {path -> route}
        self._pattern_routes: Dict[str, List[Route]] = {}  # method -> [routes with params]
        self._middleware: List[Callable] = []
        self._cors_config: Optional[CORSConfig] = None
        self._error_registry = get_global_registry()

    def _validate_root_path(self, root_path: str) -> str:
        """root_path をバリデーションして正規化"""
        if not root_path:
            return ""

        # 先頭にスラッシュがない場合は追加
        if not root_path.startswith("/"):
            root_path = f"/{root_path}"

        # 末尾スラッシュを除去
        root_path = root_path.rstrip("/")

        # 重複スラッシュを正規化
        root_path = re.sub(r"/+", "/", root_path)

        return root_path

    def _normalize_path(self, path: str) -> str:
        """root_path を考慮してパスを正規化"""
        if not self.root_path:
            return path

        # 完全一致または / で区切られた場合のみ除去
        if path == self.root_path:
            return "/"
        elif path.startswith(f"{self.root_path}/"):
            return path[len(self.root_path) :]
        else:
            return path

    def add_middleware(self, middleware: Callable) -> None:
        """ミドルウェアを追加"""
        self._middleware.append(middleware)

    def error_handler(self, exception_type: Type[Exception]) -> Callable:
        """エラーハンドラーデコレータ"""

        def decorator(handler_func: Callable) -> Callable:
            self._error_registry.register(exception_type, handler_func)
            return handler_func

        return decorator

    def default_error_handler(self, handler_func: Callable) -> Callable:
        """デフォルトエラーハンドラーデコレータ"""
        self._error_registry.set_default_handler(handler_func)
        return handler_func

    def _update_route_index(self, route: Route) -> None:
        """ルートを高速検索用インデックスに追加"""
        method = route.method

        # メソッド別辞書の初期化
        if method not in self._exact_routes:
            self._exact_routes[method] = {}
        if method not in self._pattern_routes:
            self._pattern_routes[method] = []

        # パスパラメータがない場合は完全一致テーブルに追加
        if "{" not in route.path:
            self._exact_routes[method][route.path] = route
        else:
            # パスパラメータがある場合はパターンマッチング用リストに追加
            self._pattern_routes[method].append(route)

    def _rebuild_route_index(self) -> None:
        """ルートインデックスを再構築（include_router 時に使用）"""
        self._exact_routes.clear()
        self._pattern_routes.clear()

        for route in self.routes:
            self._update_route_index(route)

    def enable_cors(
        self,
        origins: Union[str, List[str]] = "*",
        methods: Optional[List[str]] = None,
        headers: Optional[List[str]] = None,
        allow_credentials: bool = False,
        max_age: Optional[int] = None,
        expose_headers: Optional[List[str]] = None,
    ) -> None:
        """CORS を有効にする

        Args:
            origins: 許可するオリジン（'*' または具体的な URL）
            methods: 許可する HTTP メソッド
            headers: 許可するヘッダー
            allow_credentials: 認証情報の送信を許可するか
            max_age: プリフライトリクエストのキャッシュ時間（秒）
            expose_headers: ブラウザに公開するレスポンスヘッダー
        """
        self._cors_config = create_cors_config(
            origins=origins,
            methods=methods,
            headers=headers,
            allow_credentials=allow_credentials,
            max_age=max_age,
            expose_headers=expose_headers,
        )

    def include_router(
        self, router: Any, prefix: str = "", tags: Optional[List[str]] = None
    ) -> None:
        """ルーターを追加"""
        from .router import Router

        if isinstance(router, Router):
            # プレフィックスやタグが指定されている場合は新しいルーターを作成
            if prefix or tags:
                new_router = Router(prefix=prefix, tags=tags or [])
                for route in router.routes:
                    # 既存のルートを新しいプレフィックス付きでコピー
                    new_path = f"{prefix.rstrip('/')}{route.path}" if prefix else route.path
                    new_route = Route(
                        new_path,
                        route.method,
                        route.handler,
                    )
                    new_router.routes.append(new_route)
                self.routes.extend(new_router.routes)
            else:
                self.routes.extend(router.routes)

        # ルートインデックスを再構築
        self._rebuild_route_index()

    def include_auth(self, auth: Any) -> None:
        """認証機能を追加"""
        from .auth.dynamodb_auth import DynamoDBAuth
        from .auth.auth_router import create_auth_router

        if isinstance(auth, DynamoDBAuth):
            # 認証エンドポイントのルーターを作成して追加
            auth_router = create_auth_router(auth)
            self.include_router(auth_router)
        else:
            raise ValueError("auth は DynamoDBAuth のインスタンスである必要があります")

    def _add_route(
        self,
        path: str,
        method: str,
        handler: Callable,
        cors: Union[bool, CORSConfig, None] = None,
    ) -> Callable:
        """ルートを追加"""
        cors_config = None
        if cors is True:
            # デフォルトの CORS 設定を使用
            cors_config = create_cors_config()
        elif isinstance(cors, CORSConfig):
            cors_config = cors

        route = Route(path, method, handler, cors_config)
        self.routes.append(route)
        self._update_route_index(route)
        return handler

    def _find_route(
        self, path: str, method: str
    ) -> tuple[Optional[Route], Optional[Dict[str, str]]]:
        """マッチするルートを検索（最適化版）"""
        # root_path を考慮してパスを正規化
        normalized_path = self._normalize_path(path)

        # 1. 完全一致検索（O(1)）
        exact_routes = self._exact_routes.get(method, {})
        if normalized_path in exact_routes:
            return exact_routes[normalized_path], {}

        # 2. パターンマッチング検索（パラメータ付きルート）
        pattern_routes = self._pattern_routes.get(method, [])
        for route in pattern_routes:
            path_params = route.match(normalized_path, method)
            if path_params is not None:
                return route, path_params

        return None, None

    def _call_handler_with_params(
        self, route: Route, request: Request, path_params: Optional[Dict[str, str]]
    ) -> Any:
        """パスパラメータとクエリパラメータを自動注入してハンドラーを呼び出し"""
        handler = route.handler

        # signature キャッシュを使用
        if handler not in _SIGNATURE_CACHE:
            _SIGNATURE_CACHE[handler] = inspect.signature(handler)
        signature = _SIGNATURE_CACHE[handler]
        handler_params = signature.parameters

        # 最初の引数が request かどうかをチェック（従来の方式）
        param_names = list(handler_params.keys())
        if param_names and param_names[0] in ["request", "req"]:
            # 従来の方式（request を第一引数に渡す）
            return handler(request)

        # 新しい依存性注入システムを使用するかチェック
        dependencies = get_function_dependencies(handler)
        if dependencies:
            # 新しい依存性注入システムを使用
            # 認証が必要な場合は事前に認証処理を実行
            if getattr(handler, "_auth_required", False):
                # require_role デコレータのロジックを手動実行
                self._handle_authentication_for_dependency_injection(handler, request)

            return self._call_handler_with_dependencies(handler, request, path_params)
        else:
            # 従来のパラメータ注入システムを使用
            return self._call_handler_legacy_params(handler, request, path_params, signature)

    def _call_handler_with_dependencies(
        self, handler: Callable, request: Request, path_params: Optional[Dict[str, str]]
    ) -> Any:
        """新しい依存性注入システムでハンドラーを呼び出し"""
        try:
            # 認証ユーザーを取得（必要に応じて）
            authenticated_user = getattr(request, "_authenticated_user", None)

            # 依存性を解決
            resolved_params = resolve_function_dependencies(
                handler, request, path_params, authenticated_user
            )

            # 従来のパラメータも処理（互換性のため）
            legacy_params = self._get_legacy_params(handler, request, path_params)

            # 依存性注入パラメータを優先し、従来パラメータで補完
            final_params = {**legacy_params, **resolved_params}

            return handler(**final_params) if final_params else handler()

        except ValidationError:
            # バリデーションエラーはそのまま再発生させる（エラーハンドラーで処理される）
            raise
        except Exception:
            # その他のエラーが発生した場合は従来システムにフォールバック
            signature = inspect.signature(handler)
            return self._call_handler_legacy_params(handler, request, path_params, signature)

    def _call_handler_legacy_params(
        self,
        handler: Callable,
        request: Request,
        path_params: Optional[Dict[str, str]],
        signature: inspect.Signature,
    ) -> Any:
        """従来のパラメータ注入システムでハンドラーを呼び出し"""
        handler_params = signature.parameters
        param_names = list(handler_params.keys())
        call_args: Dict[str, Any] = {}

        # パスパラメータをマッチング
        if path_params:
            for param_name in param_names:
                if param_name in path_params:
                    call_args[param_name] = path_params[param_name]

        # クエリパラメータをマッチング
        query_params = request.query_params
        for param_name, param_info in handler_params.items():
            if param_name in call_args or param_name == "request":
                continue

            # クエリパラメータから値を取得
            if param_name in query_params:
                value = query_params[param_name]
                # 型変換を実行
                call_args[param_name] = self._convert_param_type(value, param_info)
            elif param_info.default != inspect.Parameter.empty:
                # デフォルト値を使用
                call_args[param_name] = param_info.default

        # request 引数がある場合は追加
        if "request" in handler_params:
            call_args["request"] = request

        # キーワード引数として渡す、もしくは引数なしで呼び出し
        return handler(**call_args) if call_args else handler()

    def _get_legacy_params(
        self, handler: Callable, request: Request, path_params: Optional[Dict[str, str]]
    ) -> Dict[str, Any]:
        """従来のパラメータ処理ロジックから基本パラメータを取得"""
        signature = inspect.signature(handler)
        handler_params = signature.parameters
        param_names = list(handler_params.keys())
        call_args: Dict[str, Any] = {}

        # パスパラメータをマッチング（依存性注入で処理されないもの）
        if path_params:
            for param_name in param_names:
                if param_name in path_params:
                    # 依存性注入対象でない場合のみ追加
                    param_info = handler_params.get(param_name)
                    if param_info and not hasattr(param_info.default, "source"):
                        call_args[param_name] = self._convert_param_type(
                            path_params[param_name], param_info
                        )

        # request 引数がある場合は追加
        if "request" in handler_params:
            call_args["request"] = request

        return call_args

    def _convert_param_type(self, value: str, param_info: inspect.Parameter) -> Any:
        """パラメータの型アノテーションに基づいて値を変換（最適化版）"""
        converter = _get_type_converter(param_info.annotation)
        return converter(value)

    def _handle_cors_preflight(self, request: Request) -> Optional[Dict[str, Any]]:
        """CORS プリフライトリクエストを処理"""
        if request.method == "OPTIONS" and self._cors_config:
            origin = request.headers.get("origin") or request.headers.get("Origin")
            cors_headers = self._cors_config.get_cors_headers(origin)
            response = Response("", status_code=200, headers=cors_headers)
            return response.to_lambda_response()
        return None

    def _handle_route_not_found(self, request: Request) -> Dict[str, Any]:
        """ルートが見つからない場合の処理"""
        response = Response({"error": "Not Found"}, status_code=404)
        response = self._apply_cors_headers(request, response, None)
        return response.to_lambda_response()

    def _process_path_params(
        self, request: Request, path_params: Optional[Dict[str, str]]
    ) -> Request:
        """パスパラメータを処理"""
        if path_params:
            if "pathParameters" not in self.event:
                self.event["pathParameters"] = {}
            self.event["pathParameters"].update(path_params)
            request = Request(self.event)  # 更新された event で Request を再作成
        return request

    def _execute_handler(
        self, route: Route, request: Request, path_params: Optional[Dict[str, str]]
    ) -> Any:
        """ハンドラーを実行"""
        try:
            return self._call_handler_with_params(route, request, path_params)
        except Exception as e:
            # カスタムエラーハンドリング
            error_response = self._error_registry.handle_error(e, request, self.context)
            error_response = self._apply_cors_headers(request, error_response, route)
            return error_response.to_lambda_response()

    def _process_response(self, result: Any, route: Route, request: Request) -> Response:
        """レスポンスを処理"""
        # レスポンスフォーマットバリデーション
        if isinstance(result, dict) and "statusCode" in result:
            return Response(result)  # エラーレスポンス

        # 結果を Response オブジェクトに変換
        if isinstance(result, Response):
            response = result
        elif isinstance(result, dict):
            response = Response(result)
        else:
            # Pydantic BaseModel の場合は辞書に変換
            if hasattr(result, "model_dump"):
                response = Response(result.model_dump())
            elif hasattr(result, "dict"):
                response = Response(result.dict())
            else:
                response = Response({"result": result})

        # ミドルウェアを適用
        response = self._apply_middleware(request, response)

        # CORS ヘッダーを追加
        response = self._apply_cors_headers(request, response, route)

        return response

    def _handle_authentication_for_dependency_injection(
        self, handler: Callable, request: Request
    ) -> None:
        """依存性注入システム用の認証処理"""
        try:
            # ハンドラーから認証情報を取得
            required_roles = getattr(handler, "_required_roles", [])

            # DynamoDBAuth インスタンスを見つける（フレームスタックから探索）
            import sys

            auth_instance = None
            frame = sys._getframe()

            while frame and auth_instance is None:
                frame_locals = frame.f_locals
                frame_globals = frame.f_globals

                # ローカル変数から auth を探索
                for var_name, var_value in frame_locals.items():
                    if hasattr(var_value, "get_authenticated_user") and hasattr(
                        var_value, "_required_roles"
                    ):
                        auth_instance = var_value
                        break

                # グローバル変数から auth を探索
                if auth_instance is None:
                    for var_name, var_value in frame_globals.items():
                        if hasattr(var_value, "get_authenticated_user") and hasattr(
                            var_value, "_required_roles"
                        ):
                            auth_instance = var_value
                            break

                frame = frame.f_back  # type: ignore

            if auth_instance is None:
                # ハンドラーのモジュールから auth をインポート
                handler_module = sys.modules.get(handler.__module__)
                if handler_module and hasattr(handler_module, "auth"):
                    auth_instance = handler_module.auth

            if auth_instance:
                # 認証ユーザーを取得
                user = auth_instance.get_authenticated_user(request)

                # ロール権限チェック
                if auth_instance.user_model._is_role_permission_enabled():
                    user_role = getattr(user, "role", None)
                    if user_role not in required_roles:
                        from .exceptions import AuthorizationError

                        raise AuthorizationError(f"必要なロール: {', '.join(required_roles)}")

                # 認証ユーザーを request に設定
                setattr(request, "_authenticated_user", user)
            else:
                # 認証インスタンスが見つからない場合はパス（エラーは後で発生する）
                pass

        except Exception:
            # 認証エラーをそのまま再発生させる
            raise

    def _handle_global_error(self, error: Exception) -> Dict[str, Any]:
        """グローバルエラーハンドリング"""
        try:
            # request の作成を試みる
            try:
                request = Request(self.event)
                error_response = self._error_registry.handle_error(error, request, self.context)
                error_response = self._apply_cors_headers(request, error_response)
            except Exception:
                # request が作成できない場合のフォールバック
                from .exceptions import InternalServerError

                internal_error = InternalServerError("Request processing failed")
                error_response = self._error_registry._handle_unknown_error(
                    internal_error, None, self.context
                )
        except Exception:
            # エラーハンドリング自体でエラーが発生した場合のフォールバック
            error_response = Response(
                {"error": "INTERNAL_ERROR", "message": "An unexpected error occurred"},
                status_code=500,
            )

        return error_response.to_lambda_response()

    def _apply_cors_headers(
        self, request: Request, response: Response, route: Optional[Route] = None
    ) -> Response:
        """CORS ヘッダーをレスポンスに追加"""
        if isinstance(response, Response):
            # 個別ルートの CORS 設定を優先
            cors_config = None
            if route and route.cors_config:
                cors_config = route.cors_config
            elif self._cors_config:
                cors_config = self._cors_config

            if cors_config:
                origin = request.headers.get("origin") or request.headers.get("Origin")
                cors_headers = cors_config.get_cors_headers(origin)
                response.headers.update(cors_headers)
        return response

    def _apply_middleware(self, request: Request, response: Any) -> Any:
        """ミドルウェアを適用"""
        for middleware in self._middleware:
            response = middleware(request, response)
        return response

    def handle_request(self) -> Dict[str, Any]:
        """メインのリクエスト処理"""
        try:
            request = Request(self.event)

            # OPTIONS リクエストの自動処理（CORS プリフライト）
            cors_response = self._handle_cors_preflight(request)
            if cors_response:
                return cors_response

            # ルート検索
            route, path_params = self._find_route(request.path, request.method)
            if not route:
                return self._handle_route_not_found(request)

            # パスパラメータを処理
            request = self._process_path_params(request, path_params)

            # ハンドラー実行
            result = self._execute_handler(route, request, path_params)
            if isinstance(result, dict) and "statusCode" in result:
                return result  # エラーレスポンスの場合

            # レスポンス処理
            response = self._process_response(result, route, request)

            return response.to_lambda_response()

        except Exception as e:
            return self._handle_global_error(e)

__init__(event, context, root_path='')

Source code in lambapi/core.py

def __init__(self, event: Dict[str, Any], context: Any, root_path: str = ""):
    self.event = event
    self.context = context
    self.root_path = self._validate_root_path(root_path)
    self.routes: List[Route] = []
    # 高速ルート検索のための最適化構造
    self._exact_routes: Dict[str, Dict[str, Route]] = {}  # method -> {path -> route}
    self._pattern_routes: Dict[str, List[Route]] = {}  # method -> [routes with params]
    self._middleware: List[Callable] = []
    self._cors_config: Optional[CORSConfig] = None
    self._error_registry = get_global_registry()

include_router(router, prefix='', tags=None)

ルーターを追加

Source code in lambapi/core.py

def include_router(
    self, router: Any, prefix: str = "", tags: Optional[List[str]] = None
) -> None:
    """ルーターを追加"""
    from .router import Router

    if isinstance(router, Router):
        # プレフィックスやタグが指定されている場合は新しいルーターを作成
        if prefix or tags:
            new_router = Router(prefix=prefix, tags=tags or [])
            for route in router.routes:
                # 既存のルートを新しいプレフィックス付きでコピー
                new_path = f"{prefix.rstrip('/')}{route.path}" if prefix else route.path
                new_route = Route(
                    new_path,
                    route.method,
                    route.handler,
                )
                new_router.routes.append(new_route)
            self.routes.extend(new_router.routes)
        else:
            self.routes.extend(router.routes)

    # ルートインデックスを再構築
    self._rebuild_route_index()

add_middleware(middleware)

ミドルウェアを追加

Source code in lambapi/core.py

def add_middleware(self, middleware: Callable) -> None:
    """ミドルウェアを追加"""
    self._middleware.append(middleware)

enable_cors(origins='*', methods=None, headers=None, allow_credentials=False, max_age=None, expose_headers=None)

CORS を有効にする

Parameters:

Name	Type	Description	Default
origins	Union[str, List[str]]	許可するオリジン（'*' または具体的な URL）	'*'
methods	Optional[List[str]]	許可する HTTP メソッド	None
headers	Optional[List[str]]	許可するヘッダー	None
allow_credentials	bool	認証情報の送信を許可するか	False
max_age	Optional[int]	プリフライトリクエストのキャッシュ時間（秒）	None
expose_headers	Optional[List[str]]	ブラウザに公開するレスポンスヘッダー	None

Source code in lambapi/core.py

def enable_cors(
    self,
    origins: Union[str, List[str]] = "*",
    methods: Optional[List[str]] = None,
    headers: Optional[List[str]] = None,
    allow_credentials: bool = False,
    max_age: Optional[int] = None,
    expose_headers: Optional[List[str]] = None,
) -> None:
    """CORS を有効にする

    Args:
        origins: 許可するオリジン（'*' または具体的な URL）
        methods: 許可する HTTP メソッド
        headers: 許可するヘッダー
        allow_credentials: 認証情報の送信を許可するか
        max_age: プリフライトリクエストのキャッシュ時間（秒）
        expose_headers: ブラウザに公開するレスポンスヘッダー
    """
    self._cors_config = create_cors_config(
        origins=origins,
        methods=methods,
        headers=headers,
        allow_credentials=allow_credentials,
        max_age=max_age,
        expose_headers=expose_headers,
    )

error_handler(exception_type)

エラーハンドラーデコレータ

Source code in lambapi/core.py

def error_handler(self, exception_type: Type[Exception]) -> Callable:
    """エラーハンドラーデコレータ"""

    def decorator(handler_func: Callable) -> Callable:
        self._error_registry.register(exception_type, handler_func)
        return handler_func

    return decorator

default_error_handler(handler_func)

デフォルトエラーハンドラーデコレータ

Source code in lambapi/core.py

def default_error_handler(self, handler_func: Callable) -> Callable:
    """デフォルトエラーハンドラーデコレータ"""
    self._error_registry.set_default_handler(handler_func)
    return handler_func

handle_request()

メインのリクエスト処理

Source code in lambapi/core.py

def handle_request(self) -> Dict[str, Any]:
    """メインのリクエスト処理"""
    try:
        request = Request(self.event)

        # OPTIONS リクエストの自動処理（CORS プリフライト）
        cors_response = self._handle_cors_preflight(request)
        if cors_response:
            return cors_response

        # ルート検索
        route, path_params = self._find_route(request.path, request.method)
        if not route:
            return self._handle_route_not_found(request)

        # パスパラメータを処理
        request = self._process_path_params(request, path_params)

        # ハンドラー実行
        result = self._execute_handler(route, request, path_params)
        if isinstance(result, dict) and "statusCode" in result:
            return result  # エラーレスポンスの場合

        # レスポンス処理
        response = self._process_response(result, route, request)

        return response.to_lambda_response()

    except Exception as e:
        return self._handle_global_error(e)

lambapi のメインクラスです。すべての API 機能はこのクラスを通じて提供されます。

基本的な使用法

from lambapi import API, create_lambda_handler

def create_app(event, context):
    app = API(event, context)

    @app.get("/")
    def hello():
        return {"message": "Hello, World!"}

    return app

lambda_handler = create_lambda_handler(create_app)

初期化

API(event, context)

Lambda から渡されるイベントとコンテキストで API インスタンスを初期化します。

パラメータ:

• event (Dict[str, Any]): Lambda イベントオブジェクト
• context (Any): Lambda コンテキストオブジェクト

例:

def lambda_handler(event, context):
    app = API(event, context)
    # ルート定義...
    return app.handle_request()

HTTP メソッドデコレータ

get(path, request_format=None, response_format=None, cors=None)

GET リクエストのエンドポイントを定義します。

パラメータ:

• path (str): エンドポイントのパス（デフォルト: "/"）
• request_format (Type, optional): リクエストのバリデーション用データクラス
• response_format (Type, optional): レスポンスのバリデーション用データクラス
• cors (Union[bool, CORSConfig, None]): CORS 設定

例:

@app.get("/")
def root():
    return {"message": "Root endpoint"}

@app.get("/users/{user_id}")
def get_user(user_id: str):
    return {"id": user_id, "name": f"User {user_id}"}

@app.get("/search")
def search(q: str = "", limit: int = 10):
    return {"query": q, "results": [], "limit": limit}

post(path, request_format=None, response_format=None, cors=None)

POST リクエストのエンドポイントを定義します。

例:

@app.post("/users")
def create_user(request):
    user_data = request.json()
    return {"message": "User created", "user": user_data}

put(path, request_format=None, response_format=None, cors=None)

PUT リクエストのエンドポイントを定義します。

delete(path, request_format=None, response_format=None, cors=None)

DELETE リクエストのエンドポイントを定義します。

patch(path, request_format=None, response_format=None, cors=None)

PATCH リクエストのエンドポイントを定義します。

ルーター統合

include_router(router, prefix="", tags=None)

Router インスタンスを API に統合します。

パラメータ:

• router (Router): 統合する Router インスタンス
• prefix (str): すべてのルートに追加するプレフィックス
• tags (List[str], optional): ルートに付与するタグ

例:

from lambapi import Router

# ユーザー関連のルーター
user_router = Router(prefix="/users")

@user_router.get("/")
def list_users():
    return {"users": []}

@user_router.get("/{id}")
def get_user(id: str):
    return {"id": id}

# メインアプリに統合
app.include_router(user_router)
# /users/ と /users/{id} が利用可能になる

ミドルウェア

add_middleware(middleware)

ミドルウェア関数を追加します。

パラメータ:

• middleware (Callable): ミドルウェア関数

ミドルウェア関数のシグネチャ:

def middleware(request: Request, response: Any) -> Any:
    # 前処理（任意）

    # 後処理
    return response

例:

def logging_middleware(request, response):
    print(f"{request.method} {request.path}")
    return response

def cors_middleware(request, response):
    if isinstance(response, Response):
        response.headers.update({
            'Access-Control-Allow-Origin': '*'
        })
    return response

app.add_middleware(logging_middleware)
app.add_middleware(cors_middleware)

CORS 設定

enable_cors(origins="*", methods=None, headers=None, allow_credentials=False, max_age=None, expose_headers=None)

グローバル CORS 設定を有効にします。

パラメータ:

• origins (Union[str, List[str]]): 許可するオリジン（デフォルト: "*"）
• methods (List[str], optional): 許可する HTTP メソッド
• headers (List[str], optional): 許可するヘッダー
• allow_credentials (bool): 認証情報の送信を許可（デフォルト: False）
• max_age (int, optional): プリフライトキャッシュ時間（秒）
• expose_headers (List[str], optional): ブラウザに公開するレスポンスヘッダー

例:

# 基本的な CORS 設定
app.enable_cors()

# 詳細な CORS 設定
app.enable_cors(
    origins=["https://example.com", "https://app.example.com"],
    methods=["GET", "POST", "PUT", "DELETE"],
    headers=["Content-Type", "Authorization"],
    allow_credentials=True,
    max_age=3600
)

エラーハンドリング

error_handler(exception_type)

特定の例外タイプに対するカスタムエラーハンドラーを登録します。

パラメータ:

• exception_type (Type[Exception]): 処理する例外タイプ

例:

class BusinessError(Exception):
    def __init__(self, message: str, code: str):
        self.message = message
        self.code = code

@app.error_handler(BusinessError)
def handle_business_error(error, request, context):
    return Response({
        "error": "BUSINESS_ERROR",
        "message": error.message,
        "code": error.code
    }, status_code=422)

default_error_handler(handler_func)

すべての未処理例外に対するデフォルトエラーハンドラーを設定します。

例:

@app.default_error_handler
def handle_unknown_error(error, request, context):
    return Response({
        "error": "INTERNAL_ERROR",
        "message": "An unexpected error occurred",
        "request_id": context.aws_request_id
    }, status_code=500)

リクエスト処理

handle_request()

メインのリクエスト処理メソッド。Lambda ハンドラーから呼び出されます。

戻り値:

Dict[str, Any]: Lambda レスポンス形式の辞書

例:

def lambda_handler(event, context):
    app = API(event, context)

    @app.get("/")
    def hello():
        return {"message": "Hello!"}

    return app.handle_request()

バリデーション

リクエストバリデーション

request_format パラメータを使用してリクエストボディを自動検証できます：

from dataclasses import dataclass

@dataclass
class CreateUserRequest:
    name: str
    email: str
    age: int = 0

@app.post("/users", request_format=CreateUserRequest)
def create_user(request: CreateUserRequest):
    # request は自動的に CreateUserRequest インスタンスに変換される
    return {
        "message": f"User {request.name} created",
        "email": request.email
    }

レスポンスバリデーション

response_format パラメータを使用してレスポンスを自動検証できます：

@dataclass
class UserResponse:
    id: str
    name: str
    email: str

@app.get("/users/{id}", response_format=UserResponse)
def get_user(id: str) -> UserResponse:
    # 戻り値は UserResponse として検証される
    return {
        "id": id,
        "name": f"User {id}",
        "email": f"user{id}@example.com"
    }

パラメータ注入

lambapi は関数シグネチャを解析して、自動的にパラメータを注入します：

パスパラメータ

@app.get("/users/{user_id}/posts/{post_id}")
def get_post(user_id: str, post_id: str):
    return {"user_id": user_id, "post_id": post_id}

クエリパラメータ

@app.get("/items")
def get_items(limit: int = 10, offset: int = 0, active: bool = True):
    return {
        "limit": limit,     # 自動的に int に変換
        "offset": offset,   # 自動的に int に変換
        "active": active    # 自動的に bool に変換
    }

型変換

型注釈	変換動作
str	そのまま文字列
int	int() で変換、失敗時は 0
float	float() で変換、失敗時は 0.0
bool	'true', '1', 'yes', 'on' を True として認識

Request オブジェクトの使用

従来の方式で Request オブジェクトを直接使用することも可能です：

@app.post("/upload")
def upload_file(request):
    # 全リクエスト情報にアクセス
    content_type = request.headers.get("content-type")
    body = request.body
    method = request.method
    path = request.path

    return {"uploaded": True}

高度な使用例

条件付きルート

@app.get("/admin/users")
def admin_users(request):
    # 認証チェック
    auth_header = request.headers.get("authorization")
    if not auth_header:
        raise AuthenticationError("Authentication required")

    return {"users": ["admin_user1", "admin_user2"]}

カスタムレスポンス

from lambapi import Response

@app.get("/download")
def download_file():
    return Response(
        "file content",
        status_code=200,
        headers={
            "Content-Type": "application/octet-stream",
            "Content-Disposition": "attachment; filename=file.txt"
        }
    )

非同期処理（シミュレート）

import time

@app.post("/process")
def start_processing(request):
    # 非同期処理のシミュレート
    task_id = f"task_{int(time.time())}"

    return Response(
        {"task_id": task_id, "status": "processing"},
        status_code=202
    )

関連項目

• Router - ルーターによるエンドポイントのグループ化
• Request - リクエストオブジェクトの詳細
• Response - レスポンスオブジェクトの詳細
• Exceptions - エラーハンドリングと例外クラス