QuartService

class elg.QuartService(name: str, request_size_limit: Optional[int] = None)

Class to help the creation of an ELG compatible service from a python model using Quart. Extra dependencies need to be install to use the QuartService class. Please run: pip install elg[quart].

The QuartService class is suitable for services that execute the request directly, for example of a simple language detection service:

from elg import QuartService
from elg.model import AnnotationsResponse
import langdetect

class ELGService(QuartService):
    async def process_text(self, content):
        langs = langdetect.detect_langs(content.content)
        ld = {}
        for l in langs:
            ld[l.lang] = l.prob
        return AnnotationsResponse(features=ld)

service = ELGService("LangDetection")
app = service.app

The QuartService class is also particulary useful for proxy services that forward the request to the actual LT service. For example a proxy for a Speech-to-text service running outside the ELG cluster:

import traceback
import aiohttp

from loguru import logger

from elg import QuartService
from elg.model import TextsResponse
from elg.quart_service import ProcessingError


class Proxy(QuartService):

    consume_generator = False

    async def setup(self):
        self.session = aiohttp.ClientSession()

    async def shutdown(self):
        if self.session is not None:
            await self.session.close()

    async def process_audio(self, content):
        try:
            # Make the remote call
            async with self.session.post("https://example.com/endpoint", data=content.generator) as client_response:
                status_code = client_response.status
                content = await client_response.json()
        except:
            traceback.print_exc()
            raise ProcessingError.InternalError('Error calling API')

        if status_code >= 400:
            # if your API returns sensible error messages you could include that
            # instead of the generic message
            raise ProcessingError.InternalError('Error calling API')

        logger.info("Return the text response")
        return TextsResponse(texts=[{"content": content["text"]}])

service = Proxy("Proxy")
app = service.app
to_json(obj)

Hook that can be overridden by subclasses to customise JSON encoding.

FlaskService can convert the following types to JSON by default, in addition to the types handled natively by json.dump:

  • date, time, datetime (via .isoformat())

  • uuid.UUID (converted to str)

  • any pydantic.BaseModel including the ELG message types (via .dict(by_alias=True, exclude_none=True))

  • anything Iterable (as a list)

  • any dataclass (converted to a dict via dataclasses.asdict)

  • anything with a __json__ or for_json method (which is expected to return a serializable type)

To handle other types, or to change the standard behaviour for any of the above types, subclasses can override this method, which will be called whenever an object other than a string, number, bool, list or dict must be serialized and is expected to return a JSON-serializable object to be used in place of the original, or None to fall back to the default behaviour.

The default implementation of this method always returns None.

Parameters

obj – the object to convert

Returns

a substitute object suitable for JSON serialization, or None to use the default behaviour.

run()

Method to start the app.

async setup()

One-time setup tasks that must happen before the first request is handled. For example, it is possible to open an aiohttp ClientSessions() to use it : self.session = aiohttp.ClientSession()

async shutdown()

Logic that must run at shutdown time, after the last request has been handled. For example closing the aiohttp ClientSessions(): ``` if self.session is not None:

await self.session.close()

```

async process()

Main request processing logic - accepts a JSON request and returns a JSON response.

async process_request(request)

Method to process the request object. This method only calls the right process method regarding the type of the request.

async process_text(request: elg.model.request.TextRequest.TextRequest)

Method to implement if the service takes text as input. This method must be implemented as async.

Parameters

request (TextRequest) – TextRequest object.

async process_structured_text(request: elg.model.request.StructuredTextRequest.StructuredTextRequest)

Method to implement if the service takes structured text as input. This method must be implemented as async.

Parameters

request (StructuredTextRequest) – StructuredTextRequest object.

async process_audio(request: elg.model.request.AudioRequest.AudioRequest)

Method to implement if the service takes audio as input. This method must be implemented as async.

Parameters

request (AudioRequest) – AudioRequest object.

classmethod create_requirements(requirements: List = [], path: Optional[str] = None)

Class method to create the correct requirements.txt file.

Parameters
  • requirements (List, optional) – List of required pip packages. Defaults to [].

  • path (str, optional) – Path where to generate the file. Defaults to None.

classmethod create_docker_files(required_files: List[str] = [], required_folders: List[str] = [], commands: List[str] = [], base_image: str = 'python:slim', path: Optional[str] = None, log_level: str = 'INFO')

Class method to create the correct Dockerfile.

Parameters
  • required_files (List[str], optional) – List of files needed for the service. Defaults to [].

  • required_folders (List[str], optional) – List of folders needed for the service. Defaults to [].

  • commands (List[str], optional) – List off additional commands to run in the Dockerfile. Defaults to [].

  • base_image (str, optional) – Name of the base Docker image used in the Dockerfile. Defaults to ‘python:slim’.

  • path (str, optional) – Path where to generate the file. Defaults to None.

  • log_level (str, optional) – The minimum severity level from which logged messages should be displayed. Defaults to ‘INFO’.

classmethod docker_build_image(tag: str, pull: bool = True, path: Optional[str] = None, **kwargs)

Class method to do docker build … in python. Better to use the docker cli instead of this method.

classmethod docker_push_image(repository: str, tag: str, username: Optional[str] = None, password: Optional[str] = None, **kwargs)

Class method to do docker push … in python. Better to use the docker cli instead of this method.