FastAPI

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

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

사용 방법

FastAPI 생성

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

VSCode

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

CLI

pnpm

pnpm nx g @aws/nx-plugin:py#fast-api

yarn

yarn nx g @aws/nx-plugin:py#fast-api

npm

npx nx g @aws/nx-plugin:py#fast-api

bun

bunx nx g @aws/nx-plugin:py#fast-api

어떤 파일이 변경될지 확인하기 위해 드라이 런을 수행할 수도 있습니다

pnpm

pnpm nx g @aws/nx-plugin:py#fast-api --dry-run

yarn

yarn nx g @aws/nx-plugin:py#fast-api --dry-run

npm

npx nx g @aws/nx-plugin:py#fast-api --dry-run

bun

bunx nx g @aws/nx-plugin:py#fast-api --dry-run

옵션

매개변수	타입	기본값	설명
name 필수	string	-	Name of the API project to generate
computeType	string	ServerlessApiGatewayRestApi	The type of compute to use to deploy this API. Choose between ServerlessApiGatewayRestApi (default) or ServerlessApiGatewayHttpApi.
auth	string	IAM	The method used to authenticate with your API. Choose between IAM (default), Cognito or None.
directory	string	packages	The directory to store the application in.
moduleName	string	-	Python module name

참고

API를 배포할 때 REST API 또는 HTTP API 중 선택할 수 있습니다. 결정에 도움이 필요하면 API Gateway Developer Guide를 참조하세요.

생성자 출력 결과

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

• project.json 프로젝트 구성 및 빌드 타겟
• pyproject.toml Python 프로젝트 구성 및 의존성
• 디렉터리<module_name>

  • __init__.py 모듈 초기화
  • init.py FastAPI 앱 설정 및 powertools 미들웨어 구성
  • main.py API 구현
• 디렉터리scripts

  • generate_open_api.py FastAPI 앱에서 OpenAPI 스키마 생성 스크립트

생성자는 또한 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("아이템 조회 중", 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는 양수여야 합니다")
    return {"item_id": item_id}

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

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

팁

api-connection 생성자를 사용하는 경우 더 나은 코드 생성을 위해 API 작업에 대한 응답 모델을 명시하는 것이 좋습니다. 자세한 내용 참조.

스트리밍

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

인프라 변경

AWS API Gateway는 스트리밍 응답을 지원하지 않으므로 이를 지원하는 플랫폼에 FastAPI를 배포해야 합니다. 가장 간단한 옵션은 AWS Lambda Function URL을 사용하는 것입니다. 이를 위해 생성된 common/constructs/src/app/apis/<name>-api.ts 구성을 Function URL을 배포하는 구성으로 교체할 수 있습니다.

스트리밍 FunctionURL 구성 예시

import { Duration, Stack, CfnOutput } from 'aws-cdk-lib';
import { IGrantable, Grant } from 'aws-cdk-lib/aws-iam';
import {
  Runtime,
  Code,
  Tracing,
  LayerVersion,
  FunctionUrlAuthType,
  InvokeMode,
  Function,
} from 'aws-cdk-lib/aws-lambda';
import { Construct } from 'constructs';
import url from 'url';
import { RuntimeConfig } from '../../core/runtime-config.js';

export class MyApi extends Construct {
  public readonly handler: Function;

  constructor(scope: Construct, id: string) {
    super(scope, id);

    this.handler = new Function(this, 'Handler', {
      runtime: Runtime.PYTHON_3_12,
      handler: 'run.sh',
      code: Code.fromAsset(
        url.fileURLToPath(
          new URL(
            '../../../../../../dist/packages/my_api/bundle',
            import.meta.url,
          ),
        ),
      ),
      timeout: Duration.seconds(30),
      tracing: Tracing.ACTIVE,
      environment: {
        AWS_CONNECTION_REUSE_ENABLED: '1',
      },
    });

    const stack = Stack.of(this);
    this.handler.addLayers(
      LayerVersion.fromLayerVersionArn(
        this,
        'LWALayer',
        `arn:aws:lambda:${stack.region}:753240598075:layer:LambdaAdapterLayerX86:24`,
      ),
    );
    this.handler.addEnvironment('PORT', '8000');
    this.handler.addEnvironment('AWS_LWA_INVOKE_MODE', 'response_stream');
    this.handler.addEnvironment('AWS_LAMBDA_EXEC_WRAPPER', '/opt/bootstrap');
    const functionUrl = this.handler.addFunctionUrl({
      authType: FunctionUrlAuthType.AWS_IAM,
      invokeMode: InvokeMode.RESPONSE_STREAM,
      cors: {
        allowedOrigins: ['*'],
        allowedHeaders: [
          'authorization',
          'content-type',
          'x-amz-content-sha256',
          'x-amz-date',
          'x-amz-security-token',
        ],
      },
    });

    new CfnOutput(this, 'MyApiUrl', { value: functionUrl.url });

    // 클라이언트 디스커버리를 위해 런타임 구성에 API URL 등록
    RuntimeConfig.ensure(this).config.apis = {
      ...RuntimeConfig.ensure(this).config.apis!,
      MyApi: functionUrl.url,
    };
  }

  public grantInvokeAccess(grantee: IGrantable) {
    Grant.addToPrincipal({
      grantee,
      actions: ['lambda:InvokeFunctionUrl'],
      resourceArns: [this.handler.functionArn],
      conditions: {
        StringEquals: {
          'lambda:FunctionUrlAuthType': 'AWS_IAM',
        },
      },
    });
  }
}

참고

종단간 예시는 던전 어드벤처 튜토리얼을 참조하세요.

구현

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

• StreamingResponse 반환
• 각 응답 청크의 반환 타입 선언
• API 연결 사용 시 OpenAPI 벤더 확장 x-streaming: true 추가

예를 들어 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 연결 생성자를 사용하여 타입 안전한 청크 반복 방법을 제공할 수 있습니다.

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', {
      integrations: MyApi.defaultIntegrations(this).build(),
    });
  }
}

이 설정은 다음을 구성합니다:

1. FastAPI 애플리케이션의 각 작업에 대한 AWS Lambda 함수
2. 함수 트리거로 API Gateway HTTP/REST API
3. IAM 역할 및 권한
4. CloudWatch 로그 그룹
5. X-Ray 추적 구성
6. CloudWatch 메트릭 네임스페이스

참고

Cognito 인증을 선택한 경우 API 구성에 identity 속성을 제공해야 합니다:

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

export class ExampleStack extends Stack {
  constructor(scope: Construct, id: string) {
    const identity = new UserIdentity(this, 'Identity');

    const api = new MyApi(this, 'MyApi', {
      integrations: MyApi.defaultIntegrations(this).build(),
      identity,
    });
  }
}

UserIdentity 구성은 ts#cloudscape-website-auth 생성자를 사용하여 생성할 수 있습니다.

타입 안전 통합

REST/HTTP API CDK 구성체는 각 작업에 대한 통합을 정의하기 위한 타입 안전 인터페이스를 제공하도록 구성됩니다.

기본 통합

정적 defaultIntegrations를 사용하여 각 작업별로 개별 AWS Lambda 함수를 정의하는 기본 패턴을 활용할 수 있습니다:

new MyApi(this, 'MyApi', {
  integrations: MyApi.defaultIntegrations(this).build(),
});

통합 접근

API 구성체의 integrations 속성을 통해 타입 안전 방식으로 기본 AWS Lambda 함수에 접근할 수 있습니다. 예를 들어 API가 sayHello 작업을 정의하고 이 함수에 일부 권한을 추가해야 하는 경우 다음과 같이 할 수 있습니다:

const api = new MyApi(this, 'MyApi', {
  integrations: MyApi.defaultIntegrations(this).build(),
});

// sayHello는 API에 정의된 작업에 맞게 타입이 지정됩니다
api.integrations.sayHello.handler.addToRolePolicy(new PolicyStatement({
  effect: Effect.ALLOW,
  actions: [...],
  resources: [...],
}));

기본 옵션 커스터마이징

기본 통합 생성 시 사용되는 옵션을 커스터마이징하려면 withDefaultOptions 메서드를 사용할 수 있습니다. 예를 들어 모든 Lambda 함수를 VPC에 배치하려는 경우:

const vpc = new Vpc(this, 'Vpc', ...);

new MyApi(this, 'MyApi', {
  integrations: MyApi.defaultIntegrations(this)
    .withDefaultOptions({
      vpc,
    })
    .build(),
});

통합 재정의

withOverrides 메서드를 사용하여 특정 작업에 대한 통합을 재정의할 수 있습니다. 각 재정의는 HTTP 또는 REST API에 적합한 CDK 통합 구성체를 타입 안전 방식으로 지정해야 합니다. 예를 들어 getDocumentation API를 외부 웹사이트 호스팅 문서로 재정의하려는 경우:

new MyApi(this, 'MyApi', {
  integrations: MyApi.defaultIntegrations(this)
    .withOverrides({
      getDocumentation: {
        integration: new HttpIntegration('https://example.com/documentation'),
      },
    })
    .build(),
});

재정된 통합은 api.integrations.getDocumentation을 통해 접근할 때 더 이상 handler 속성을 갖지 않습니다.

추가 속성을 통합에 포함시켜 타입 안전성을 유지하면서 다른 유형의 통합을 추상화할 수 있습니다. 예를 들어 REST API용 S3 통합을 생성한 후 특정 작업에 대한 버킷을 참조하려는 경우:

const storageBucket = new Bucket(this, 'Bucket', { ... });

const apiGatewayRole = new Role(this, 'ApiGatewayS3Role', {
  assumedBy: new ServicePrincipal('apigateway.amazonaws.com'),
});

storageBucket.grantRead(apiGatewayRole);

const api = new MyApi(this, 'MyApi', {
  integrations: MyApi.defaultIntegrations(this)
    .withOverrides({
      getFile: {
        bucket: storageBucket,
        integration: new AwsIntegration({
          service: 's3',
          integrationHttpMethod: 'GET',
          path: `${storageBucket.bucketName}/{fileName}`,
          options: {
            credentialsRole: apiGatewayRole,
            requestParameters: {
              'integration.request.path.fileName': 'method.request.querystring.fileName',
            },
            integrationResponses: [{ statusCode: '200' }],
          },
        }),
        options: {
          requestParameters: {
            'method.request.querystring.fileName': true,
          },
          methodResponses: [{
            statusCode: '200',
          }],
        }
      },
    })
    .build(),
});

// 다른 파일에서 타입 안전 방식으로 정의한 버킷 속성에 접근 가능
api.integrations.getFile.bucket.grantRead(...);

인증자 재정의

통합에서 options를 제공하여 특정 메서드 옵션(예: Cognito 인증 사용)을 재정의할 수 있습니다:

new MyApi(this, 'MyApi', {
  integrations: MyApi.defaultIntegrations(this)
    .withOverrides({
      getDocumentation: {
        integration: new HttpIntegration('https://example.com/documentation'),
        options: {
          authorizer: new CognitoUserPoolsAuthorizer(...) // REST용 또는 HTTP API용 HttpUserPoolAuthorizer
        }
      },
    })
    .build(),
});

명시적 통합

기본 통합 대신 각 작업에 직접 통합을 제공할 수 있습니다. 이는 각 작업이 다른 유형의 통합을 사용해야 하거나 새 작업 추가 시 타입 오류를 받고자 할 때 유용합니다:

new MyApi(this, 'MyApi', {
  integrations: {
    sayHello: {
      integration: new LambdaIntegration(...),
    },
    getDocumentation: {
      integration: new HttpIntegration(...),
    },
  },
});

라우터 패턴

단일 Lambda 함수를 사용하여 모든 API 요청을 처리하려는 경우 API의 defaultIntegrations 메서드를 수정하여 통합별이 아닌 단일 함수를 생성할 수 있습니다:

packages/common/constructs/src/app/apis/my-api.ts

export class MyApi<...> extends ... {

  public static defaultIntegrations = (scope: Construct) => {
    const router = new Function(scope, 'RouterHandler', { ... });
    return IntegrationBuilder.rest({
      ...
      defaultIntegrationOptions: {},
      buildDefaultIntegration: (op) => {
        return {
          // 모든 통합에서 동일한 라우터 Lambda 핸들러 참조
          integration: new LambdaIntegration(router),
        };
      },
    });
  };
}

router 함수를 메서드 내부에서 생성하는 대신 defaultIntegrations의 매개변수로 정의하는 등 다른 방식으로 코드를 수정할 수도 있습니다.

코드 생성

FastAPI의 작업이 Python으로 정의되고 인프라가 TypeScript로 정의되므로 타입 안전 통합을 위한 메타데이터를 CDK 구성에 제공하기 위해 코드 생성을 도입합니다.

공통 구성의 project.json에 generate:<ApiName>-metadata 타겟이 추가되어 packages/common/constructs/src/generated/my-api/metadata.gen.ts와 같은 파일을 생성합니다. 이는 빌드 시 생성되므로 버전 관리에서 제외됩니다.

참고

API를 변경할 때마다 CDK 구성에서 사용하는 타입이 최신 상태인지 확인하기 위해 빌드를 실행해야 합니다.

pnpm

pnpm nx run-many --target build --all

yarn

yarn nx run-many --target build --all

npm

npx nx run-many --target build --all

bun

bunx nx run-many --target build --all

팁

CDK 인프라와 FastAPI를 동시에 작업하는 경우 nx watch를 사용하여 API 변경 시마다 타입을 재생성할 수 있습니다:

pnpm

pnpm nx watch --projects=<FastAPIProject> -- \
pnpm nx run <InfraProject>:"generate:<ApiName>-metadata"

yarn

yarn nx watch --projects=<FastAPIProject> -- \
yarn nx run <InfraProject>:"generate:<ApiName>-metadata"

npm

npx nx watch --projects=<FastAPIProject> -- \
npx nx run <InfraProject>:"generate:<ApiName>-metadata"

bun

bunx nx watch --projects=<FastAPIProject> -- \
bunx nx run <InfraProject>:"generate:<ApiName>-metadata"

접근 권한 부여 (IAM 전용)

IAM 인증을 선택한 경우 grantInvokeAccess 메소드를 사용하여 API 접근 권한을 부여할 수 있습니다:

api.grantInvokeAccess(myIdentityPool.authenticatedRole);

로컬 개발

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

pnpm

pnpm nx run my-api:serve

yarn

yarn nx run my-api:serve

npm

npx nx run my-api:serve

bun

bunx nx run my-api:serve

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

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

FastAPI 호출

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