AIFO Code Snippets

README.md

Short description of the files

docker-watch.sh

The docker watcher: Restarts the docker container if the API file changes in the linux file system. Runs daemonized (screen) on the linux server within the user scope

Dockerfile

The Dockerfile used to build the API container.

main.py

Shows the handle_dialog_flow function. It is the main entrypoint for the API and is responsible for the intent-matching.
Based on the given intent a different action is being executed.

docker-watch.sh

unset IFS

inotifywait -m -e close_nowrite /docker/aifo/opt/app/main.py |
  while read dir op file
  do
    docker-compose -f /docker/aifo/opt/docker-compose.yaml stop && docker-compose -f /docker/aifo/opt/docker-compose.yaml up -d && echo "Restarted container"  && break

    ((step+=1))
  done

  echo "Processed $step events."

Dockerfile

FROM python:3.10

WORKDIR /code

COPY ./requirements.txt /code/requirements.txt

RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt

COPY ./app /code/app

CMD ["uvicorn", "app.main:app", "--proxy-headers", "--host", "0.0.0.0", "--port", "80"]

main.py

import logging
from typing import List, Optional

import mariadb
from fastapi import FastAPI, Body

config = {
    # removed - sensitive parameters
}
app = FastAPI()


# Models
########

# removed to save space


# Helper functions
##################

# removed to save space


# API endpoints
###############
@app.post("/")
def handle_dialog_flow(data=Body({})):
    # response template
    response = {
        'fulfillmentText': '',
        'fulfillmentMessages': [],
        'outputContexts': [],
        'sessionEntityTypes': []
    }

    try:
        # request data
        session_uuid = data['session'].split('/')[-1]  # just the uuid is important
        query = data['queryResult']
        intent = query['intent']
        t = config['translations'][query['languageCode']]  # translations

        if intent['displayName'] == 'module.list':
            modules = get_modules()
            text = t["module.list.modules_available"]
            text += '\n'.join(list(map(lambda m: m.name, modules)))
            response['fulfillmentText'] = text

        elif intent['displayName'] == 'user.setup - yes':
            if 'outputContexts' in query:
                params = context_parameters(query['outputContexts'])
                if 'email' in params:
                    student = Student(
                        None,
                        ' '.join(params['given-name']),
                        ' '.join(params['last-name']),
                        params['email']
                    )
                    student.id = save_student(student)
                    create_session(session_uuid, student)

                    # prepare response
                    response['fulfillmentText'] = t['user.setup.welcome_msg'].format(firstname=student.firstname,
                                                                                     id=student.id)

        elif intent['displayName'] == 'module.enroll':
            params = context_parameters(query['outputContexts'])
            response = enroll_module(get_student_from_session(session_uuid), params, response, t)

        elif intent['displayName'] == 'module.enrolled.personal':
            student = get_student_from_session(session_uuid)
            text = t['module.enrolled.personal.your_modules']
            text += '\n'.join(list(map(lambda module: module.name, student_enrolled_modules(student))))
            response['fulfillmentText'] = text

        elif intent['displayName'] == 'module.enrolled.public':
            student = get_student_by_email(query['parameters']['email'])
            text = t['module.enrolled.public.their_modules'].format(
                firstname=student.firstname,
                lastname=student.lastname,
                email=student.email
            )
            text += '\n'.join(list(map(lambda m: m.name, student_enrolled_modules(student))))
            response['fulfillmentText'] = text

    except Exception as e:
        response['fulfillmentText'] = f'API exception: "{str(e)}"'
        logging.getLogger('aifo').exception(e)

    return response