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! 🚀
Top comments (0)