컨텐츠로 건너뛰기
@aws/nx-plugin
Blog

FastAPI

FastAPI는 Python으로 API를 구축하기 위한 프레임워크입니다.

FastAPI 생성자는 AWS CDK 인프라 설정이 포함된 새로운 FastAPI를 생성합니다. 생성된 백엔드는 서버리스 배포를 위해 AWS Lambda를 사용하며 AWS API Gateway HTTP API를 통해 노출됩니다. 로깅, AWS X-Ray 추적 및 Cloudwatch 메트릭을 포함한 관측 가능성을 위해 AWS Lambda Powertools가 설정됩니다.

사용 방법

FastAPI 생성

다음 두 가지 방법으로 새로운 FastAPI를 생성할 수 있습니다:

  1. 설치 Nx Console VSCode Plugin 아직 설치하지 않았다면
  2. VSCode에서 Nx 콘솔 열기
  3. 클릭 Generate (UI) "Common Nx Commands" 섹션에서
  4. 검색 @aws/nx-plugin - py#fast-api
  5. 필수 매개변수 입력
    • 클릭 Generate

    옵션

    매개변수 타입 기본값 설명
    name 필수 string - project name.
    directory string packages The directory to store the application in.

    생성자 출력

    생성자는 <directory>/<api-name> 디렉토리에 다음 프로젝트 구조를 생성합니다:

    • project.json 프로젝트 설정 및 빌드 타겟
    • pyproject.toml Python 프로젝트 설정 및 의존성
    • 디렉터리<module_name>
      • __init__.py 모듈 초기화
      • init.py FastAPI 앱 설정 및 powertools 미들웨어 구성
      • main.py API 구현

    생성자는 또한 API 배포에 사용할 수 있는 CDK 구문을 생성하며, 이는 packages/common/constructs 디렉토리에 위치합니다.

    FastAPI 구현

    주요 API 구현은 main.py에 있습니다. 여기서 API 경로와 구현을 정의합니다. 예시:

    from .init import app, tracer
    from pydantic import BaseModel
    

    class Item(BaseModel):
      name: str
    

    @app.get("/items/{item_id}")
    def get_item(item_id: int) -> Item:
        return Item(name=...)
    

    @app.post("/items")
    def create_item(item: Item):
        return ...

    생성자는 자동으로 여러 기능을 설정합니다:

    1. 관측 가능성을 위한 AWS Lambda Powertools 통합
    2. 오류 처리 미들웨어
    3. 요청/응답 상관 관계
    4. 메트릭 수집
    5. Mangum을 사용한 AWS Lambda 핸들러

    AWS Lambda Powertools를 통한 관측 가능성

    로깅

    생성자는 AWS Lambda Powertools를 사용한 구조화된 로깅을 구성합니다. 라우트 핸들러에서 로거에 접근할 수 있습니다:

    from .init import app, logger
    

    @app.get("/items/{item_id}")
    def read_item(item_id: int):
        logger.info("Fetching item", extra={"item_id": item_id})
        return {"item_id": item_id}

    로거는 자동으로 다음을 포함합니다:

    • 요청 추적을 위한 상관 ID
    • 요청 경로 및 메소드
    • Lambda 컨텍스트 정보
    • 콜드 스타트 표시기

    추적

    AWS X-Ray 추적이 자동으로 구성됩니다. 추적에 사용자 정의 하위 세그먼트를 추가할 수 있습니다:

    from .init import app, tracer
    

    @app.get("/items/{item_id}")
    @tracer.capture_method
    def read_item(item_id: int):
        # 새로운 하위 세그먼트 생성
        with tracer.provider.in_subsegment("fetch-item-details"):
            # 로직 구현
            return {"item_id": item_id}

    메트릭

    CloudWatch 메트릭이 각 요청에 대해 자동으로 수집됩니다. 사용자 정의 메트릭을 추가할 수 있습니다:

    from .init import app, metrics
    from aws_lambda_powertools.metrics import MetricUnit
    

    @app.get("/items/{item_id}")
    def read_item(item_id: int):
        metrics.add_metric(name="ItemViewed", unit=MetricUnit.Count, value=1)
        return {"item_id": item_id}

    기본 메트릭 포함 사항:

    • 요청 횟수
    • 성공/실패 횟수
    • 콜드 스타트 메트릭
    • 경로별 메트릭

    오류 처리

    생성자는 포괄적인 오류 처리를 포함합니다:

    from fastapi import HTTPException
    

    @app.get("/items/{item_id}")
    def read_item(item_id: int):
        if item_id < 0:
            raise HTTPException(status_code=400, detail="Item ID must be positive")
        return {"item_id": item_id}

    처리되지 않은 예외는 미들웨어에 의해 포착되어:

    1. 전체 예외와 스택 추적 로깅
    2. 실패 메트릭 기록
    3. 클라이언트에게 안전한 500 응답 반환
    4. 상관 ID 유지

    스트리밍

    FastAPI를 사용하면 StreamingResponse 응답 타입으로 호출자에게 스트리밍 응답을 보낼 수 있습니다.

    인프라 변경

    AWS API Gateway는 스트리밍 응답을 지원하지 않으므로 FastAPI를 이를 지원하는 플랫폼에 배포해야 합니다. 가장 간단한 옵션은 AWS Lambda Function URL을 사용하는 것입니다. 이를 위해 생성된 HttpApi 구문을 조정하여 스트리밍 옵션을 추가하고 관련 구문을 조건부로 인스턴스화할 수 있습니다.

    변경 예시
     import { Construct } from 'constructs';
    import { CfnOutput, Duration } from 'aws-cdk-lib';
    import { CfnOutput, Duration, Stack } from 'aws-cdk-lib';
     import {
       CorsHttpMethod,
       HttpApi as _HttpApi,
    @@ -7,7 +7,16 @@ import {
       IHttpRouteAuthorizer,
     } from 'aws-cdk-lib/aws-apigatewayv2';
    

           },
         });
    

        this.api = new _HttpApi(this, id, {
          corsPreflight: {
            allowOrigins: props.allowedOrigins ?? ['*'],
            allowMethods: [CorsHttpMethod.ANY],
            allowHeaders: [
              'authorization',
              'content-type',
              'x-amz-content-sha256',
              'x-amz-date',
              'x-amz-security-token',
            ],
          },
          defaultAuthorizer: props.defaultAuthorizer,
        });
        let apiUrl;
        if (props.apiType === 'api-gateway') {
          this.api = new _HttpApi(this, id, {
            corsPreflight: {
              allowOrigins: props.allowedOrigins ?? ['*'],
              allowMethods: [CorsHttpMethod.ANY],
              allowHeaders: [
                'authorization',
                'content-type',
                'x-amz-content-sha256',
                'x-amz-date',
                'x-amz-security-token',
              ],
            },
            defaultAuthorizer: props.defaultAuthorizer,
          });
    

        this.api.addRoutes({
          path: '/{proxy+}',
          methods: [
            HttpMethod.GET,
            HttpMethod.DELETE,
            HttpMethod.POST,
            HttpMethod.PUT,
            HttpMethod.PATCH,
            HttpMethod.HEAD,
          ],
          integration: new HttpLambdaIntegration(
            'RouterIntegration',
            this.routerFunction,
          ),
        });
          this.api.addRoutes({
            path: '/{proxy+}',
            methods: [
              HttpMethod.GET,
              HttpMethod.DELETE,
              HttpMethod.POST,
              HttpMethod.PUT,
              HttpMethod.PATCH,
              HttpMethod.HEAD,
            ],
            integration: new HttpLambdaIntegration(
              'RouterIntegration',
              this.routerFunction,
            ),
          });
          apiUrl = this.api.url;
        } else {
          const stack = Stack.of(this);
          this.routerFunction.addLayers(
            LayerVersion.fromLayerVersionArn(
              this,
              'LWALayer',
              `arn:aws:lambda:${stack.region}:753240598075:layer:LambdaAdapterLayerX86:24`,
            ),
          );
          this.routerFunction.addEnvironment('PORT', '8000');
          this.routerFunction.addEnvironment(
            'AWS_LWA_INVOKE_MODE',
            'response_stream',
          );
          this.routerFunction.addEnvironment(
            'AWS_LAMBDA_EXEC_WRAPPER',
            '/opt/bootstrap',
          );
          this.routerFunctionUrl = this.routerFunction.addFunctionUrl({
            authType: FunctionUrlAuthType.AWS_IAM,
            invokeMode: InvokeMode.RESPONSE_STREAM,
            cors: {
              allowedOrigins: props.allowedOrigins ?? ['*'],
              allowedHeaders: [
                'authorization',
                'content-type',
                'x-amz-content-sha256',
                'x-amz-date',
                'x-amz-security-token',
              ],
            },
          });
          apiUrl = this.routerFunctionUrl.url;
        }
    

        new CfnOutput(this, `${props.apiName}Url`, { value: this.api.url! });
        new CfnOutput(this, `${props.apiName}Url`, { value: apiUrl! });
    

         RuntimeConfig.ensure(this).config.httpApis = {
           ...RuntimeConfig.ensure(this).config.httpApis!,
          [props.apiName]: this.api.url!,
          [props.apiName]: apiUrl,
         };
       }
    

       public grantInvokeAccess(role: IRole) {
        role.addToPrincipalPolicy(
          new PolicyStatement({
            effect: Effect.ALLOW,
            actions: ['execute-api:Invoke'],
            resources: [this.api.arnForExecuteApi('*', '/*', '*')],
          }),
        );
        if (this.api) {
          role.addToPrincipalPolicy(
            new PolicyStatement({
              effect: Effect.ALLOW,
              actions: ['execute-api:Invoke'],
              resources: [this.api.arnForExecuteApi('*', '/*', '*')],
            }),
          );
        } else if (this.routerFunction) {
          role.addToPrincipalPolicy(
            new PolicyStatement({
              effect: Effect.ALLOW,
              actions: ['lambda:InvokeFunctionUrl'],
              resources: [this.routerFunction.functionArn],
              conditions: {
                StringEquals: {
                  'lambda:FunctionUrlAuthType': 'AWS_IAM',
                },
              },
            }),
          );
        }
       }
     }

    이 변경 사항을 적용한 후 packages/common/constructs/src/app/http-apis/<my-api>.ts를 업데이트하여 새로운 함수 URL 옵션을 사용하도록 해야 합니다.

    구현

    스트리밍을 지원하도록 인프라를 업데이트한 후 FastAPI에서 스트리밍 API를 구현할 수 있습니다. API는 다음을 수행해야 합니다:

    예를 들어 API에서 JSON 객체 스트림을 보내려면 다음과 같이 구현할 수 있습니다:

    from pydantic import BaseModel
    from fastapi.responses import StreamingResponse
    

    class Chunk(BaseModel):
      message: str
      timestamp: datetime
    

    async def stream_chunks():
      for i in range(0, 100):
        yield Chunk(message=f"This is chunk {i}", timestamp=datetime.now())
    

    @app.get("/stream", openapi_extra={'x-streaming': True})
    def my_stream() -> Chunk:
        return StreamingResponse(stream_chunks(), media_type="application/json")

    소비

    스트리밍 응답을 소비하려면 API Connection Generator를 사용하여 스트리밍 청크를 반복하는 타입 세이프 메서드를 활용할 수 있습니다.

    FastAPI 배포

    FastAPI 생성자는 common/constructs 폴더에 API 배포를 위한 CDK 구문을 생성합니다. CDK 애플리케이션에서 이를 사용할 수 있습니다:

    import { MyApi } from ':my-scope/common-constructs';
    

    export class ExampleStack extends Stack {
      constructor(scope: Construct, id: string) {
        // 스택에 API 추가
        const api = new MyApi(this, 'MyApi');
      }
    }

    이 설정에는 다음이 포함됩니다:

    1. FastAPI 애플리케이션을 실행하는 AWS Lambda 함수
    2. 함수 트리거로 사용되는 API Gateway HTTP API
    3. IAM 역할 및 권한
    4. CloudWatch 로그 그룹
    5. X-Ray 추적 구성
    6. CloudWatch 메트릭 네임스페이스

    접근 권한 부여

    grantInvokeAccess 메서드를 사용하여 API 접근 권한을 부여할 수 있습니다:

    api.grantInvokeAccess(myIdentityPool.authenticatedRole);

    로컬 개발

    생성자는 다음 명령으로 실행할 수 있는 로컬 개발 서버를 구성합니다:

    Terminal window
    pnpm nx run my-api:serve

    이 서버는 다음 기능을 제공합니다:

    • 코드 변경 시 자동 재로드
    • /docs 또는 /redoc에서 대화형 API 문서
    • /openapi.json에서 OpenAPI 스키마

    FastAPI 호출

    React 웹사이트에서 API를 호출하려면 api-connection 생성자를 사용할 수 있습니다.