FastAPI Lambda with GraphQL & Kafka Integration:

 

FastAPI Lambda with GraphQL & KafkaIntegration: Complete Solution

Step 1: Lambda Code (app.py)

from fastapi import FastAPI, HTTPException, Request

from google.protobuf.json_format import MessageToJson

import puls_pb2  # Your compiled Protobuf file

from datetime import datetime

import strawberry

from strawberry.fastapi import GraphQLRouter

from mangum import Mangum  # Lambda integration package

app = FastAPI()

def authenticate(request: Request):

    username = request.headers.get("username")

    password = request.headers.get("password")

    if username != "aaa" or password != "bbb":

        raise HTTPException(status_code=401, detail="Invalid credentials")

@strawberry.type

class Trade:

    feedmsgseq: int

    sourcetime: int

    symbol: str

    tradeid: int

    price: float

    volume: int

    tradecondition1: str

    tradecondition2: str

    marketid: str

@strawberry.type

class Query:

    @strawberry.field

    def trade(self, symbol: str, start_date: str, end_date: str, info: strawberry.types.Info) -> Trade:

        authenticate(info.context['request'])

        try:

            # Validate date format

            datetime.strptime(start_date, "%Y-%m-%d")

            datetime.strptime(end_date, "%Y-%m-%d")

        except ValueError:

            raise HTTPException(status_code=400, detail="Invalid date format")

        data = puls_pb2.Trade(

            feedmsgseq=123456789,

            sourcetime=1631023200,

            symbol=symbol,

            tradeid=1001,

            price=150.5,

            volume=200,

            tradecondition1="A",

            tradecondition2="B",

            marketid="NASDAQ"

        )

        return Trade(

            feedmsgseq=data.feedmsgseq,

            sourcetime=data.sourcetime,

            symbol=data.symbol,

            tradeid=data.tradeid,

            price=data.price,

            volume=data.volume,

            tradecondition1=data.tradecondition1,

            tradecondition2=data.tradecondition2,

            marketid=data.marketid

        )

# Setup the GraphQL router

schema = strawberry.Schema(query=Query)

graphql_app = GraphQLRouter(schema)

# Include the GraphQL app

app.include_router(graphql_app, prefix="/graphql")

# Use Mangum to integrate with Lambda

handler = Mangum(app)

Step 2: Prepare Lambda Package

1. Create a requirements.txt file:
   

fastapi==0.95.0

mangum==0.10.0

strawberry-graphql==0.19.1

google==3.0.3

protobuf==3.20.1

Step 3 Install dependencies:

pip install -r requirements.txt -t lambda_package/

Step 4 Package Lambda Code:

cd lambda_package

zip -r ../your-code.zip .

This will create the your-code.zip file containing all the necessary code and dependencies.

Step 5: Upload Lambda Code to S3

Go to the AWS S3 console.

Upload the your-code.zip file to an S3 bucket.

Step 6: CloudFormation Template (CFT)

Here is the CloudFormation Template (CFT) for deploying the Lambda, API Gateway, and Load Balancer.

AWSTemplateFormatVersion: '2010-09-09'

Resources:

  

  # Lambda Function

  FastAPILambdaFunction:

    Type: 'AWS::Lambda::Function'

    Properties:

      Handler: 'app.handler'

      Runtime: 'python3.8'

      Code:

        S3Bucket: 'your-bucket-name'  # Replace with your S3 bucket name

        S3Key: 'your-code.zip'        # Replace with the path to your zip file in S3

      MemorySize: 128

      Timeout: 30

      Environment:

        Variables:

          ENVIRONMENT: 'production'

  # API Gateway (HTTP API)

  ApiGateway:

    Type: 'AWS::ApiGatewayV2::Api'

    Properties:

      ProtocolType: 'HTTP'

      Target:

        Fn::GetAtt:

          - 'FastAPILambdaFunction'

          - 'Arn'

  # Load Balancer

  LoadBalancer:

    Type: 'AWS::ElasticLoadBalancingV2::LoadBalancer'

    Properties:

      Subnets:

        - Ref: 'SubnetId'

      SecurityGroups:

        - Ref: 'SecurityGroupId'

Outputs:

  FastAPIAppURL:

    Description: 'URL of the FastAPI application'

    Value: !Sub 'http://${LoadBalancer.DNSName}/graphql'

Key Parts of the CloudFormation Template:

• Lambda Function (FastAPILambdaFunction): Defines the Lambda function that runs your FastAPI code. The .zip file from S3 is used as the code source.

• API Gateway (ApiGateway): The HTTP API Gateway that triggers the Lambda function for HTTP requests.

• Load Balancer (LoadBalancer): An Application Load Balancer to distribute traffic.

Step 7: Deploying CloudFormation

1. Upload the CloudFormation Template to AWS CloudFormation via the AWS Console or AWS CLI.

2. Update the S3 bucket and .zip file paths in the template (your-bucket-name and your-code.zip).

3. Deploy the CloudFormation stack. This will automatically create:

• The Lambda function with your FastAPI code.

• The API Gateway to handle incoming HTTP requests.

• The Load Balancer to distribute traffic.

Step 8: Example GraphQL Query

Once the stack is deployed, you can query your FastAPI app using GraphQL. The app is exposed through the Load Balancer's DNS name.

GraphQL Query Example (POST request to http://<your-load-balancer-dns-name>/graphql):

{

  "query": "query { trade(symbol: \"AAPL\", start_date: \"2024-03-01\", end_date: \"2024-03-10\") { feedmsgseq sourcetime symbol tradeid price volume tradecondition1 tradecondition2 marketid }}"

}

Summary

This document guides you through:

1. Lambda Code: Writing a FastAPI application integrated with GraphQL.

2. Packaging the Lambda: Installing dependencies and creating a deployable .zip file.

3. CloudFormation Template: Automating the deployment of the Lambda function, API Gateway, and Load Balancer.

4. Deployment: Instructions on uploading the Lambda code to S3 and deploying with CloudFormation.

5. Querying: How to query the deployed FastAPI app via GraphQL.