Create a Python compute module

This tutorial provides a step-by-step walkthrough of how to create a basic compute module that runs Python.

Prerequisites

• Must have a Docker client installed
• Must have Python installed

Writing the compute module

1. Begin by creating a new directory for your compute module.
2. Create a file called Dockerfile in the directory.
3. Copy and paste the following into the Dockerfile:

# Change the platform based on your Foundry resource queue
FROM --platform=linux/amd64 python:latest

COPY requirements.txt .
RUN pip install -r requirements.txt
COPY src .

# This is required for running Compute Modules in Foundry
USER 5000
CMD ["python", "app.py"]

4. Create a new file called requirements.txt. This file specifies our dependencies for our Python application. Copy and paste the following into the file:

requests == 2.31.0

5. Create a new subdirectory called src. This is where we will put our Python app.
6. Inside the src directory, create a file called app.py.
7. Your directory should now look like this:

MyComputeModule
|- Dockerfile
|- requirements.txt
|- src
   |- app.py

8. Inside of app.py copy and paste the following code:

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
import requests
import os
import time
import logging as log
import json

log.basicConfig(level=log.INFO)

certPath = os.environ['CONNECTIONS_TO_OTHER_PODS_CA_PATH']

with open(os.environ["MODULE_AUTH_TOKEN"], 'r') as f:
    moduleAuthToken = f.read()

getJobUri = "https://localhost:8945/interactive-module/api/internal-query/job"
postResultUri = "https://localhost:8945/interactive-module/api/internal-query/results"


# Gets a job from the runtime. Jobs are only present when
# the status code is 200. If status code 204 is returned try again.
# This endpoint has long-polling enabled, and may be called without delay.
def getJobBlocking():
    while True:
        response = requests.get(
                getJobUri,
                headers={"Module-Auth-Token": moduleAuthToken},
                verify=certPath)
        if response.status_code == 200:
            return response.json()
        elif response.status_code == 204:
            log.info("No job found, trying again!")

# Process the query based on type
def get_result(query_type, query):
    if query_type == "multiply":
        return float(query) * 2
    elif query_type == "divide":
        return float(query) / 2
    else:
        log.info(f"Unknown query type: {query_type}")


# Posts job results to the runtime. All jobs received must have a result posted,
# otherwise new jobs may not be routed to this worker.
def postResult(jobId, result):
    response = requests.post(
            f"{postResultUri}/{jobId}",
            data=json.dumps(result).encode('utf-8'),
            headers={"Module-Auth-Token": moduleAuthToken, "Content-Type": "application/octet-stream"},
            verify=certPath)
    if response.status_code != 204:
        log.info(f"Failed to post result: {response.status_code}")


# Try forever
while True:
    try:
        job = getJobBlocking()
        v1 = job["computeModuleJobV1"]
        job_id = v1['jobId']
        query_type = v1['queryType']
        query = v1['query']
        result = get_result(query_type, query)
        postResult(job_id, result)
    except Exception as e:
        log.info(f"Something failed {str(e)}")
        time.sleep(1)

9. Run docker build -t mycomputemodule:0.0.1 . to create your Docker image called mycomputemodule.

You have now successfully created a compute module that can run in Foundry.

To run your container in Foundry, proceed to the tutorial on how to deploy a compute module.