DEV Community: Emmanuel Akolbire

Project Translate: The Translate API (Part 4)

Emmanuel Akolbire — Fri, 21 Feb 2025 12:42:21 +0000

Hey developers! 👋 For the last post in the series, we'll provision the infrastructure on AWS and deploy our API. Let's dive in! You can check out my GitHub for the complete code.

Configure Terraform

First, we'll define the Terraform version and set up the backend to store the state in an S3 bucket. Since Terraform doesn't create the backend bucket automatically, you'll need to provision it beforehand. Variables cannot be used in the backend configuration so the values to be hardcoded but feel free to change them.

#infrastructure/versions.tf

terraform {
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~>5.0"
    }
  }
}

#infrastructure/terraform.tf

terraform {
  backend "s3" {
    bucket = "translate-state-bucket"
    region = "us-east-1"
    key    = "terraform.tfstate"
  }
}

We'll also create a few local variables

#infrastructure/locals.tf

locals {
  tags = {
    Terraform   = true
    Application = "translate"
  }
}

DynamoDB Table

Now we'll define the configuration to provision the DynamoDB table. DynamoDB tables require a name, a hash_key,which should also specified as an attribute, and a billing_mode.

resource "aws_dynamodb_table" "this" {
  name         = "translate-table"
  hash_key     = "id"
  billing_mode = "PAY_PER_REQUEST"

  attribute {
    name = "id"
    type = "S"  #string
  }

  tags = local.tags
}

S3 Bucket

Next, we'll provision the output S3 bucket by specifying a name using either the bucket or bucket_prefix field. To manage storage efficiently, we'll add a lifecycle configuration that automatically deletes files after one day. Lifecycle configurations require the bucket name and at least one rule to define the retention policy.

resource "aws_s3_bucket" "this" {
  bucket_prefix = "translate-output-"
  tags          = local.tags
}


resource "aws_s3_bucket_lifecycle_configuration" "this" { # A lifecycle hook to delete files after 1 day
  bucket = aws_s3_bucket.this.id

  rule {
    id = "rule-1"
    filter {}
    expiration {
      days = 1
    }
    status = "Enabled"
  }
}

Lambda functions

We'll start by assigning the AWSLambdaBasicExecutionRole, which grants essential permissions like creating CloudWatch logs.

To package the Python scripts for the Translate Text and Translate File endpoints, we'll use the archive_file data resource. Since the Translate File API has dependencies, we'll install them beforehand using a null_resource before packaging.

We'll also create an IAM role for the Lambda functions and attach the necessary policies. Finally, we'll define the Lambda functions using the packaged files, set the handler to main.handler, and configure the required environment variables.

data "aws_iam_policy" "this" {
  name = "AWSLambdaBasicExecutionRole"
}


data "archive_file" "translate" {
  type        = "zip"
  source_dir  = "${path.module}/../api/translate"
  output_path = "${path.module}/files/translate.zip"
}


data "archive_file" "translate-file" {
  type        = "zip"
  source_dir  = "${path.module}/files/translate_file"
  output_path = "${path.module}/files/translate-file.zip"

  depends_on = [ null_resource.this ]
}


resource "null_resource" "this" {
  provisioner "local-exec" {
    command = <<EOT
      rm -rf ${path.module}/files/translate_file
      cp -r ${path.module}/../api/translate_file ${path.module}/files/translate_file
      cd ${path.module}/files/translate_file
      pip install -r requirements.txt -t ./
    EOT
  }

  triggers = {
    always_run = timestamp()
  }
}



data "aws_iam_policy_document" "assumeRole" {
  statement {
    effect  = "Allow"
    actions = ["sts:AssumeRole"]
    principals {
      identifiers = ["lambda.amazonaws.com"]
      type        = "Service"
    }
  }
}

data "aws_iam_policy_document" "this" {
  statement {
    effect = "Allow"
    actions = ["translate:TranslateText",
      "translate:TranslateDocument",
      "comprehend:DetectDominantLanguage"
    ]
    resources = ["*"]
  }

  statement {
    effect  = "Allow"
    actions = ["dynamodb:*"]
    resources = [
      aws_dynamodb_table.this.arn,
      "${aws_dynamodb_table.this.arn}/*"
    ]
  }

  statement {
    effect  = "Allow"
    actions = ["s3:*"]
    resources = [
      aws_s3_bucket.this.arn,
      "${aws_s3_bucket.this.arn}/*"
    ]
  }
}

resource "aws_iam_role" "this" {
  name_prefix        = "translate-app-role"
  assume_role_policy = data.aws_iam_policy_document.assumeRole.json
  tags               = local.tags
}

resource "aws_iam_role_policy" "this" {
  role   = aws_iam_role.this.name
  policy = data.aws_iam_policy_document.this.json
}

resource "aws_iam_role_policy_attachment" "this" {
  role       = aws_iam_role.this.name
  policy_arn = data.aws_iam_policy.this.arn
}


resource "aws_lambda_function" "translate" {
  function_name    = "translate-app-translate-lambda"
  description      = "Lambda function for the /translate endpoint"
  role             = aws_iam_role.this.arn
  filename         = data.archive_file.translate.output_path
  handler          = "main.handler"
  source_code_hash = data.archive_file.translate.output_base64sha256
  runtime          = "python3.10"

  environment {
    variables = {
      DYNAMODB_TABLE = aws_dynamodb_table.this.name
    }
  }

  tags = local.tags
}

resource "aws_lambda_function" "translate-file" {
  function_name    = "translate-app-translate-file-lambda"
  description      = "Lambda function for the /translate/file endpoint"
  role             = aws_iam_role.this.arn
  filename         = data.archive_file.translate-file.output_path
  handler          = "main.handler"
  source_code_hash = data.archive_file.translate-file.output_base64sha256
  runtime          = "python3.10"

  environment {
    variables = {
      DYNAMODB_TABLE = aws_dynamodb_table.this.name
      S3_BUCKET      = aws_s3_bucket.this.bucket
    }
  }

  tags = local.tags
}

API Gateway

Finally, we'll define the API Gateway API. To handle file uploads properly, we'll include multipart/form-data in the binary_media_types configuration, ensuring that multipart requests are base64-encoded before being sent to the Lambda function.

Additionally, we'll configure the API resources as deployment triggers, so any changes to their properties automatically trigger a new deployment.

data "aws_caller_identity" "this" {}

data "aws_region" "this" {}


resource "aws_api_gateway_rest_api" "this" {
  name               = "translate-app-api"
  description        = "The translate app API"
  binary_media_types = ["multipart/form-data"]
  tags               = local.tags
}

resource "aws_api_gateway_resource" "translate" {
  rest_api_id = aws_api_gateway_rest_api.this.id
  parent_id   = aws_api_gateway_rest_api.this.root_resource_id
  path_part   = "translate"
}

resource "aws_api_gateway_method" "translate" {
  rest_api_id   = aws_api_gateway_rest_api.this.id
  resource_id   = aws_api_gateway_resource.translate.id
  http_method   = "POST"
  authorization = "NONE"
}

resource "aws_api_gateway_integration" "translate" {
  rest_api_id             = aws_api_gateway_rest_api.this.id
  resource_id             = aws_api_gateway_resource.translate.id
  http_method             = aws_api_gateway_method.translate.http_method
  integration_http_method = "POST"

  type = "AWS_PROXY"
  uri  = aws_lambda_function.translate.invoke_arn
}



resource "aws_api_gateway_resource" "translate-file" {
  rest_api_id = aws_api_gateway_rest_api.this.id
  parent_id   = aws_api_gateway_resource.translate.id
  path_part   = "file"
}

resource "aws_api_gateway_method" "translate-file" {
  rest_api_id   = aws_api_gateway_rest_api.this.id
  resource_id   = aws_api_gateway_resource.translate-file.id
  http_method   = "POST"
  authorization = "NONE"
}

resource "aws_api_gateway_integration" "translate-file" {
  rest_api_id             = aws_api_gateway_rest_api.this.id
  resource_id             = aws_api_gateway_resource.translate-file.id
  http_method             = aws_api_gateway_method.translate-file.http_method

  content_handling = "CONVERT_TO_TEXT"
  integration_http_method = "POST"
  type                    = "AWS_PROXY"
  uri                     = aws_lambda_function.translate-file.invoke_arn
}



resource "aws_api_gateway_deployment" "this" {
  rest_api_id = aws_api_gateway_rest_api.this.id

  triggers = {
    redeployment = sha1(jsonencode([
      aws_api_gateway_rest_api.this,
      aws_api_gateway_resource.translate,
      aws_api_gateway_method.translate,
      aws_api_gateway_integration.translate,
      aws_api_gateway_resource.translate-file,
      aws_api_gateway_method.translate-file,
      aws_api_gateway_integration.translate-file,

    ]))
  }

  lifecycle {
    create_before_destroy = true
  }
}


resource "aws_api_gateway_stage" "this" {
  deployment_id = aws_api_gateway_deployment.this.id
  rest_api_id   = aws_api_gateway_rest_api.this.id
  stage_name    = "prod"

  tags = local.tags
}


resource "aws_lambda_permission" "translate" {
  action        = "lambda:InvokeFunction"
  function_name = aws_lambda_function.translate.function_name
  principal     = "apigateway.amazonaws.com"

  source_arn = "arn:aws:execute-api:${data.aws_region.this.name}:${data.aws_caller_identity.this.account_id}:${aws_api_gateway_rest_api.this.id}/*/${aws_api_gateway_method.translate.http_method}${aws_api_gateway_resource.translate.path}"
}

resource "aws_lambda_permission" "translate-file" {
  action        = "lambda:InvokeFunction"
  function_name = aws_lambda_function.translate-file.function_name
  principal     = "apigateway.amazonaws.com"

  source_arn = "arn:aws:execute-api:${data.aws_region.this.name}:${data.aws_caller_identity.this.account_id}:${aws_api_gateway_rest_api.this.id}/*/${aws_api_gateway_method.translate-file.http_method}${aws_api_gateway_resource.translate-file.path}"
}

We'll also define a few outputs.

#infrastructure/outputs.tf

output api_url {
    value = aws_api_gateway_stage.this.invoke_url
}

To provision the infrastructure, we first configure the AWS CLI with our credentials:

aws configure

Next, we initialize Terraform and apply the configuration:

terraform init
terraform apply

After successful provisioning, we can access the API via the output url.

To wrap up this series, we've walked through writing python code for lambda functions and provisioning a complete infrastructure using Terraform and AWS.
If you've followed along, you should now have an API that can translate text and files running on AWS. But this is just the beginning! There’s always more to explore—whether it’s optimizing performance, integrating monitoring tools, or adding a CI/CD pipeline.

I’d love to hear your thoughts! Drop a comment below if you have questions, insights, or ideas for future topics. 🚀 Thanks for reading, and happy coding! 🎉

Project Translate: The Translate API (Part 3)

Emmanuel Akolbire — Mon, 17 Feb 2025 22:54:58 +0000

Hey developers! 👋 Continuing from the previous post, we'll implement the file translation endpoint using Python, AWS Lambda, and a clean Hexagonal Architecture. Let's dive in! You can check out my GitHub for the complete code.

The Translation Record Model

We'll reuse the Record model from the translate endpoint. This time, instead the input and output text we will store the base64 encoded string of the the input and output bytes

#translate_file/models.py

from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class Record:
    id: str
    input_text: str
    output_text: str
    created_at: datetime = field(default=datetime.now)

Next, let's define the ports that handle different aspects of our application:

FilePersistencePort – Responsible for storing raw byte data to a file.
RequestPersistencePort – Similar to TextPersistencePort, this port manages the persistence of request input and output.
TranslationPort – Handles the translation of file content. These ports act as abstractions, making our system more modular and easier to extend.

#translate_file/ports.py

from typing import Protocol
from models import Record


class RequestPersistencePort(Protocol):
    def save(self, input: bytes, output: bytes) -> Record:
        pass


class FilePersistencePort(Protocol):
    def save(self, file: bytes, extension: str) -> str:
        pass


class TranslationPort(Protocol):
    def translate(self, file: bytes, content_Type: str, lang: str) -> bytes:
        pass

Now, let's define the adapters that implement our ports:

RequestPersistenceAdapter – Stores the request input and output in DynamoDB and returns a Record object.
FilePersistenceAdapter – Saves the file as an object in an S3 bucket and returns a pre-signed URL for access.
AWSTranslateAdapter – Uses AWS Translate to process the file and outputs the translated file as bytes. By following this adapter pattern, we keep our architecture clean, modular, and easily extendable.

#translate_file/adapters.py

import boto3
import base64
from uuid import uuid4
from datetime import datetime

from models import Record


class RequestPersistenceAdapter:
    """
    Implementation of RequestPersistencePort using DynamoDB as storage.
    """

    def __init__(self, table_name: str):
        dynamodb = boto3.resource("dynamodb")
        self.table = dynamodb.Table(table_name)

    def save(self, input: bytes, output: bytes) -> Record:
        """
        Save input and output byptes to DynamoDB.

        Args:
            input: The input bytes to save
            output: The output bytes to save

        Returns:
            Record object containing saved data
        """

        id = str(uuid4())
        created_at = datetime.now()

        input_text = base64.b64encode(input).decode()
        output_text = base64.b64encode(output).decode()

        self.table.put_item(
            Item={
                "id": str(uuid4()),
                "input_text": input_text,
                "output_text": output_text,
                "created_at": str(created_at),
            }
        )

        record = Record(id, input_text, output_text, created_at)
        return record


class FilePersistenceAdpater:
    """
    Implementation of FilePersistencePort using S3
    """

    def __init__(self, bucket_name: str):
        self.client = boto3.client("s3")
        self.bucket_name = bucket_name

    def save(self, file: bytes, extension: str) -> str:
        """
        Save file as object in S3 bucket

        Args:
            file: The file to save

        Returns:
            The url to the file
        """

        key = str(uuid4()).replace("-", "")
        if extension:
            key += f".{extension}"

        response = self.client.put_object(Key=key, Body=file, Bucket=self.bucket_name)

        url = self.client.generate_presigned_url(
            "get_object", Params={"Bucket": self.bucket_name, "Key": key}, ExpiresIn=300
        )

        return url


class AWSTranslateAdapter:
    """
    Implementation of TranslationPort using AWSTranslate.
    """

    def __init__(self):
        self.client = boto3.client("translate")

    def translate(self, file: bytes, content_type: str, lang: str) -> bytes:
        """
        Translate input file to lang

        Args:
            file: The file to translate
            lang: The language code to translate to

        Returns:
            The translated file
        """

        result = self.client.translate_document(
            Document={"Content": file, "ContentType": content_type},
            SourceLanguageCode="auto",
            TargetLanguageCode=lang,
        )

        translated_file = result["TranslatedDocument"]["Content"]

        return translated_file

Next, we define the Handler class to process requests sent to the Lambda function. The files and target language are uploaded using multipart/form-data.

To preserve the integrity of the binary data AWS API Gateway base64 encodes the body when the binaryMediaTypes field is set to multipart/form-data. The TranslationRequest class is responsible for:

Decoding the request body and parsing multipart/form-data.
Extracting the uploaded files.
Identifying the target language for translation.

#translate_file/main.py

import os
import json
import base64
import logging
from dataclasses import dataclass
from typing import Dict, Any, Tuple
from requests_toolbelt import MultipartDecoder

from ports import RequestPersistencePort, FilePersistencePort, TranslationPort
from adapters import (
    RequestPersistenceAdapter,
    FilePersistenceAdpater,
    AWSTranslateAdapter,
)

logger = logging.getLogger(__name__)


@dataclass
class TranslationRequest:
    """Dataclass for translation requests"""

    files: list
    lang: str

    @staticmethod
    def get_name(header: bytes) -> str:
        return header.decode().split(";")[1].split("=")[1].strip('"')

    @staticmethod
    def get_file_info(header: bytes) -> Tuple[str, str]:
        filename, extension = None, None
        if len(header.decode().split(";")) >= 3:
            filename = header.decode().split(";")[2].split("=")[1].strip('"')
            file_ext = filename.split(".")
            if len(file_ext) == 2:
                extension = file_ext[1]

        return filename, extension

    @classmethod
    def from_dict(cls, data: Dict) -> "TranslationRequest":
        body = data["body"]
        content_type = data["headers"].get("content-type", None) or data[
            "headers"].get("Content-Type", None)

        files = []
        lang = ""

        if data["isBase64Encoded"]:
            body = base64.b64decode(body)

        decoder = MultipartDecoder(body, content_type)
        for part in decoder.parts:
            filename, extension = cls.get_file_info(
                part.headers[b"Content-Disposition"]
            )
            if filename:
                content_type = part.headers[b"content-type"].decode()
                files.append((part.content, content_type, extension))
            elif cls.get_name(part.headers[b"Content-Disposition"]) == "lang":
                lang = part.content.decode()

        if not files:
            raise ValueError("files must be provided")

        if not lang:
            raise ValueError("lang must be a string")

        return cls(files, lang)


class Handler:

    def __init__(
        self,
        request_port: RequestPersistencePort,
        file_port: FilePersistencePort,
        translate_port: TranslationPort,
    ):

        self.request_port = request_port
        self.translate_port = translate_port
        self.file_port = file_port

    def __call__(self, request, *args):
        """
        Process a translation request.

        Args:
            request: dict containing the request data

        Returns:
            dict with status code and response body
        """

        try:
            request = TranslationRequest.from_dict(request)
        except (json.JSONDecodeError, KeyError) as e:
            logger.exception(f"Invalid request: {str(e)}")
            return self._get_error_response("Invalid request", status_code=400)

        urls = []
        try:
            lang = request.lang
            for file in request.files:
                result = self.translate_port.translate(file[0], file[1], lang)
                output = self.request_port.save(file[0], result)
                logger.info(f"Saved record with ID: {output.id}")

                url = self.file_port.save(result, file[2])
                urls.append(url)

            return self._get_success_response(urls)
        except Exception as e:
            logger.exception(f"Error translating: {str(e)}")
            return self._get_error_response("An error was encountered", status_code=500)

    def _get_success_response(self, urls: list):
        """
        Generate a successful response.

        Args:
            url: Path to the output file

        Returns:
            Dictionary with status code and response body
        """

        return {"statusCode": "200", 
                "headers": {
                    "Access-Control-Allow-Headers": "Content-Type",
                    "Access-Control-Allow-Origin": "*",
                    "Access-Control-Allow-Methods": "OPTIONS,POST"
                },
                "body": json.dumps({"urls": urls})}

    def _get_error_response(self, error: str, status_code: int):
        """
        Generate an error response.

        Args:
            error: The error message
            status_code: HTTP status code

        Returns:
            Dictionary with status code and response body
        """

        return {"statusCode": str(status_code),
                "headers": {
                    "Access-Control-Allow-Headers": "Content-Type",
                    "Access-Control-Allow-Origin": "*",
                    "Access-Control-Allow-Methods": "OPTIONS,POST"
                },
                "body": json.dumps({"detail": error})}


request_port = RequestPersistenceAdapter(os.environ.get("DYNAMODB_TABLE"))
file_port = FilePersistenceAdpater(os.environ.get("S3_BUCKET"))
translate_port = AWSTranslateAdapter()

handler = Handler(request_port, file_port, translate_port)

Finally we add the required dependencies to a requirement.txt.

#requirements.txt

requests-toolbelt==1.0.0

Super! With our endpoint scripts complete, we now have a functional API for handling translation requests. 🎉
In the next installment, we'll dive into writing the Terraform configuration to provision our infrastructure and deploy our API seamlessly. Stay tuned! 🚀

Project Translate: The Translate API (Part 2)

Emmanuel Akolbire — Sun, 16 Feb 2025 18:52:31 +0000

Hey developers! 👋 In this post, we'll implement the text translation endpoint using Python, AWS Lambda, and a clean Hexagonal Architecture. Let's dive in! You can check out my GitHub for the complete code.

The Setup

We create a new project with the directory structure shown in the picture
Then we install the dependency, namely boto3, with pip. We also make sure to create a requirements.txt file so we know which version to install when the script is packaged.

We'll be employing Hexagonal(Layered) Architecture in the design of our API. Hexagonal Architecture or Ports and Adpaters is a design pattern that aims at creating loosely coupled components. A helpful guide can be found here. Although python is a dynamically typed language, we can still use this pattern.

We'll be using the project directory structure shown below

The Translation Record Model

Let's start with a simple but effective model to track our translations. We'll use Python's dataclasses - they're clean, efficient, and give us nice features out of the box.

#translate/models.py

from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class Record:
    id: str
    input_text: str
    output_text: str
    created_at: datetime = field(default=datetime.now)

Let's break down what each field does:

id: A unique identifier for each translation record
input_text: The original text that needs translation
output_text: The translated result
created_at: Timestamp of when the translation was performed, automatically set to the current time

You might wonder why we're using @dataclass instead of a regular class. Here's what makes dataclasses great for our use case:

Less Boilerplate: We don't need to write init, repr, or eq methods
Default Values: Easy handling of default values with the field function
Type Hints: Built-in support for type hints, making our code more maintainable

Next, we'll define our ports using Python's Protocol class - a more Pythonic approach to interfaces. Let's dive in!
Why Protocols Over Abstract Base Classes?
Before we jump into the code, let's understand why we're choosing Protocols:

More Pythonic - follows duck typing principles
Structural subtyping instead of nominal subtyping
Better integration with static type checkers
No explicit inheritance required

#translate/ports.py

from typing import Protocol
from models import Record


class TextPersistencePort(Protocol):
    def save(self, input_text: str, output_text: str) -> Record:
        pass


class TranslationPort(Protocol):
    def translate(self, text: str, lang: str) -> str:
        pass

Now we define the adapters that implement the ports. The DynamoDBPersistenceAdapter stores the input and output in DynamoDB and return a Record object. The AWSTranslateAdapter translates the text with AWS Translate and returns the result.

#translate/adapters.py

import boto3
from uuid import uuid4
from datetime import datetime

from models import Record


class DynamoDBPersistenceAdapter:
    """
    Implementation of TextPersistencePort using DynamoDB as storage.
    """

    def __init__(self, table_name: str):
        dynamodb = boto3.resource("dynamodb")
        self.table = dynamodb.Table(table_name)

    def save(self, input_text, output_text):
        """
        Save input and output text to DynamoDB.

        Args:
            input_text: The input text to save
            output_text: The output text to save

        Returns:
            Record object containing saved data
        """

        id = str(uuid4())
        created_at = datetime.now()

        self.table.put_item(
            Item={
                "id": str(uuid4()),
                "input_text": input_text,
                "output_text": output_text,
                "created_at": str(created_at),
            }
        )

        record = Record(id, input_text, output_text, created_at)
        return record


class AWSTranslateAdapter:
    """
    Implementation of TranslationPort using DynamoDB as storage.
    """

    def __init__(self):
        self.client = boto3.client("translate")

    def translate(self, text, lang):
        """
        Translate text to lang

        Args:
            text: The text to translate
            lang: The language code to translate to

        Returns:
            The translated string
        """

        result = self.client.translate_text(
            Text=text,
            SourceLanguageCode="auto",
            TargetLanguageCode=lang,
        )

        translated_text = result["TranslatedText"]

        return translated_text

Now we'll create the Lambda handler that ties everything together.
We'll define the Handler class with handles the requests to Lambda from the API Gateway. It parses the body for the required fields, translates the text, stores the input and output and returns a response

#translate/main.py

import os
import json
import logging
from dataclasses import dataclass
from typing import Dict, Any

from ports import TextPersistencePort, TranslationPort
from adapters import DynamoDBPersistenceAdapter, AWSTranslateAdapter

logger = logging.getLogger(__name__)


@dataclass
class TranslationRequest:
    """Dataclass for translation requests"""

    text: str
    lang: str

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "TranslationRequest":
        if not isinstance(data["text"], str):
            raise ValueError("text must be a string")

        if not isinstance(data["lang"], str):
            raise ValueError("lang must be a string")

        return cls(data.get("text"), data.get("lang"))


class Handler:

    def __init__(
        self,
        text_port: TextPersistencePort,
        translate_port: TranslationPort,
    ):

        self.text_port = text_port
        self.translate_port = translate_port

    def __call__(self, request, *args):
        """
        Process a translation request.

        Args:
            request: dict containing the request data

        Returns:
            dict with status code and response body
        """

        try:
            body: dict = json.loads(request["body"])
            request = TranslationRequest.from_dict(body)
        except (json.JSONDecodeError, KeyError) as e:
            logger.error(f"Invalid request: {str(e)}")
            return self._get_error_response("Invalid request", status_code=400)

        try:
            result = self.translate_port.translate(request.text, request.lang)

            output = self.text_port.save(request.text, result)
            logger.info(f"Saved record with ID: {output.id}")

            return self._get_success_response(result)
        except Exception as e:
            return self._get_error_response("An error was encountered", status_code=500)

    def _get_success_response(self, text: str):
        """
        Generate a successful response.

        Args:
            text: The translated text

        Returns:
            Dictionary with status code and response body
        """

        return {
            "statusCode": "200", 
            "headers": {
                "Access-Control-Allow-Headers": "Content-Type",
                "Access-Control-Allow-Origin": "*",
                "Access-Control-Allow-Methods": "OPTIONS,POST,GET"
            },
            "body": json.dumps({"result": text})}

    def _get_error_response(self, error: str, status_code: int):
        """
        Generate an error response.

        Args:
            error: The error message
            status_code: HTTP status code

        Returns:
            Dictionary with status code and response body
        """

        return {"statusCode": str(status_code),
                "headers": {
                    "Access-Control-Allow-Headers": "Content-Type",
                    "Access-Control-Allow-Origin": "*",
                    "Access-Control-Allow-Methods": "OPTIONS,POST,GET"
                },
                "body": json.dumps({"detail": error})}


text_port = DynamoDBPersistenceAdapter(os.environ.get("DYNAMODB_TABLE"))
translate_port = AWSTranslateAdapter()

handler = Handler(text_port, translate_port)

In order to allow Cross Origin Requests we add the Access-Control-Allow headers to the reponse object. For example, in the _get_success_response method


def _get_error_response(self, error: str, status_code: int):
       ...
       return {"statusCode": str(status_code),
                "headers": {
                    "Access-Control-Allow-Headers": "Content-Type",
                    "Access-Control-Allow-Origin": "*",
                    "Access-Control-Allow-Methods": "OPTIONS,POST,GET"
                },
                "body": json.dumps({"detail": error})}

In the next installment of this series, we'll dive into the code that handles file translation. Stay tuned! 🚀

Project Translate : Architecture

Emmanuel Akolbire — Wed, 29 Jan 2025 22:39:17 +0000

Hey there, fellow developers! 👋

Today I'm excited to kick off a series about building Project Translate - a full-featured translation service that handles both text and file translations across multiple languages. The best part? It's built entirely on AWS serverless infrastructure, making it both scalable and cost-effective.

Throughout this series, I'll walk you through the entire journey of creating this REST API from scratch. Whether you're new to serverless or an experienced AWS developer, you'll find valuable insights into building production-ready services in the cloud.

Architecture

Let's break down the architecture that powers our serverless translation service. We'll explore how AWS services work together to handle both text and file translations efficiently.

API Gateway
Serves as the entry point for all REST API requests, handling request routing based on endpoint paths. It provides request validation and rate limiting.

Lambda Functions
The lambda functions process the requests and translate the files and text with AWS Translate.

S3 bucket and DynamoDB
The S3 bucket is used to store output files and the DynamoDB table is used to store input reqeust and output responses.A DynamoDB table is also used to throttle requests to the API

Request Flow

Let's walk through what happens when a translation request hits our API:

Everything starts at the API Gateway, where the request first lands
Based on the endpoint path, API Gateway routes the request to the appropriate Lambda function
Before any translation happens, our Lambda:
Validates the incoming request
Checks if we're within throttling limits
Ensures everything is good to go
Here's where things get interesting. We handle two types of translations:

For Text Translation

Lambda sends text directly to AWS Translate
Stores results in DynamoDB
Returns translated text to the client

For File Translation

File comes in through API Gateway
Lambda kicks off the translation process
Stores the translated file in S3
Generates a pre-signed URL for secure access
Returns the URL to the client

This architecture gives us several advantages:

Scales automatically with demand
Pay only for what we use
No servers to maintain
Built-in high availability

In the next post, we'll dive into the actual implementation, starting with setting up our API Gateway and writing our first Lambda function. Stay tuned! 🚀