In this guide, we will use CGI (Common Gateway Interface) Scripts to serve a simple python crud API. We will be dockerizing our code and using docker compose to test the application with mariadb.
The code
This is the simple crud python code that we are trying to run in docker with apache using cgi
#!/usr/bin/python3
import json
import os
import pymysql
DB_USER = os.getenv("DB_USER") or "tasksapi"
DB_PSWD = os.getenv("DB_PSWD") or "tasksapi"
DB_NAME = os.getenv("DB_NAME") or "tasksapi"
DB_HOST = os.getenv("DB_HOST") or "mariadb"
db = pymysql.connect(
host=DB_HOST,
user=DB_USER,
password=DB_PSWD,
database=DB_NAME,
cursorclass=pymysql.cursors.DictCursor,
)
cursor = db.cursor()
# Create a new task
def create_task(title):
try:
cursor.execute("INSERT INTO tasks (title) VALUES (%s)", title)
db.commit()
cursor.execute("SELECT MAX(id) AS id FROM tasks")
row = cursor.fetchone()
resp = get_task(row["id"])
return (resp[0], 201)
except Exception as e:
return (str(e), 500)
# Get all tasks
def get_tasks():
try:
cursor.execute(
"SELECT id, title, date_format(created, '%Y-%m-%d %H:%i') as created FROM tasks"
)
return (cursor.fetchall(), 200)
except Exception as e:
return (str(e), 500)
# Get an individual task
def get_task(id):
try:
cursor.execute(
"SELECT id, title, date_format(created, '%Y-%m-%d %H:%i') as created \
FROM tasks WHERE id="
+ str(id)
)
row = cursor.fetchone()
return (row if row is not None else "", 200 if row is not None else 404)
except Exception as e:
return ("", 404)
# Update an existing task
def update_task(id, title):
try:
cursor.execute("UPDATE tasks SET title=%s WHERE id=%s", (title, id))
db.commit()
return get_task(id)
except Exception as e:
return (str(e), 500)
# Delete an existing task
def delete_task(id):
try:
resp = get_task(id)
if resp[1] == 200:
cursor.execute("DELETE FROM tasks WHERE id=%s", id)
db.commit()
return ("", 200)
else:
return resp
except Exception as e:
return (str(e), 500)
# Returns the HTTP request method
def get_method():
return os.getenv("REQUEST_METHOD") or "GET"
# Returns the query string
def get_query_string():
query_string = os.getenv("QUERY_STRING") or ""
return query_string.replace("%20", " ").replace("%2F", "/").replace("+", " ")
# Returns the task ID if set in the request query string
def get_task_id():
query_string = get_query_string()
qs_parts = query_string.split("/")
return qs_parts[0] if qs_parts[0].isnumeric() else None
# Returns the task title from the query string if set
def get_task_title():
title = None
query_string = get_query_string()
if query_string != "":
qs_parts = query_string.split("/")
title = qs_parts[1] if len(qs_parts) > 1 else qs_parts[0]
title = None if title.isnumeric() else title
return title
# Returns True if title is valid, False otherwise
def title_is_valid(title):
return (
True
if isinstance(title, str) and len(title) >= 6 and len(title) <= 255
else False
)
# Returns a status code method
def get_status_msg(code):
msg = "OK"
msg = "Created" if code == 201 else msg
msg = "Not Found" if code == 404 else msg
msg = "Bad Request" if code == 400 else msg
msg = "Internal Server Error" if code == 500 else msg
return msg
method = get_method()
id = get_task_id()
title = get_task_title()
if method == "GET" and not id is None:
resp = get_task(id)
elif method == "GET":
resp = get_tasks()
elif method == "DELETE":
resp = delete_task(id)
elif not title_is_valid(title):
resp = ("", 400)
elif method == "POST":
resp = create_task(title)
elif method == "PUT":
resp = update_task(id, title)
print("Status: %d %s" % (resp[1], get_status_msg(resp[1])))
print("Content-type: application/json\n")
print(json.dumps(resp[0]))
If you would want a simple script instead to test that this works, use this
Save this as test-db.py
#!/usr/bin/python3
import os
import pymysql
# Database connection parameters - update as needed
DB_USER = os.getenv("DB_USER") or "root"
DB_PSWD = os.getenv("DB_PSWD") or None
DB_HOST = os.getenv("DB_HOST") or "localhost"
db = pymysql.connect(
host=DB_HOST,
user=DB_USER,
password=DB_PSWD,
database="mysql",
cursorclass=pymysql.cursors.DictCursor,
)
cursor = db.cursor()
cursor.execute("SHOW DATABASES")
print("Content-type: text/html\n")
print("Setup Successful")
The Dockerfile
We will use this Dockerfile to package our application so we can serve it anywhere.
FROM httpd:2.4-alpine
WORKDIR /usr/local/apache2/cgi-bin/
RUN apk update && \
apk add python3 python3-dev py3-pip gcc g++ libffi-dev && \
rm /var/cache/apk/*
COPY requirements.txt .
RUN pip3 install -U pip --break-system-packages && \
pip3 install -r requirements.txt --break-system-packages
COPY tasks-api.py .
RUN chmod +x tasks-api.py
CMD httpd-foreground -c "LoadModule cgid_module modules/mod_cgid.so"
Here, we are using httpd:2.4-alpine as our base image then adding python dependencies to make the environment work for our script.
This is the contents of the requirements.txt file
pymysql
cryptography
The Docker compose logic to put together the code
This is my docker compose compose.yml file
version: "3.9"
services:
tasksapi:
image: ektowett/tasksapi:latest
working_dir: /usr/src/app
build:
dockerfile: Dockerfile
context: .
ports:
- 8080:80
depends_on:
- mariadb
environment:
- DB_USER=tasksapi
- DB_PSWD=tasksapi
- DB_NAME=tasksapi
- DB_HOST=mariadb
networks:
- tasksapi_net
mariadb:
image: mariadb:11
ports:
- 3306:3306
volumes:
- tasksapi_db:/var/lib/mysql
environment:
- MYSQL_ROOT_PASSWORD=tasksapi
- MYSQL_DATABASE=tasksapi
- MYSQL_USER=tasksapi
- MYSQL_PASSWORD=tasksapi
networks:
- tasksapi_net
volumes:
tasksapi_db:
networks:
tasksapi_net:
To run the above, use
docker-compose up -d
This will build our image and download mariadb image if it doesn’t exist.
Creating the database
Exec into the db instance
docker-compose exec mariadb bash
Then login to db
root@bd7b8ceb1225:/# mariadb -u tasksapi -ptasksapi
Welcome to the MariaDB monitor. Commands end with ; or \g.
Your MariaDB connection id is 4
Server version: 11.3.2-MariaDB-1:11.3.2+maria~ubu2204 mariadb.org binary distribution
Copyright (c) 2000, 2018, Oracle, MariaDB Corporation Ab and others.
Type 'help;' or '\h' for help. Type '\c' to clear the current input statement.
MariaDB [(none)]>
Create table
MariaDB [(none)]> show databases;
+--------------------+
| Database |
+--------------------+
| information_schema |
| tasksapi |
+--------------------+
2 rows in set (0.001 sec)
MariaDB [(none)]> use tasksapi;
Database changed
MariaDB [tasksapi]> CREATE TABLE tasks (
-> id INT NOT NULL AUTO_INCREMENT,
-> title VARCHAR(255) NOT NULL,
-> created TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
-> PRIMARY KEY (id),
-> INDEX(created)
-> );
Query OK, 0 rows affected (0.012 sec)
MariaDB [tasksapi]> show tables;
+--------------------+
| Tables_in_tasksapi |
+--------------------+
| tasks |
+--------------------+
1 row in set (0.001 sec)
Testing our code
Get all tasks
$ curl http://127.0.0.1:8080/cgi-bin/tasks-api.py
[]
Add Task
$ curl -X POST -s -D - http://127.0.0.1:8080/cgi-bin/tasks-api.py\?First+Task
HTTP/1.1 201 Created
Date: Wed, 13 Mar 2024 16:05:51 GMT
Server: Apache/2.4.58 (Unix)
Transfer-Encoding: chunked
Content-Type: application/json
{"id": 1, "title": "First Task", "created": "2024-03-13 16:05"}
$ curl -X POST -s -D - http://127.0.0.1:8080/cgi-bin/tasks-api.py?Second+Task
HTTP/1.1 201 Created
Date: Wed, 13 Mar 2024 16:06:19 GMT
Server: Apache/2.4.58 (Unix)
Transfer-Encoding: chunked
Content-Type: application/json
{"id": 2, "title": "Second Task", "created": "2024-03-13 16:06"}
Get Tasks
$ curl http://127.0.0.1:8080/cgi-bin/tasks-api.py
[{"id": 1, "title": "First Task", "created": "2024-03-13 16:05"}, {"id": 2, "title": "Second Task", "created": "2024-03-13 16:06"}]
Get task
$ curl http://127.0.0.1:8080/cgi-bin/tasks-api.py\?1
{"id": 1, "title": "First Task", "created": "2024-03-13 16:05"}