Initial commit for textract

moto/__init__.py

@@ -149,6 +149,7 @@ XRaySegment = lazy_load(".xray", "XRaySegment")
 mock_xray = lazy_load(".xray", "mock_xray")
 mock_xray_client = lazy_load(".xray", "mock_xray_client")
 mock_wafv2 = lazy_load(".wafv2", "mock_wafv2")
+mock_textract = lazy_load(".textract", "mock_textract")
 
 
 class MockAll(ContextDecorator):

moto/backend_index.py

@@ -148,6 +148,7 @@ backend_url_patterns = [
         re.compile("https?://ingest\\.timestream\\.(.+)\\.amazonaws\\.com/"),
     ),
     ("transcribe", re.compile("https?://transcribe\\.(.+)\\.amazonaws\\.com")),
+    ("textract", re.compile("https?://textract\\.(.+)\\.amazonaws\\.com")),
     ("wafv2", re.compile("https?://wafv2\\.(.+)\\.amazonaws.com")),
     ("xray", re.compile("https?://xray\\.(.+)\\.amazonaws.com")),
 ]

moto/textract/__init__.py

@@ -0,0 +1,5 @@
+"""textract module initialization; sets value for base decorator."""
+from .models import textract_backends
+from ..core.models import base_decorator
+
+mock_textract = base_decorator(textract_backends)

moto/textract/exceptions.py

@@ -0,0 +1,30 @@
+"""Exceptions raised by the textract service."""
+from moto.core.exceptions import JsonRESTError
+
+class InvalidJobIdException(JsonRESTError):
+    code = 400
+
+    def __init__(self):
+        super().__init__(
+            __class__.__name__,
+            "An invalid job identifier was passed.",
+        )
+
+class InvalidS3ObjectException(JsonRESTError):
+    code = 400
+
+    def __init__(self):
+        super().__init__(
+            __class__.__name__,
+            "Amazon Textract is unable to access the S3 object that's specified in the request.",
+        )
+
+class InvalidParameterException(JsonRESTError):
+    code = 400
+
+    def __init__(self):
+        super().__init__(
+            __class__.__name__,
+            "An input parameter violated a constraint. For example, in synchronous operations, an InvalidParameterException exception occurs when neither of the S3Object or Bytes values are supplied in the Document request parameter. Validate your parameter before calling the API operation again.",
+        )
+

moto/textract/models.py

@@ -0,0 +1,66 @@
+"""TextractBackend class with methods for supported APIs."""
+
+import uuid
+from random import randint
+from collections import defaultdict
+
+from moto.core import BaseBackend, BaseModel
+from moto.core.utils import BackendDict
+
+from .exceptions import InvalidParameterException, InvalidJobIdException
+
+class TextractJobStatus:
+    in_progress = "IN_PROGRESS"
+    succeeded = "SUCCEEDED"
+    failed = "FAILED"
+    partial_success = "PARTIAL_SUCCESS"
+
+class TextractJob(BaseModel):
+    def __init__(self, job):
+        self.job = job
+
+class TextractBackend(BaseBackend):
+    """Implementation of Textract APIs."""
+    JOB_STATUS = TextractJobStatus.succeeded
+    PAGES = {"Pages": randint(5, 500)}
+    BLOCKS = []
+
+    def __init__(self, region_name=None):
+        self.region_name = region_name
+        self.async_text_detection_jobs = defaultdict()
+
+    def reset(self):
+        """Re-initialize all attributes for this instance."""
+        region_name = self.region_name
+        self.async_text_detection_jobs = defaultdict()
+        self.__dict__ = {}
+        self.__init__(region_name)
+
+    def get_document_text_detection(self, job_id, max_results, next_token):
+        job = self.async_text_detection_jobs.get(job_id)
+        if not job:
+            raise InvalidJobIdException()
+        return job
+
+    def start_document_text_detection(
+        self,
+        document_location,
+        client_request_token,
+        job_tag,
+        notification_channel,
+        output_config,
+        kms_key_id,
+    ):
+        if not document_location:
+            raise InvalidParameterException()
+        job_id = uuid.uuid4()
+        self.async_text_detection_jobs[job_id] = TextractJob({
+            "Blocks": TextractBackend.BLOCKS,
+            "DetectDocumentTextModelVersion": "1.0",
+            "DocumentMetadata": {"Pages": TextractBackend.PAGES},
+            "JobStatus": TextractBackend.JOB_STATUS,
+        })
+        return job_id
+
+
+textract_backends = BackendDict(TextractBackend, "textract")

moto/textract/responses.py

@@ -0,0 +1,45 @@
+"""Handles incoming textract requests, invokes methods, returns responses."""
+import json
+
+from moto.core.responses import BaseResponse
+from .models import textract_backends
+
+
+class TextractResponse(BaseResponse):
+    """Handler for Textract requests and responses."""
+
+    @property
+    def textract_backend(self):
+        """Return backend instance specific for this region."""
+        return textract_backends[self.region]
+
+    def get_document_text_detection(self):
+        params = self._get_params()
+        job_id = params.get("JobId")
+        max_results = params.get("MaxResults")
+        next_token = params.get("NextToken")
+        job = self.textract_backend.get_document_text_detection(
+            job_id=job_id,
+            max_results=max_results,
+            next_token=next_token,
+        )
+        return json.dumps(job)
+
+    
+    def start_document_text_detection(self):
+        params = self._get_params()
+        document_location = params.get("DocumentLocation")
+        client_request_token = params.get("ClientRequestToken")
+        job_tag = params.get("JobTag")
+        notification_channel = params.get("NotificationChannel")
+        output_config = params.get("OutputConfig")
+        kms_key_id = params.get("KMSKeyId")
+        job_id = self.textract_backend.start_document_text_detection(
+            document_location=document_location,
+            client_request_token=client_request_token,
+            job_tag=job_tag,
+            notification_channel=notification_channel,
+            output_config=output_config,
+            kms_key_id=kms_key_id,
+        )
+        return json.dumps(dict(JobId=job_id))

moto/textract/urls.py

@@ -0,0 +1,12 @@
+"""textract base URL and path."""
+from .responses import TextractResponse
+
+url_bases = [
+    r"https?://textract\.(.+)\.amazonaws\.com",
+]
+
+
+
+url_paths = {
+    "{0}/$": TextractResponse.dispatch,
+}

tests/test_textract/test_server.py

@@ -0,0 +1,13 @@
+"""Test different server responses."""
+import sure  # noqa # pylint: disable=unused-import
+
+import moto.server as server
+
+
+def test_textract_list():
+    backend = server.create_backend_app("textract")
+    test_client = backend.test_client()
+
+    resp = test_client.get("/")
+    resp.status_code.should.equal(200)
+    str(resp.data).should.contain("?")

tests/test_textract/test_textract.py

@@ -0,0 +1,31 @@
+"""Unit tests for textract-supported APIs."""
+import boto3
+
+from moto import mock_textract
+
+# See our Development Tips on writing tests for hints on how to write good tests:
+# http://docs.getmoto.org/en/latest/docs/contributing/development_tips/tests.html
+
+
+
+# @mock_textract
+# def test_get_document_text_detection():
+#     client = boto3.client("textract", region_name="us-east-1")
+#     resp = client.get_document_text_detection()
+
+#     raise Exception("NotYetImplemented")
+
+
+@mock_textract
+def test_start_document_text_detection():
+    client = boto3.client("textract", region_name="us-east-2")
+    resp = client.start_document_text_detection(
+        DocumentLocation={
+            'S3Object': {
+                'Bucket': 'bucket',
+                'Name': 'name',
+            }
+        },
+    )
+
+    raise Exception("NotYetImplemented")