Project Translate: The Translate API (Part 2)

0578

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. 

<span>#translate/models.py </span>
<span>from</span> <span>dataclasses</span> <span>import</span> <span>dataclass</span><span>,</span> <span>field</span>
<span>from</span> <span>datetime</span> <span>import</span> <span>datetime</span>
<span>@dataclass</span>
<span>class</span> <span>Record</span><span>:</span>
    <span>id</span><span>:</span> <span>str</span>
    <span>input_text</span><span>:</span> <span>str</span>
    <span>output_text</span><span>:</span> <span>str</span>
    <span>created_at</span><span>:</span> <span>datetime</span> <span>=</span> <span>field</span><span>(</span><span>default</span><span>=</span><span>datetime</span><span>.</span><span>now</span><span>)</span>
<span>#translate/models.py </span>
<span>from</span> <span>dataclasses</span> <span>import</span> <span>dataclass</span><span>,</span> <span>field</span>
<span>from</span> <span>datetime</span> <span>import</span> <span>datetime</span>

<span>@dataclass</span>
<span>class</span> <span>Record</span><span>:</span>
    <span>id</span><span>:</span> <span>str</span>
    <span>input_text</span><span>:</span> <span>str</span>
    <span>output_text</span><span>:</span> <span>str</span>
    <span>created_at</span><span>:</span> <span>datetime</span> <span>=</span> <span>field</span><span>(</span><span>default</span><span>=</span><span>datetime</span><span>.</span><span>now</span><span>)</span>
#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)

Enter fullscreen mode Exit fullscreen mode

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 
<span>#translate/ports.py </span>
<span>from</span> <span>typing</span> <span>import</span> <span>Protocol</span>
<span>from</span> <span>models</span> <span>import</span> <span>Record</span>
<span>class</span> <span>TextPersistencePort</span><span>(</span><span>Protocol</span><span>):</span>
    <span>def</span> <span>save</span><span>(</span><span>self</span><span>,</span> <span>input_text</span><span>:</span> <span>str</span><span>,</span> <span>output_text</span><span>:</span> <span>str</span><span>)</span> <span>-></span> <span>Record</span><span>:</span>
        <span>pass</span>
<span>class</span> <span>TranslationPort</span><span>(</span><span>Protocol</span><span>):</span>
    <span>def</span> <span>translate</span><span>(</span><span>self</span><span>,</span> <span>text</span><span>:</span> <span>str</span><span>,</span> <span>lang</span><span>:</span> <span>str</span><span>)</span> <span>-></span> <span>str</span><span>:</span>
        <span>pass</span>
<span>#translate/ports.py </span>
<span>from</span> <span>typing</span> <span>import</span> <span>Protocol</span>
<span>from</span> <span>models</span> <span>import</span> <span>Record</span>


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


<span>class</span> <span>TranslationPort</span><span>(</span><span>Protocol</span><span>):</span>
    <span>def</span> <span>translate</span><span>(</span><span>self</span><span>,</span> <span>text</span><span>:</span> <span>str</span><span>,</span> <span>lang</span><span>:</span> <span>str</span><span>)</span> <span>-></span> <span>str</span><span>:</span>
        <span>pass</span>
#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

Enter fullscreen mode Exit fullscreen mode

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. 

<span>#translate/adapters.py </span>
<span>import</span> <span>boto3</span>
<span>from</span> <span>uuid</span> <span>import</span> <span>uuid4</span>
<span>from</span> <span>datetime</span> <span>import</span> <span>datetime</span>
<span>from</span> <span>models</span> <span>import</span> <span>Record</span>
<span>class</span> <span>DynamoDBPersistenceAdapter</span><span>:</span>
    <span>"""</span><span> Implementation of TextPersistencePort using DynamoDB as storage. </span><span>"""</span>
    <span>def</span> <span>__init__</span><span>(</span><span>self</span><span>,</span> <span>table_name</span><span>:</span> <span>str</span><span>):</span>
        <span>dynamodb</span> <span>=</span> <span>boto3</span><span>.</span><span>resource</span><span>(</span><span>"</span><span>dynamodb</span><span>"</span><span>)</span>
        <span>self</span><span>.</span><span>table</span> <span>=</span> <span>dynamodb</span><span>.</span><span>Table</span><span>(</span><span>table_name</span><span>)</span>
    <span>def</span> <span>save</span><span>(</span><span>self</span><span>,</span> <span>input_text</span><span>,</span> <span>output_text</span><span>):</span>
        <span>"""</span><span> 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 </span><span>"""</span>
        <span>id</span> <span>=</span> <span>str</span><span>(</span><span>uuid4</span><span>())</span>
        <span>created_at</span> <span>=</span> <span>datetime</span><span>.</span><span>now</span><span>()</span>
        <span>self</span><span>.</span><span>table</span><span>.</span><span>put_item</span><span>(</span>
            <span>Item</span><span>=</span><span>{</span>
                <span>"</span><span>id</span><span>"</span><span>:</span> <span>str</span><span>(</span><span>uuid4</span><span>()),</span>
                <span>"</span><span>input_text</span><span>"</span><span>:</span> <span>input_text</span><span>,</span>
                <span>"</span><span>output_text</span><span>"</span><span>:</span> <span>output_text</span><span>,</span>
                <span>"</span><span>created_at</span><span>"</span><span>:</span> <span>str</span><span>(</span><span>created_at</span><span>),</span>
            <span>}</span>
        <span>)</span>
        <span>record</span> <span>=</span> <span>Record</span><span>(</span><span>id</span><span>,</span> <span>input_text</span><span>,</span> <span>output_text</span><span>,</span> <span>created_at</span><span>)</span>
        <span>return</span> <span>record</span>
<span>class</span> <span>AWSTranslateAdapter</span><span>:</span>
    <span>"""</span><span> Implementation of TranslationPort using DynamoDB as storage. </span><span>"""</span>
    <span>def</span> <span>__init__</span><span>(</span><span>self</span><span>):</span>
        <span>self</span><span>.</span><span>client</span> <span>=</span> <span>boto3</span><span>.</span><span>client</span><span>(</span><span>"</span><span>translate</span><span>"</span><span>)</span>
    <span>def</span> <span>translate</span><span>(</span><span>self</span><span>,</span> <span>text</span><span>,</span> <span>lang</span><span>):</span>
        <span>"""</span><span> Translate text to lang Args: text: The text to translate lang: The language code to translate to Returns: The translated string </span><span>"""</span>
        <span>result</span> <span>=</span> <span>self</span><span>.</span><span>client</span><span>.</span><span>translate_text</span><span>(</span>
            <span>Text</span><span>=</span><span>text</span><span>,</span>
            <span>SourceLanguageCode</span><span>=</span><span>"</span><span>auto</span><span>"</span><span>,</span>
            <span>TargetLanguageCode</span><span>=</span><span>lang</span><span>,</span>
        <span>)</span>
        <span>translated_text</span> <span>=</span> <span>result</span><span>[</span><span>"</span><span>TranslatedText</span><span>"</span><span>]</span>
        <span>return</span> <span>translated_text</span>
<span>#translate/adapters.py </span>
<span>import</span> <span>boto3</span>
<span>from</span> <span>uuid</span> <span>import</span> <span>uuid4</span>
<span>from</span> <span>datetime</span> <span>import</span> <span>datetime</span>

<span>from</span> <span>models</span> <span>import</span> <span>Record</span>


<span>class</span> <span>DynamoDBPersistenceAdapter</span><span>:</span>
    <span>"""</span><span> Implementation of TextPersistencePort using DynamoDB as storage. </span><span>"""</span>

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

    <span>def</span> <span>save</span><span>(</span><span>self</span><span>,</span> <span>input_text</span><span>,</span> <span>output_text</span><span>):</span>
        <span>"""</span><span> 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 </span><span>"""</span>

        <span>id</span> <span>=</span> <span>str</span><span>(</span><span>uuid4</span><span>())</span>
        <span>created_at</span> <span>=</span> <span>datetime</span><span>.</span><span>now</span><span>()</span>

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

        <span>record</span> <span>=</span> <span>Record</span><span>(</span><span>id</span><span>,</span> <span>input_text</span><span>,</span> <span>output_text</span><span>,</span> <span>created_at</span><span>)</span>
        <span>return</span> <span>record</span>


<span>class</span> <span>AWSTranslateAdapter</span><span>:</span>
    <span>"""</span><span> Implementation of TranslationPort using DynamoDB as storage. </span><span>"""</span>

    <span>def</span> <span>__init__</span><span>(</span><span>self</span><span>):</span>
        <span>self</span><span>.</span><span>client</span> <span>=</span> <span>boto3</span><span>.</span><span>client</span><span>(</span><span>"</span><span>translate</span><span>"</span><span>)</span>

    <span>def</span> <span>translate</span><span>(</span><span>self</span><span>,</span> <span>text</span><span>,</span> <span>lang</span><span>):</span>
        <span>"""</span><span> Translate text to lang Args: text: The text to translate lang: The language code to translate to Returns: The translated string </span><span>"""</span>

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

        <span>translated_text</span> <span>=</span> <span>result</span><span>[</span><span>"</span><span>TranslatedText</span><span>"</span><span>]</span>

        <span>return</span> <span>translated_text</span>
#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

Enter fullscreen mode Exit fullscreen mode

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 

<span>#translate/main.py </span>
<span>import</span> <span>os</span>
<span>import</span> <span>json</span>
<span>import</span> <span>logging</span>
<span>from</span> <span>dataclasses</span> <span>import</span> <span>dataclass</span>
<span>from</span> <span>typing</span> <span>import</span> <span>Dict</span><span>,</span> <span>Any</span>
<span>from</span> <span>ports</span> <span>import</span> <span>TextPersistencePort</span><span>,</span> <span>TranslationPort</span>
<span>from</span> <span>adapters</span> <span>import</span> <span>DynamoDBPersistenceAdapter</span><span>,</span> <span>AWSTranslateAdapter</span>
<span>logger</span> <span>=</span> <span>logging</span><span>.</span><span>getLogger</span><span>(</span><span>__name__</span><span>)</span>
<span>@dataclass</span>
<span>class</span> <span>TranslationRequest</span><span>:</span>
    <span>"""</span><span>Dataclass for translation requests</span><span>"""</span>
    <span>text</span><span>:</span> <span>str</span>
    <span>lang</span><span>:</span> <span>str</span>
    <span>@classmethod</span>
    <span>def</span> <span>from_dict</span><span>(</span><span>cls</span><span>,</span> <span>data</span><span>:</span> <span>Dict</span><span>[</span><span>str</span><span>,</span> <span>Any</span><span>])</span> <span>-></span> <span>"</span><span>TranslationRequest</span><span>"</span><span>:</span>
        <span>if</span> <span>not</span> <span>isinstance</span><span>(</span><span>data</span><span>[</span><span>"</span><span>text</span><span>"</span><span>],</span> <span>str</span><span>):</span>
            <span>raise</span> <span>ValueError</span><span>(</span><span>"</span><span>text must be a string</span><span>"</span><span>)</span>
        <span>if</span> <span>not</span> <span>isinstance</span><span>(</span><span>data</span><span>[</span><span>"</span><span>lang</span><span>"</span><span>],</span> <span>str</span><span>):</span>
            <span>raise</span> <span>ValueError</span><span>(</span><span>"</span><span>lang must be a string</span><span>"</span><span>)</span>
        <span>return</span> <span>cls</span><span>(</span><span>data</span><span>.</span><span>get</span><span>(</span><span>"</span><span>text</span><span>"</span><span>),</span> <span>data</span><span>.</span><span>get</span><span>(</span><span>"</span><span>lang</span><span>"</span><span>))</span>
<span>class</span> <span>Handler</span><span>:</span>
    <span>def</span> <span>__init__</span><span>(</span>
        <span>self</span><span>,</span>
        <span>text_port</span><span>:</span> <span>TextPersistencePort</span><span>,</span>
        <span>translate_port</span><span>:</span> <span>TranslationPort</span><span>,</span>
    <span>):</span>
        <span>self</span><span>.</span><span>text_port</span> <span>=</span> <span>text_port</span>
        <span>self</span><span>.</span><span>translate_port</span> <span>=</span> <span>translate_port</span>
    <span>def</span> <span>__call__</span><span>(</span><span>self</span><span>,</span> <span>request</span><span>,</span> <span>*</span><span>args</span><span>):</span>
        <span>"""</span><span> Process a translation request. Args: request: dict containing the request data Returns: dict with status code and response body </span><span>"""</span>
        <span>try</span><span>:</span>
            <span>body</span><span>:</span> <span>dict</span> <span>=</span> <span>json</span><span>.</span><span>loads</span><span>(</span><span>request</span><span>[</span><span>"</span><span>body</span><span>"</span><span>])</span>
            <span>request</span> <span>=</span> <span>TranslationRequest</span><span>.</span><span>from_dict</span><span>(</span><span>body</span><span>)</span>
        <span>except </span><span>(</span><span>json</span><span>.</span><span>JSONDecodeError</span><span>,</span> <span>KeyError</span><span>)</span> <span>as</span> <span>e</span><span>:</span>
            <span>logger</span><span>.</span><span>error</span><span>(</span><span>f</span><span>"</span><span>Invalid request: </span><span>{</span><span>str</span><span>(</span><span>e</span><span>)</span><span>}</span><span>"</span><span>)</span>
            <span>return</span> <span>self</span><span>.</span><span>_get_error_response</span><span>(</span><span>"</span><span>Invalid request</span><span>"</span><span>,</span> <span>status_code</span><span>=</span><span>400</span><span>)</span>
        <span>try</span><span>:</span>
            <span>result</span> <span>=</span> <span>self</span><span>.</span><span>translate_port</span><span>.</span><span>translate</span><span>(</span><span>request</span><span>.</span><span>text</span><span>,</span> <span>request</span><span>.</span><span>lang</span><span>)</span>
            <span>output</span> <span>=</span> <span>self</span><span>.</span><span>text_port</span><span>.</span><span>save</span><span>(</span><span>request</span><span>.</span><span>text</span><span>,</span> <span>result</span><span>)</span>
            <span>logger</span><span>.</span><span>info</span><span>(</span><span>f</span><span>"</span><span>Saved record with ID: </span><span>{</span><span>output</span><span>.</span><span>id</span><span>}</span><span>"</span><span>)</span>
            <span>return</span> <span>self</span><span>.</span><span>_get_success_response</span><span>(</span><span>result</span><span>)</span>
        <span>except</span> <span>Exception</span> <span>as</span> <span>e</span><span>:</span>
            <span>return</span> <span>self</span><span>.</span><span>_get_error_response</span><span>(</span><span>"</span><span>An error was encountered</span><span>"</span><span>,</span> <span>status_code</span><span>=</span><span>500</span><span>)</span>
    <span>def</span> <span>_get_success_response</span><span>(</span><span>self</span><span>,</span> <span>text</span><span>:</span> <span>str</span><span>):</span>
        <span>"""</span><span> Generate a successful response. Args: text: The translated text Returns: Dictionary with status code and response body </span><span>"""</span>
        <span>return</span> <span>{</span>
            <span>"</span><span>statusCode</span><span>"</span><span>:</span> <span>"</span><span>200</span><span>"</span><span>,</span> 
            <span>"</span><span>headers</span><span>"</span><span>:</span> <span>{</span>
                <span>"</span><span>Access-Control-Allow-Headers</span><span>"</span><span>:</span> <span>"</span><span>Content-Type</span><span>"</span><span>,</span>
                <span>"</span><span>Access-Control-Allow-Origin</span><span>"</span><span>:</span> <span>"</span><span>*</span><span>"</span><span>,</span>
                <span>"</span><span>Access-Control-Allow-Methods</span><span>"</span><span>:</span> <span>"</span><span>OPTIONS,POST,GET</span><span>"</span>
            <span>},</span>
            <span>"</span><span>body</span><span>"</span><span>:</span> <span>json</span><span>.</span><span>dumps</span><span>({</span><span>"</span><span>result</span><span>"</span><span>:</span> <span>text</span><span>})}</span>
    <span>def</span> <span>_get_error_response</span><span>(</span><span>self</span><span>,</span> <span>error</span><span>:</span> <span>str</span><span>,</span> <span>status_code</span><span>:</span> <span>int</span><span>):</span>
        <span>"""</span><span> Generate an error response. Args: error: The error message status_code: HTTP status code Returns: Dictionary with status code and response body </span><span>"""</span>
        <span>return</span> <span>{</span><span>"</span><span>statusCode</span><span>"</span><span>:</span> <span>str</span><span>(</span><span>status_code</span><span>),</span>
                <span>"</span><span>headers</span><span>"</span><span>:</span> <span>{</span>
                    <span>"</span><span>Access-Control-Allow-Headers</span><span>"</span><span>:</span> <span>"</span><span>Content-Type</span><span>"</span><span>,</span>
                    <span>"</span><span>Access-Control-Allow-Origin</span><span>"</span><span>:</span> <span>"</span><span>*</span><span>"</span><span>,</span>
                    <span>"</span><span>Access-Control-Allow-Methods</span><span>"</span><span>:</span> <span>"</span><span>OPTIONS,POST,GET</span><span>"</span>
                <span>},</span>
                <span>"</span><span>body</span><span>"</span><span>:</span> <span>json</span><span>.</span><span>dumps</span><span>({</span><span>"</span><span>detail</span><span>"</span><span>:</span> <span>error</span><span>})}</span>
<span>text_port</span> <span>=</span> <span>DynamoDBPersistenceAdapter</span><span>(</span><span>os</span><span>.</span><span>environ</span><span>.</span><span>get</span><span>(</span><span>"</span><span>DYNAMODB_TABLE</span><span>"</span><span>))</span>
<span>translate_port</span> <span>=</span> <span>AWSTranslateAdapter</span><span>()</span>
<span>handler</span> <span>=</span> <span>Handler</span><span>(</span><span>text_port</span><span>,</span> <span>translate_port</span><span>)</span>
<span>#translate/main.py </span>
<span>import</span> <span>os</span>
<span>import</span> <span>json</span>
<span>import</span> <span>logging</span>
<span>from</span> <span>dataclasses</span> <span>import</span> <span>dataclass</span>
<span>from</span> <span>typing</span> <span>import</span> <span>Dict</span><span>,</span> <span>Any</span>

<span>from</span> <span>ports</span> <span>import</span> <span>TextPersistencePort</span><span>,</span> <span>TranslationPort</span>
<span>from</span> <span>adapters</span> <span>import</span> <span>DynamoDBPersistenceAdapter</span><span>,</span> <span>AWSTranslateAdapter</span>

<span>logger</span> <span>=</span> <span>logging</span><span>.</span><span>getLogger</span><span>(</span><span>__name__</span><span>)</span>


<span>@dataclass</span>
<span>class</span> <span>TranslationRequest</span><span>:</span>
    <span>"""</span><span>Dataclass for translation requests</span><span>"""</span>

    <span>text</span><span>:</span> <span>str</span>
    <span>lang</span><span>:</span> <span>str</span>

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

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

        <span>return</span> <span>cls</span><span>(</span><span>data</span><span>.</span><span>get</span><span>(</span><span>"</span><span>text</span><span>"</span><span>),</span> <span>data</span><span>.</span><span>get</span><span>(</span><span>"</span><span>lang</span><span>"</span><span>))</span>


<span>class</span> <span>Handler</span><span>:</span>

    <span>def</span> <span>__init__</span><span>(</span>
        <span>self</span><span>,</span>
        <span>text_port</span><span>:</span> <span>TextPersistencePort</span><span>,</span>
        <span>translate_port</span><span>:</span> <span>TranslationPort</span><span>,</span>
    <span>):</span>

        <span>self</span><span>.</span><span>text_port</span> <span>=</span> <span>text_port</span>
        <span>self</span><span>.</span><span>translate_port</span> <span>=</span> <span>translate_port</span>

    <span>def</span> <span>__call__</span><span>(</span><span>self</span><span>,</span> <span>request</span><span>,</span> <span>*</span><span>args</span><span>):</span>
        <span>"""</span><span> Process a translation request. Args: request: dict containing the request data Returns: dict with status code and response body </span><span>"""</span>

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

        <span>try</span><span>:</span>
            <span>result</span> <span>=</span> <span>self</span><span>.</span><span>translate_port</span><span>.</span><span>translate</span><span>(</span><span>request</span><span>.</span><span>text</span><span>,</span> <span>request</span><span>.</span><span>lang</span><span>)</span>

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

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

    <span>def</span> <span>_get_success_response</span><span>(</span><span>self</span><span>,</span> <span>text</span><span>:</span> <span>str</span><span>):</span>
        <span>"""</span><span> Generate a successful response. Args: text: The translated text Returns: Dictionary with status code and response body </span><span>"""</span>

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

    <span>def</span> <span>_get_error_response</span><span>(</span><span>self</span><span>,</span> <span>error</span><span>:</span> <span>str</span><span>,</span> <span>status_code</span><span>:</span> <span>int</span><span>):</span>
        <span>"""</span><span> Generate an error response. Args: error: The error message status_code: HTTP status code Returns: Dictionary with status code and response body </span><span>"""</span>

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


<span>text_port</span> <span>=</span> <span>DynamoDBPersistenceAdapter</span><span>(</span><span>os</span><span>.</span><span>environ</span><span>.</span><span>get</span><span>(</span><span>"</span><span>DYNAMODB_TABLE</span><span>"</span><span>))</span>
<span>translate_port</span> <span>=</span> <span>AWSTranslateAdapter</span><span>()</span>

<span>handler</span> <span>=</span> <span>Handler</span><span>(</span><span>text_port</span><span>,</span> <span>translate_port</span><span>)</span>
#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)

Enter fullscreen mode Exit fullscreen mode

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 

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

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})}

Enter fullscreen mode Exit fullscreen mode

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

原文链接：Project Translate: The Translate API (Part 2)

展开阅读全文
© 版权声明
文章版权声明 1、本网站名称：拾光赋
2、本站永久网址：https://www.blogs.ink
3、本网站的文章部分内容可能来源于网络，仅供大家学习与参考，如有侵权，请联系站长QQ：805375623进行删除处理。
4、本站一切资源不代表本站立场，并不代表本站赞同其观点和对其真实性负责。
5、本站一律禁止以任何方式发布或转载任何违法的相关信息，访客发现请向站长举报
6、本站资源大多存储在云盘，如发现链接失效，请联系我们我们会第一时间更新。

THE END
Python（EN）
# python# aws# lambda
喜欢就支持一下吧
点赞8 分享
Do not let dream just be your dream.
别让梦想只停留在梦里
kity的头像-拾光赋
相关推荐
评论 抢沙发

请登录后发表评论

    暂无评论内容

kity的头像-拾光赋
今日剩余 97.4%
本周剩余 71.1%
本月剩余 19.3%
本年剩余 77.0%
最近评论
Vito的头像-拾光赋

Vito5个月前2

你就是我心中的那首忐忑，总是让我惊心动魄
Jason.Meng的头像-拾光赋钻石会员

Jason.Meng徽章-人气大使-拾光赋5个月前2

总是会在人生的不同阶段反复爱上大佬的文章~
每日一言
Sometimes a winner is just a dreamer that never gives up.
有时候，成功者只是坚持梦想不放弃的人