Deploy a containerized python app to Azure (Part 1)

A container is a standard unit of software that packages up code and all its dependencies so the application runs quickly and reliably from one computing environment to another. Using Containers for application deployment ensures efficiency, scalability and cross-platform support.

Python is growing as a popular choice for backend services these days and containerization provides several benefits for Python applications, including portability, scalability, and consistency across different environments. By using containers, developers can ensure that their applications run the same way regardless of the infrastructure or third-party library updates.

This is a two-part series to explore creating docker container for python app in your local machine first, and then setting up continuous deployment for the same in Azure app service.

We assume that you already have some basic idea on Containers, Docker, Python, GitHub etc.

Considering real-world scenarios, I have created a Python FastAPI app which interacts with an Azure SQL database and exposes APIs which can be consumed in other services. We will then create a Linux container for this app along with its dependencies, and test the container locally.

If you want to skip to the continuous deployment configuration, head over to part 2 of this series.

Deploy a Containerized Python App to Azure (Part 2)

Pre-requisites

Local system

1. Python ≥ 3.10
2. Docker desktop
3. Postman (Optional, for testing APIs)
4. VS code & Docker extension in VS code (Optional)

Azure

1. An Azure subscription
2. Set up an Azure SQL Server and Database.

Create SQL username and password while creating the Server, we will use this to access the DB from our app.

In the networking section, make sure following option is checked Allow Azure Services and resources to access this server. This is required for our app service to access the database. You can add your own IP address too if you want to access this DB from your local machine.

We create a todolist database in this server for our app

(Optional) Use this script to initialize the DB schema

CREATE TABLE [dbo].[todolist](
 Id int not null identity(1,1) primary key,
 Task nvarchar(max) not null,
 CreatedOn DateTime not null,
 ModifiedOn DateTime,
 CreatedBy nvarchar(255),
 Category nvarchar(255),
 IsCompleted bit not null
)

Code

Now we start with the Python application. We’ll use FastAPI framework to develop our APIs which will read and write data in Azure SQL database using pyodbc library.

Following is the structure of the solution

mypythonwebapp (root folder)
  |--src
      |--main.py
      |--repository.py
      |--requirements.txt
  |--.env
  |--README.MD

1. Populate .env with following information

SQL_SERVER="<your sql server name>"
SQL_DB="<DB name>"
SQL_USERNAME="<sql username>"
SQL_PWD="{sql password}"
APPLICATIONINSIGHTS_CONNECTION_STRING="<App insights connection string>"

2. Add following in the requirements.txt file and install using pip

python-dotenv==1.0.0
uvicorn==0.23.2
fastapi==0.101.0
azure-monitor-opentelemetry==1.0.0b16
azure-identity==1.14.0
pyodbc==4.0.39

pip install -r requirements.txt

3. Start with our repository.py script to initialize the CRUD operations

Creating functions for 4 operations, which will be utilized by our Python APIs.

1. Get all tasks in ToDo list
2. Create a new task in ToDo list
3. Get tasks created by a user in ToDo list
4. Mark a Task as ‘Completed’ in ToDo list

import pyodbc
import logging
import os
import datetime
from dotenv import load_dotenv

class Repository():
    def __init__(self, logger):
        print('Repository created')
        self.logger = logger

    # Initialize connection
    async def get_conn(self):
        try:
            load_dotenv()
            driver='{ODBC Driver 17 for SQL Server}'
            connection = pyodbc.connect(f'DRIVER={driver};SERVER=tcp:'+os.getenv("SQL_SERVER")+\
                                        '.database.windows.net;PORT=1433;DATABASE='+os.getenv("SQL_DB")+\
                                        ';UID='+os.getenv("SQL_USERNAME")+';PWD='+ os.getenv("SQL_PWD"))
            return connection
        except Exception as e:
            self.logger.error(e)
            raise e

    async def get_all(self) -> list|None:
        print('get_all called')
        try:
            # Get all items from the database
            connection = await self.get_conn()
            cursor = connection.cursor()
            cursor.execute("SELECT Id, Task, CreatedOn, CreatedBy, Category, IsCompleted, ModifiedOn FROM todolist")
            rows = cursor.fetchall()
            items = []
            for row in rows:
                items.append({
                    "id": row.Id,
                    "task": row.Task,
                    "createdOn": row.CreatedOn,
                    "createdBy": row.CreatedBy,
                    "category": row.Category,
                    "isCompleted": row.IsCompleted,
                    "modifiedOn": row.ModifiedOn
                })
            self.logger.info("All tasks retrieved successfully")
            return items
        except Exception as e:
            self.logger.error(e)
            raise e
        finally:
            cursor.close()
            connection.close()

    async def get_by_user(self, created_by_user: str) -> list|None:
        try:
        # Get all items from the database
            connection = await self.get_conn()
            cursor = connection.cursor()
            cursor.execute(f"SELECT Id, Task, CreatedOn, CreatedBy, Category, IsCompleted, ModifiedOn FROM todolist WHERE CreatedBy = {created_by_user}", created_by_user)
            rows = cursor.fetchall()
            items = []
            for row in rows:
                items.append({
                    "id": row.Id,
                    "task": row.Task,
                    "createdOn": row.CreatedOn,
                    "createdBy": row.CreatedBy,
                    "category": row.Category,
                    "isCompleted": row.IsCompleted,
                    "modifiedOn": row.ModifiedOn
                })
            self.logger.info(f"Tasks for user {created_by_user} retrieved successfully")
            return items
        except Exception as e:
            self.logger.error(e)
            raise e
        finally:
            cursor.close()
            connection.close()

    async def add_task(self, description: str, user: str, category: str) -> int:
        # Add a new item to the database
        try:
            INSERT_STATEMENT = "INSERT INTO todolist (Task, CreatedBy, CreatedOn, Category, IsCompleted) OUTPUT INSERTED.Id VALUES (?, ?, ?, ?, 0)"
            connection = await self.get_conn()
            cursor = connection.cursor()
            cursor.execute(INSERT_STATEMENT,
                            description,
                            user,
                            datetime.datetime.now(),
                            category)
            taskId = cursor.fetchval()
            connection.commit()
            self.logger.info(f"Task with {taskId} for user {user} added successfully")
            return taskId
        except Exception as e:
            self.logger.error(e)
            raise e
        finally:
            cursor.close()
            connection.close()

    async def update_task_status(self, task_id: int) -> None:
        # Update an item in the database
        try:
            UPDATE_STATEMENT = "UPDATE todolist SET IsCompleted = 1, ModifiedOn = ? WHERE Id = ?"
            connection = await self.get_conn()
            cursor = connection.cursor()
            cursor.execute(UPDATE_STATEMENT, datetime.datetime.now(), task_id)
            connection.commit()
            self.logger.info(f"Task with {task_id} updated successfully")
        except Exception as e:
            self.logger.error(e)
            raise e
        finally:
            cursor.close()
            connection.close()

4. Next, initialize APIs in main.py and call services defined in repository.py

from dotenv import load_dotenv
import os
from fastapi import FastAPI
from fastapi import Request, Response
from pydantic import BaseModel
from repository import Repository
from azure.monitor.opentelemetry import configure_azure_monitor
from opentelemetry import trace
from opentelemetry.trace import (
    SpanKind,
    get_tracer_provider,
)
from opentelemetry.propagate import extract
from logging import getLogger, INFO

load_dotenv()

# Initialize OpenTelemetry with Azure Monitor
configure_azure_monitor(
    connection_string=os.getenv("APPLICATIONINSIGHTS_CONNECTION_STRING")
)
tracer = trace.get_tracer(__name__, tracer_provider=get_tracer_provider())
logger = getLogger(__name__)
logger.setLevel(INFO)

# This schema is needed for POST request - /create endpoint
class Task(BaseModel):
    description: str
    user: str
    category: str

# Define the app
app = FastAPI(
    title="MyApp",
    description="Hello API developer!",
    version="0.1.0"
)

# Initialize repository
repository = Repository(logger)

# API for health check of App
@app.get("/")
async def main(request: Request):
    with tracer.start_as_current_span(
        "main_request",
        context=extract(request.headers),
        kind=SpanKind.SERVER
    ):
        logger.info("Hello World endpoint was reached. . .")
        return {"message": "Hello World"}

# API to get all the data from ToDo list
@app.get("/all")
async def main(request: Request):
    with tracer.start_as_current_span("get_all_tasks",
                                      context=extract(request.headers),
                                      kind=SpanKind.SERVER):
        logger.info("Get all tasks endpoint was reached. . . validating the scope of request. . .")
        try:
            print('get_all_tasks endpoint was reached. . .')
            return repository.get_all()
        except Exception as ex:
            logger.error(ex)
            return Response(content=str(ex), status_code=500)


# API to submit data to ToDo list
@app.post("/create")
async def submit(task: Task, request: Request):
    with tracer.start_as_current_span("create_task",
                                      context=extract(request.headers),
                                      kind=SpanKind.SERVER):
        logger.info("/create endpoint was reached. . .")
        try:
            repository.add_task(task.description, task.user, task.category)
            return {"message": f"Data submitted successfully"}
        except Exception as ex:
            logger.error(ex)
            return Response(content=str(ex), status_code=500)


# API to mark a task as 'Completed'
@app.post("/done")
async def complete(task_id: int, request: Request):
    with tracer.start_as_current_span("mark_task_as_completed",
                                      context=extract(request.headers),
                                      kind=SpanKind.SERVER):
        logger.info("/done endpoint was reached. . .")
        try:
            await repository.update_task_status(task_id)
            return {"message": f"Task marked as completed successfully"}
        except Exception as ex:
            logger.error(ex)
            return Response(content=str(ex), status_code=500)

Now we do a quick test by running this app locally. Open terminal in src folder and run the following command:

uvicorn main:app --reload

The output in terminal should be like this

INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: Started reloader process [24392] using WatchFiles
Repository created
INFO: Started server process [24416]
INFO: Waiting for application startup.
INFO: Application startup complete.

Note: Your IP address should be added to Azure SQL Server in order to test the APIs. Azure SQL server disables public access by default.

1. Test /create endpoint

2. Test /all endpoint

3. Test /done endpoint

Now moving on to the most-awaited part — CREATING THE CONTAINER! :)

#STEP 1 — Set up Dockerfile

Start by creating a Dockerfile in the root folder with following code snippet

# parent image
FROM python:3.10.9-slim-buster

WORKDIR /app

ADD /src/requirements.txt .
ADD /src/repository.py .
ADD /src/main.py .
ADD .env .

RUN apt-get update -y && apt-get install -y curl gnupg g++ unixodbc-dev
RUN curl https://packages.microsoft.com/keys/microsoft.asc | apt-key add -
RUN curl https://packages.microsoft.com/config/debian/10/prod.list > /etc/apt/sources.list.d/mssql-release.list 

RUN exit
RUN apt-get update
RUN ACCEPT_EULA=Y apt-get install -y msodbcsql17

COPY /odbc.ini / 
RUN odbcinst -i -s -f /odbc.ini -l
RUN cat /etc/odbc.ini 

RUN pip install --upgrade -r requirements.txt

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Notice that there are new installations which we didn’t need when developing the app. Let’s walk through these updates.

We’re targeting this container to be running in Linux platform. pyodbc connects our app to Database using ODBC driver. unixODBC is a driver manager for Linux/Unix-based platform and it is required by pyodbc. Since pyodbc does not contain copy of unixODBC , hence it needs to be installed separately through Dockerfile. [Reference]

Next, create odbc.ini file which in your same directory as Dockerfile (root directory in this solution) and add following code. This is where Microsoft’s installer for the ODBC driver registers itself. [Reference]

[ODBC Driver 17 for SQL Server]
Description=Microsoft ODBC Driver 17 for SQL Server
Driver=/opt/microsoft/msodbcsql17/lib64/libmsodbcsql-17.4.so.1.1
UsageCount=1

Notice that we use the first line as our driver in code.

Next is to create a “System DSN”, which happens using following commands in Dockerfile

RUN odbcinst -i -s -f /odbc.ini -l
RUN cat /etc/odbc.ini

The final solution structure now looks like

mypythonwebapp (root folder)
  |--src
      |--main.py
      |--repository.py
      |--requirements.txt
  |--.env
  |--Dockerfile
  |--README.MD
  |--odbc.ini

Note: If you use Ubuntu platform, you can use FreeTDS ODBC driver. Dockerfile needs to be updated for the same.

#STEP 2 — Create image and container

Start the docker desktop in background. Open a terminal in the root folder where Dockerfile is located and run this command to build the image for our app

docker build -t todolist-app .

Once image builds successfully, you can see it in VS code extension. Alternatively, it will show up in Docker desktop too.

Create and run the container. Since container runs in isolation and our uvicorn worker is running inside the container, we need to publish a port from the container, in order to test the APIs.

docker run -d --name mycontainer -p 8000:8000 todolist-app

Open https://localhost:8000/all in browser and verify if the tasks are getting populated. Alternatively, test the APIs from Postman by updating the base address to localhost:8000.

If you face any issue in running the container or with the APIs, see the logs by doing this —

This concludes our development and testing for Python app and docker container. In part 2 article, we will configure continuous deployment to Azure app service.

References:

1. https://stackoverflow.com/questions/76296702/what-is-the-correct-dockerfile-to-install-pyodbc-on-python3-10-9-slim-buster-im
2. https://fastapi.tiangolo.com/deployment/docker/

Out now! Check out part 2 here

Deploy a Containerized Python App to Azure (Part 2)

Hope this article helped in your Python & Azure journey. Please leave a comment if you have any feedback for me.