DEV Community: Alexander Kammerer

TIL: How to reset all files in one folder in git

Alexander Kammerer — Wed, 23 Jul 2025 12:09:17 +0000

We use snapshot testing at work quite heavily because it enables us to test the output of complex functions that do a lot of things at the same time and produce a lot of output.

During a PR, I often accumulate a lot of files in one folder and when the PR is done, I want to clean up this folder, reset it to the state of the development branch, and re-create just the files I need.

To reset a whole folder to the state of this folder on another branch in git, you can do:

git restore --source=develop --staged --worktree tests/data

This resets the tests/data folder to the state of that folder on the develop branch.

Copy Markdown to Teams

Alexander Kammerer — Sat, 19 Jul 2025 21:11:22 +0000

I write all my notes at work and also my private notes in Obsidian. I find that it works perfect for my personal workflow and helps me be more organized. When I work on a new project, I like to write out a detailed proposal that highlights a possible plan, any issues that might occur, and any open questions.

I find myself copying that text over to Teams so that my colleagues also have access to these notes. But when I copy the markdown that I have written in Obsidian, the formatting is not correctly copied over.

While Teams does support some markdown (you can use **something** to make some text bold) it does not work well when pasting markdown.

Other people have noticed this issue as well and there are some workarounds but none that really work for me. I just want to copy the markdown and paste it into Teams.

The best workaround I found was pasting the markdown into an editor that supports a rich text preview and the copying the rich text preview. This is quite annoying and I do not want to use any online tool because I do not want to paste possibly sensitive text into a website.

The Plan: Markdown to HTML to Teams

After some research, I found the following workflow that seems to work rather consistently and produce well formatted text:

Copy the markdown from Obsidian.
Translate the markdown into HTML using the markdown package in Python.
Copying and pasting the HTML into Teams.

Teams knows how turn HTML into formatted text.

Nushell

To make this more convenient, I turned this into some shell script that takes the current clipboard content and converts it into html.

I use uv to invoke the CLI of the markdown package in Python. I then copy the output back into the clipboard.

Just put the following function into your nu config (config nu)

# convert markdown to html for Teams / other Microsoft products
def md2html [] {
    let markdown = (powershell -c "Get-Clipboard -Raw")
    let html = ($markdown | uvx --from markdown --with markdown python -m markdown -e utf8)
    $html | powershell -c "Set-Clipboard -AsHtml -Value $input"
}

Powershell

You can do the same thing in Powershell. Just put the following function into your powershell profile Microsoft.PowerShell_profile.ps1.

Function md2html {
    $OutputEncoding = [System.Text.Encoding]::UTF8
    $markdown = Get-Clipboard -Raw
    $html = $markdown | uvx --from markdown --with markdown python -m markdown -e utf8
    $html | Set-Clipboard -AsHtml
}

Creating a Shortcut (on Windows)

Right now, you will have to open a terminal and run md2html whenever you want to convert something, which is not the best UX.

On Windows, you can create a shortcut and assign a keyboard shortcut to it:

Create a new Shortcut: Right-click on your Desktop (or in any other folder) and select New > Shortcut.
Set the Target: In the "Type the location of the item" field, paste the following command. This command tells Windows to run PowerShell, hide its window, and then execute the function (which has to be in your powershell profile): powershell.exe -WindowStyle Hidden md2html.
Name the Shortcut: Give it a clear name, like "Convert Markdown to HTML", and click Finish.
Assign a Hotkey:
- Right-click on your newly created shortcut and select Properties.
- Look for the Shortcut field. Click into the field.
- Press the key combination you want to use, for example, Ctrl + Alt + M. Windows will automatically register it.
- Click OK to save.

Your hotkey is now active system-wide. You can try it out and for a short moment, a Powershell console should open.

The New and Improved Usage

With the hotkey is active, the process is more user friendly:

Copy the Markdown notes from Obsidian (Ctrl + C).
Press your new global hotkey (Ctrl + Alt + M or whatever you chose).
Click into the Teams message box and paste (Ctrl + V). The result should be perfectly formatted text in your Teams chat or channel post, without ever having to manually open a terminal or break your focus.

Clean up your Azure Container Registry

Alexander Kammerer — Fri, 04 Apr 2025 17:03:54 +0000

Azure Container Registry is one way to privately publish docker images and use these in your Azure resources.

You might observe that over time, your used storage goes up and up even though you store just a few images. This is due to some dangling image layers that are no longer in use but still count against your storage limit.

To get rid of these, I use this small powershell script:

$registry = "name-of-your-registry"
$repositories = az acr repository list --name $registry --output tsv

$manifestsToDelete = $repositories | ForEach-Object -Parallel {
    Write-Host "Check ${_}"
    $manifests = (az acr repository show-manifests --name ${using:registry} --repository $_ --query "[?tags[0]==null].digest" -o tsv) -split "`n"
    foreach ($manifest in $manifests -split "`n") { "${_}@${manifest}" }
} -ThrottleLimit 10

Write-Host "Total manifests to delete: $($manifestsToDelete.Count)"

$manifestsToDelete | ForEach-Object -Parallel {
    az acr repository delete --name ${using:registry} --image $_ --yes
} -ThrottleLimit 20

Just fill in the name of your registry. The script will then list the repositories (your images) and find the manifests to delete.

Finally, it will delete the manifests (in parallel, to speed up the process).

Converting Plotly charts into images in parallel

Alexander Kammerer — Thu, 02 Jan 2025 13:24:39 +0000

We use Plotly charts extensively in the company I work for. They make it easy to create interactive graphics that look good. The Python experience via the Plotly Express library is great and the bar to get started is low.

We have two main use-cases for Plotly graphs:

For our interactive dashboards with Plotly Dash. The integration of Plotly charts into Dash is obviously great.
For our PDF reports where we convert the chart into an image before rendering the PDF.

For a typical PDF report, we use 5-20 figures to show the evolution of a particular metric over time, the distribution of some value over a number of categories, or the comparison of different categories next to each other.

To create our PDF reports, we use a combination of Weasyprint, Jinja, and Plotly charts. To render a report as a PDF, we first have to render all graphs as images.

Rendering graphs with Kaleido

To do so, we use the great Kaleido package. It uses a Chrome browser to render the graph and save it as an image. The API is straight forward to use.

from kaleido.scopes.plotly import PlotlyScope

scope = PlotlyScope()
img_bytes = scope.transform(
    figure=figure, format="png", width=1000, height=1000, scale=4,
)

This renders the figure in figure as an image with 1000px height and width and a rendering scale of 4 (i.e. the image actually has dimensions 4000px x 4000px). The higher the scale, the more DPI the final image has, the better it looks, and the larger the final PDF is.

Rendering a lot of graphs

Rendering graphs takes a little bit of time and if you render a lot of them (10-20), it will be a significant share of your program's runtime. To speed up our PDF rendering pipeline, we deployed the following solution.

Internally, Kaleido just outsources the problem of rendering the graph as an image to an included Chrome webbrowser. This means, for Python itself rendering this image is basically waiting for I/O.

To speed up this particular process and since we just wait for I/O, we can use multithreading.

Creating a random graph

Let's start by creating a random figure, like so:

import pandas as pd
import numpy as np
import plotly.graph_objects as go

def get_random_figure() -> go.Figure:
    n_bars = 50
    dates = pd.date_range(start="2021-01-01", end="2021-12-31", freq="M")

    figure = go.Figure()
    for i in range(n_bars):
        values = np.random.rand(len(dates))
        figure.add_trace(go.Bar(x=dates, y=values, name=f"Label {i+1}"))

    figure.update_layout(
        dict(
            barmode="group",
            legend=dict(orientation="h", yanchor="top", xanchor="left"),
        )
    )
    figure.update_layout(yaxis=dict(tickformat=".0%"), xaxis=dict(showgrid=False))
    return figure

Now, turning a figure into an image can be done using the code from above:

from kaleido.scopes.plotly import PlotlyScope
import plotly.graph_objects as go

def figure_to_bytes(figure: go.Figure) -> bytes:
    scope = PlotlyScope()
    return scope.transform(figure=figure, format="png", width=1000, height=1000, scale=4)

And finally we also define for later:

def transform_random_figure() -> bytes:
    return figure_to_bytes(get_random_figure())

Running the image transformation in threads

You may or may not know that due to the GIL (global interpreter lock) in Python, only one thread can execute Python code at the same time. Since the transformation of the graph to an image is not Python code, we can make use of threads to start the transformation of a lot of graphs at the same time and then collect the results.

For that, we define a helper class:

from threading import Thread

# Make the join() method return the value returned by the target function.
class ThreadWithReturnValue(Thread):
    def __init__(self, group=None, target=None, name=None, args=(), kwargs={}, Verbose=None):
        Thread.__init__(self, group, target, name, args, kwargs)
        self._return = None

    def run(self):
        if self._target is not None:
            self._return = self._target(*self._args, **self._kwargs)

    def join(self, *args):
        Thread.join(self, *args)
        return self._return

This class will help us retrieve the result of the transformation (i.e. the bytes of the image).

The next thing we have to do is follow the standard pattern for working with threads in Python:

Start the threads you want to start with the start() method.
Using the join() method to wait for the thread to return the result.

Our threads should each call transform_random_figure() and then return the bytes. We start 10 threads in this case.

n_threads = 10
threads = [ThreadWithReturnValue(target=transform_random_figure) for _ in range(n_threads)]
for thread in threads:
    thread.start()

The start() method will also call the run() method of the thread that starts the actual logic (i.e. executes the given function, which in our case means transform_random_figure()).

To collect the results, we use the join() method of the threads and write the result into files.

results = [thread.join() for thread in threads]
for i, res in enumerate(results):
    with open(f"result{i}.png", "wb") as f:
        f.write(res)

How it works

The main idea here is to, whenever we want to transform a graph into an image, we start a thread and this thread will wait for the graph to be finished in the background.

Once we put together the whole report, we call join() on all threads and retrieve the images for all graphs and then put them into the report.

This way, we can already generate the whole report without graphs and save time by not waiting for every graph on its own to be transformed.

Summary

In summary, if you want to transform multiple Plotly charts into images, use the multithreading module in the Python standard library to speed up your conversion process.

You can do so very easily just by moving the transform() call into a thread and then waiting for all threads to finish.

Appendix: The Code

import logging
from threading import Thread

import numpy as np
import pandas as pd
import plotly.graph_objects as go
from kaleido.scopes.plotly import PlotlyScope

logging.basicConfig(format="%(asctime)s %(message)s", level=logging.DEBUG)
logger = logging.getLogger()


def get_random_figure() -> go.Figure:
    n_bars = 50
    dates = pd.date_range(start="2021-01-01", end="2021-12-31", freq="M")
    figure = go.Figure()

    for i in range(n_bars):
        values = np.random.rand(len(dates))
        figure.add_trace(go.Bar(x=dates, y=values, name=f"Label {i+1}"))

    figure.update_layout(
        dict(
            barmode="group",
            legend=dict(orientation="h", yanchor="top", xanchor="left"),
        )
    )
    figure.update_layout(yaxis=dict(tickformat=".0%"), xaxis=dict(showgrid=False))
    return figure


def figure_to_bytes(figure: go.Figure) -> bytes:
    scope = PlotlyScope()
    logger.info("Transforming the graph %s.", id(figure))
    img_bytes = scope.transform(
        figure=figure,
        format="png",
        width=1000,
        height=1000,
        scale=4,
    )
    logger.info("Done transforming the graph %s.", id(figure))
    return img_bytes


def transform_random_figure() -> bytes:
    return figure_to_bytes(get_random_figure())


# Make the join() method return the value returned by the target function.
class ThreadWithReturnValue(Thread):

    def __init__(self, group=None, target=None, name=None, args=(), kwargs={}, Verbose=None):
        Thread.__init__(self, group, target, name, args, kwargs)
        self._return = None

    def run(self):
        if self._target is not None:
            self._return = self._target(*self._args, **self._kwargs)

    def join(self, *args):
        Thread.join(self, *args)
        return self._return


n_threads = 5
threads = [ThreadWithReturnValue(target=transform_random_figure) for _ in range(n_threads)]
for thread in threads:
    thread.start()

results = [thread.join() for thread in threads]
for i, res in enumerate(results):
    with open(f"result{i}.png", "wb") as f:
        f.write(res)

Connect to Azure SQL database in SQL Alchemy using Entra ID tokens

Alexander Kammerer — Mon, 28 Oct 2024 21:09:23 +0000

We have a web application at work that needs to connect to our Azure SQL database for the duration of the request. To make querying the database easier, we use SQL Alchemy and pyodbc.

We have a few goals we want to achieve:

Every request gets its own session. We want to open a new session when the request starts and close it once it is done.
We want to connect to the database using the managed identity of the web service (Azure Function App or Azure Web App).
We want to delegate the handling of the session and connection as much as we can.

Session lifetime management

First, to make sure we open a new session for every request, we can wrap the request functions into decorators that make sure a new session is created and subsequently destroyed.

To make sure our session is easily accessible from all over the application without having to hand it over in every function call, we use a singleton pattern. However, since we run multiple threads to handle multiple requests at the same time, we have to make sure that there are no race conditions with respect to the session objects.

SQL Alchemy has a great utility to make this easier: the scoped session.

To use this, we wrap our connection factory into a scoped_session() call:

from urllib.parse import quote_plus
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session

connection_string = "..."
engine = create_engine("mssql+pyodbc:///?odbc_connect={}".format(quote_plus(connection_string))
session_factory = sessionmaker(bind=engine)
Session = scoped_session(session_factory)

Now, whenever we want to use the session, we just call Session() and SQL Alchemy makes sure we reuse the existing session object. If we are done, we can call Session.remove() to close the session. There is no way for SQL Alchemy to tell if the thread is done so we have to do this ourselves.

Connect via Entra ID tokens using managed identity

There are some posts that describe how to setup the connection to Azure SQL databases using access tokens, but the best resource for this is the SQL Alchemy docs themselves.

Let's go over the details together. First, we need a connection string. Since we want to rely on managed identity (or the Azure CLI for local development), we do not put any credentials into the connection string:

Driver={ODBC Driver 18 for SQL Server};Database=YOUR_DB;Server=tcp:you.database.windows.net,1433;Encrypt=yes;TrustServerCertificate=no;Connection Timeout=30

Our general plan is:

Add an event handler that fires whenever we connect to the database.
Retrieve an access token whenever we connect to the database.
Adjust the connection arguments so that we put the (fresh!) access token into the connection string.

Now, let's take a look at the code:

import struct
from urllib.parse import quote_plus

from azure.core.exceptions import ClientAuthenticationError
from azure.identity import DefaultAzureCredential
from sqlalchemy import event


@event.listens_for(engine, "do_connect")
def provide_token(dialect, conn_rec, cargs, cparams):
    # Uid may not be provided for the token authentication to work
    if "Uid=" in cargs[0]:
        return

    # Connection option for access tokens as defined in msodbcsql.h
    SQL_COPT_SS_ACCESS_TOKEN = 1256

    # remove the "Trusted_Connection" parameter that SQLAlchemy adds
    cargs[0] = cargs[0].replace(";Trusted_Connection=Yes", "")
    cargs[0] = cargs[0].replace(";Authentication=TokenIdentifiedPrincipal", "")

    # create token credential => try two ways to make local dev work
    scope = "https://database.windows.net/.default"
    try:
        cred = DefaultAzureCredential(exclude_managed_identity_credential=True).get_token(scope)
    except ClientAuthenticationError:
        cred = DefaultAzureCredential(
            exclude_managed_identity_credential=True,
            exclude_shared_token_cache_credential=True,
        ).get_token(scope)

    raw_token = cred.token.encode("utf-16-le")
    token_struct = struct.pack(f"<I{len(raw_token)}s", len(raw_token), raw_token)
    cparams["attrs_before"] = {SQL_COPT_SS_ACCESS_TOKEN: token_struct}

Summary

With this, we achieved our goals. With the scoped_session() we do not need to open a new session every time a request comes in (this will be handled for us) but we should close the session at the end so that we do not have too many dangling sessions.

We also connect to the SQL database using our own identity (for local dev) or the managed identity of the web service. We modify the connection string each time a new connection is created.

Acknowledgements

A big thank you to David for helping me figure out the concept of scoped sessions.

Connect to MSSQL database in Django using Tokens

Alexander Kammerer — Wed, 14 Aug 2024 13:57:39 +0000

One great way to connect to a MSSQL database for development or production in Django is to use your Microsoft Entra / Azure Active Directory identity. This is supported by the MSSQL database backend for Django.

This means, you can add users or groups from Microsoft Entra to your database users and assign them permissions. They can then login using this identity instead of a regular database user.

Create the user or group in the database

First, you will have to create the user or group in the database (and optionally assign a default schema) via:

CREATE USER [your-entra-group] FROM EXTERNAL PROVIDER;
ALTER USER [your-entra-group] WITH DEFAULT_SCHEMA=[dbo]

This makes your database recognize the user or group and ensures they can log in.

Make sure you can acquire a token

Now, make sure you are either logged in through Azure CLI or in Visual Studio Code (if you are using this editor) with your Microsoft identity that you also want to use for the authentication with the database.

Change your Django config

Then, you have to update your database config so that it uses the token method to authenticate.

First, you have to install the azure-identity package (add it to your requirements.txt / pyproject.toml / your package management). Then, add this to your settings.py before the database configuration:

import time

from azure.core.credentials import AccessToken
from azure.core.exceptions import ClientAuthenticationError
from azure.identity import DefaultAzureCredential

class DatabaseToken(str):
    cached_token: AccessToken | None = None

    def __new__(cls) -> None:
        instance = super().__new__(cls, "")
        instance.get_access_token()
        return instance

    def token_is_valid(self) -> bool:
        if self.cached_token is None:
            return False

        return self.cached_token.expires_on > time.time()

    def get_new_token(self) -> AccessToken:
        try:
            return DefaultAzureCredential().get_token("https://database.windows.net/.default")
        except ClientAuthenticationError:
            pass

        return DefaultAzureCredential(exclude_shared_token_cache_credential=True).get_token(
            "https://database.windows.net/.default"
        )

    def get_access_token(self) -> AccessToken:
        if self.token_is_valid():
            return self.cached_token

        self.cached_token = self.get_new_token()
        return self.cached_token

    def encode(self, *args, **kwargs) -> bytes:
        return self.get_access_token().token.encode(*args, **kwargs)

Finally, you have to update your database configuration in settings.py:

DATABASES = {
    "default": {
        "ENGINE": "mssql",
        "NAME": "YOUR_DATABASE",
        "TOKEN": DatabaseToken(),
        "HOST": "YOUR_URL",
        "CONN_MAX_AGE": 30 * 60,  # max of 30 min, since AT is valid for 60 min
        "OPTIONS": {"driver": "ODBC Driver 18 for SQL Server"},
        "TEST": {"NAME": "YOUR_TEST_DATABASE"},
        "TIME_ZONE": "UTC",
    }
}

Here, you have to update the values for the HOST and NAME keys. We are using ODBC Driver 18 for SQL Server which you might want to update depending on the specific driver you are using.

The above code using the DatabaseToken() instance might seem overly complex. As is described in this GitHub issue, we could also just use one line: DefaultAzureCredential().get_token("https://database.windows.net/.default").token.

I have found that the simple way sometimes leads to problems depending on the specifics of the development setup. Sometimes it will try to use a token from the shared token cache which might fail without an easy way to fix this. I have found that falling back to acquiring a token through another way (e.g. Azure CLI) works great.

Summary

Setting up token authentication for MSSQL databases with Django is quite easy thanks to the support for the specification of the token as part of the login credentials. You will have to give your Entra user or a group that your user is part of access to the database, login using that user, and then adjust your settings.py.

Azure Pipelines: uv Cache

Alexander Kammerer — Thu, 25 Jul 2024 08:56:52 +0000

I have already written about how to make your docker build faster with uv. Today, I want to quickly show how to make your CI build for a python project faster.

uv is a great pip alternative that for most projects will work as a drop-in replacement. It is a lot quicker in resolving and installing python packages than pip which is especially nice for CI pipelines (or docker builds). Faster pipelines mean a quicker dev turnaround which means higher productivity.

If you keep installing the same packages in your pipeline, you might as well reuse the cache that uv internally uses across pipeline runs.

Cache Task in the Pipeline

For that, you have to insert the following code in your azure-pipelines.yml before you run uv venv and uv pip install ... and after you have installed uv:

- bash: |
    echo "##vso[task.setvariable variable=uv-cache-path;]$(uv cache dir)"

- task: Cache@2
  displayName: Cache uv
  inputs:
    key: uv cache | "$(python.version)"
    path: $(uv-cache-path)

This sets a new variable uv-cache-path with the path to the uv cache and then it will cache this path for successive pipeline runs. I assume that you store the python version that you use in a variable called python.version. You can also replace this directly with the python version that you use. Make sure to keep the quotes around the version though as the task will otherwise interpret the version as a path.

Prune the Cache

Finally, we can use a new feature to prune the cache before we upload it. This will decrease the file size of the cache which makes the upload and download of the cache faster and therefore probably saves you time. You should run uv cache prune --ci before the end of your pipeline job:

- script: uv cache prune --ci

Summary

In summary, if you want to make your python pipeline faster, the first step should be to use uv. You can simply replace the pip ... calls by uv pip ... and see how much faster that makes your pipeline.

If you install a lot of packages that need extensive build steps, it might be worth to re-use the uv cache. For that, you can use the Cache@2 task provided by Azure Pipelines.

Connect to Azure SQL in R via Entra ID / AAD tokens from Azure CLI

Alexander Kammerer — Fri, 28 Jun 2024 13:31:15 +0000

One of the best features of the Azure SQL server is that you can connect to your database using your Azure Active Directory (AAD, now called Entra ID) identity.

This works by retrieving an authentication token from Entry ID and then specifying this in the pre-connection attribute SQL_COPT_SS_ACCESS_TOKEN. If you are interested in the details, you can checkout the issue about the implementation in the odbc package for R.

This method does not only work for R but for any language that provides a package for an odbc driver that supports it. More information can be found in the Microsoft docs for Python or a tutorial that explains the necessary steps in more detail.

Let's go back to R. First, you need to use the odbc package for R and install the Microsoft ODBC driver for SQL Server. You have probably installed the driver before if you have been connected to your sql server.

Retrieve the token

There are multiple ways to retrieve a token. If you are developing locally, one of the easiest ways is to use Azure CLI. Make sure you have Azure CLI installed.

Make sure that you are logged by running this command in a terminal: az login --allow-no-subscriptions. The flag --allow-no-subscriptions makes sure you are able to retrieve a token even if you have no Azure subscriptions.

Then, you can retrieve a token. I am using the withr package here to make my life easier:

result <- withr::with_tempfile("tf", {
  suppressWarnings({
    token <- system2(
      "az",
      c(
        "account get-access-token",
        "--resource https://database.windows.net",
        "--query accessToken",
        "--only-show-errors",
        "--output yaml"
      ),
      stdout = TRUE,
      stderr = tf
    )
  })
  system2error <- readLines(tf)
  list(token = token[1], error = system2error)
})
print(result$token)

This should print the token in your R console. Please note: this token is like a password. Please handle it with care.

Connect to the database

You can check the attributes of this JWT using a handy tool from Microsoft directly: jwt.ms. It is safe to post your token there.

Finally, you have to connect to your server:

con <- DBI::dbConnect(
  odbc::odbc(),
  # please check which version of the driver you are using (18 is the most recent one)
  driver = "ODBC Driver 18 for SQL Server",
  server = "your.database.com,1433",
  database = "your-database-name",
  Encrypt="yes",
  TrustServerCertificate="No",
  attributes = list("azure_token" = token)
)

Note that some people were unable to connect with a token retrieved for the resource https://database.windows.net but had to retrieve one for https://database.windows.net/ (notice the / at the end).

Multi-Stage Docker Builds for Python Projects using uv

Alexander Kammerer — Wed, 21 Feb 2024 00:20:09 +0000

Charlie Marsh and the brillant guys at Astral have helped the python ecosystem a lot with ruff and now they have released a new tool: uv.

It is intended as a drop-in replacement of pip. At the moment, it supports dependency resolution and installation. Compared to pip, it is a lot faster. I have observed up to 10-15x faster dependency installations (with no caching) for medium-sized projects.

Using uv to speed up Docker builds

Another nice benefit is that his can speed up your Docker build. You can even use uv without having to install it into your final Docker image using a multi-stage build.

Multi-stage builds

A typical multi-stage build for a Python projects works like this:

We use a build image that has all the necessary tools to build certain python packages.
Our final image is a slimmed down version of the build image that contains no build dependencies. We copy over the compiled dependencies from the build image to our final image.

Example: Azure Function App

Let's look at this example of a Dockerfile for an Azure Function App.

# syntax = docker/dockerfile:1.0-experimental

# Build image to compile all packages
FROM python:3.11 as build

ENV VIRTUAL_ENV=/home/packages/.venv
ADD https://astral.sh/uv/install.sh /install.sh
RUN chmod -R 655 /install.sh && /install.sh && rm /install.sh

COPY ./requirements.txt .
RUN /root/.local/bin/uv venv /home/packages/.venv
RUN /root/.local/bin/uv pip install --no-cache -r requirements.txt

# Run image for the function app code
FROM mcr.microsoft.com/azure-functions/python:4-python3.11

ENV AzureWebJobsScriptRoot=/home/site/wwwroot \
    AzureFunctionsJobHost__Logging__Console__IsEnabled=true \
    AzureWebJobsFeatureFlags=EnableWorkerIndexing \
    PATH="/home/packages/.venv/bin:$PATH" \
    VIRTUAL_ENV=/home/packages/.venv

COPY --from=build /home/packages/.venv /home/packages/.venv
COPY . /home/site/wwwroot

(Thank you to @ryxcommar for the Dockerfile commands that install uv from his blog article.)

Let's break this down assuming you have your Function App files in the directory for which you are calling docker buildx build .:

First, we use the python:3.11 image as our build image. It is close enough to our base image for the final image.
We install uv and create a virtual environment at /home/packages/.venv using uv venv /home/packages/.venv.
We install our dependencies from the requirements.txt file that we copied over.

This is it for the build image. Now, we need a final image:

Our final image is based on the image necessary for the Python Function App mcr.microsoft.com/azure-functions/python:4-python3.11.
Then, we set some environment variables. Here, the important ones are the PATH and the VIRTUAL_ENV to make sure our virtual environment is picked up by the system.
Then, we copy over the complete virtual environment from /home/packages/.venv in the build image to our final image.
Finally, we copy all the files necessary for the Function App into our image.

Example: Python app

A more general example for a Dockerfile of a Python app could be this:

# syntax = docker/dockerfile:1.0-experimental

# Base image
FROM python:3.12 as build

RUN apt-get update && apt-get install -y build-essential curl
ENV VIRTUAL_ENV=/opt/venv \
    PATH="/opt/venv/bin:$PATH"

ADD https://astral.sh/uv/install.sh /install.sh
RUN chmod -R 655 /install.sh && /install.sh && rm /install.sh
COPY ./requirements.txt .
RUN /root/.cargo/bin/uv venv /opt/venv && \
    /root/.cargo/bin/uv pip install --no-cache -r requirements.txt

# App image
FROM python:3.12-slim-bookworm
COPY --from=build /opt/venv /opt/venv

# Activate the virtualenv in the container
# See here for more information:
# https://pythonspeed.com/articles/multi-stage-docker-python/
ENV PATH="/opt/venv/bin:$PATH"

In this example, we use the python:3.12 image as the build image and then the slimmed down version python:3.12-slim-bookworm for the final app image.

You will need to add your ENTRYPOINT to make this work but the dependencies will be installed and calling python will use your virtual environment.

If you want to learn more about multi-stage docker builds, I can highly recommend this article about multi-stage docker builds for python and also this post about using virtual environments in dockerfiles from Itamar Turner-Trauring.

Update (2024-11-11):

In the newest version the path to uv changes from /root/.cargo/bin/uv to /root/.local/bin/uv.

Django and HTMX

Alexander Kammerer — Sun, 19 Feb 2023 20:09:44 +0000

I have been using Django for multiple years now at work and we have used it to build up a website that gives insight into much of the data that we store in our database. It allows users to interact with that data and make minor changes.

We do not need a large interactive application for that and have been happy with just reloading the page whenever we needed to make a request to the server. For our applications, this was fine and using one of the frontend frameworks to increase the interactivity of our website would have meant a large amount of necessary initial development and ongoing maintenance.

Then I found HTMX on Twitter and Github and I instantly fell in love. It promises multiple things:

Simplicity: use the things you already use and only as much Javascript as you want.
Ease of use: stay with your HTML templates generated on the server. No need for JSX / frontend templates.
Development speed: quickly build up interactive solutions and improve the UX massively.
Learning speed: new developers quickly grasp the main concepts and are able to maintain existing solutions and build new ones.

In the following article I want to describe our HTMX + Django setup. We will talk about the following:

Using the django-htmx package.
Setting the CSRF token as part of your HTMX requests.
Easily re-using parts of your Jinja2 templates.
Updating CSRF tokens after the user was forced to re-login.

Package: `django-htmx`

There is a great package called django-htmx. The most important things it provides are:

A middleware that will allow you to check if a request was triggered by HTMX via request.htmx and retrieve other parameters from the request.
The docs of the package also provide some tips for how to use HTMX with Django.
There are some convenient functions to return certain HTTP status codes or headers that will make HTMX do certain things. Read more on the HTTP page.

Note that the tips page describes how you can make your life easier if you often want to render a part of your page again using HTMX. This can be useful for filtering lists, validating forms, and showing content that is polled from a database in regular intervals.

But if you want to render multiple parts of your page again, these tips will not help you. I will show you a better way below.

CSRF Token

If you enable Django's CSRF protection, you will need to set the CSRF token as part of any HTMX request. This can easily be done with the following code that you place inside of your template.

let CSRF_TOKEN = '{{ csrf_token }}';
document.body.addEventListener('htmx:configRequest', (event) => {
    event.detail.headers['X-CSRFToken'] = CSRF_TOKEN;
})

This way, every HTMX request will have the X-CSRFToken header set.

Rendering parts of your template

Let's say you have a large page with multiple areas that can be dynamically changed and will be reloaded using HTMX. (This could be that you have multiple forms on the page or you have multiple lists where you want to offer dynamic filtering without reloading the page.)

Now, what you would have to do is put every such area that is reloaded in their own template and then re-use these templates in the larger template for that page. This gets annoying pretty fast as your have to split your templates in many different files.

Instead, we will show you how you can keep your whole page template in one file and still reuse this template.

There is a great repository that contains a lot of information about using Django + HTMX. One idea in particular is very interesting for us and it is called: inline partials.

For now, we will assume you are using Jinja2 as your template engine. You can make the following also work with the normal Django template engine but we will focus on the Jinja2 solution. If you are already using Jinja2 as a template engine for Django, there are some solutions that offer template fragment features:

Both of them do not work well with the way Jinja2 is integrated into Django. They have the following drawbacks:

You are giving up on the Jinja2 environment that Django configures for you.
You cannot use TemplateResponse() any longer.
You need to adjust some of your views quite heavily to make use of the new render logic.

We have a better solution: write your own Django template backend. It is less than 50 lines of code.

To make working with Django + Jinja2 + Fragments easier we have written a custom template backend that is heavily inspired by the default Django Jinja2 backend.

It mainly relies on the fact that you can render blocks in the same way you would render a full template in Jinja as the blocks in template.blocks are render functions that take a context as an input.

import jinja2
from django.template import TemplateDoesNotExist, TemplateSyntaxError
from django.template.backends.jinja2 import Jinja2, Template, get_exception_info


class Jinja2WithFragments(Jinja2):
    def from_string(self, template_code):
        return FragmentTemplate(self.env.from_string(template_code), self)

    def get_template(self, template_name):
        try:
            return FragmentTemplate(self.env.get_template(template_name), self)
        except jinja2.TemplateNotFound as exc:
            raise TemplateDoesNotExist(exc.name, backend=self) from exc
        except jinja2.TemplateSyntaxError as exc:
            new = TemplateSyntaxError(exc.args)
            new.template_debug = get_exception_info(exc)
            raise new from exc


class FragmentTemplate(Template):
    """Extend the original jinja2 template so that it supports fragments."""

    def render(self, context=None, request=None):
        from django.template.backends.utils import csrf_input_lazy, csrf_token_lazy

        if context is None:
            context = {}
        if request is not None:
            context["request"] = request
            context["csrf_input"] = csrf_input_lazy(request)
            context["csrf_token"] = csrf_token_lazy(request)

            for context_processor in self.backend.template_context_processors:
                context.update(context_processor(request))

        try:
            if "RENDER_BLOCKS" in context:
                bctx = self.template.new_context(context)
                blocks = []
                for bn in context["RENDER_BLOCKS"]:
                    blocks.extend(self.template.blocks[bn](bctx))
                return "".join(blocks)
            return self.template.render(context)
        except jinja2.TemplateSyntaxError as exc:
            new = TemplateSyntaxError(exc.args)
            new.template_debug = get_exception_info(exc)
            raise new from exc

You need to configure your Django settings to use this new template engine. Create a jinja2.py file inside your your_app folder and place the code from above in this file. Also make sure that your environment is also in this file or that you adjust the path to your environment.

TEMPLATES = [
    # ...
    {
        "BACKEND": "your_app.jinja2_backend.Jinja2WithFragments",
        "DIRS": [os.path.join(PROJECT_DIR, "jinja2")],
        "APP_DIRS": True,
        "OPTIONS": {"environment": "your_app.jinja2.environment"},
    },
]

If you define any block in your templates:

{% extends "base.html" %}

{% block body %}
    <h1>List of monsters</h1>

    {% if page_obj.paginator.count == 0 %}
        <p>We have no monsters at all!</p>
    {% else %}
        {% block page-and-paging-controls %}
            {% for monster in page_obj %}
                <p class="card">{{ monster.name }}</p>
            {% endfor %}

            {% if page_obj.has_next %}
                <p id="paging-area">
                    <a href="#"
                        hx-get="?page={{ page_obj.next_page_number }}"
                        hx-target="#paging-area"
                        hx-swap="outerHTML">Load more</a>
                </p>
            {% else %}
                <p>That's all of them!</p>
            {% endif %}
        {% endblock %}
    {% endif %}
{% endblock %}

You can now choose to render only a certain block quite easily via:

def paging_with_inline_partials(request):
   template_name = "paging_with_inline_partials.html"
   context = {
       "page_obj": get_page_by_request(request, Monster.objects.all()),
   }

   if request.headers.get("Hx-Request", False): # or use: request.htmx
       context["RENDER_BLOCKS"] = ["page-and-paging-controls"]

   return TemplateResponse(request, template_name, context)

In theory, you could also render multiple blocks at the same time even though we do not yet see the usecase for this.

Our template backend will look for the key RENDER_BLOCKS inside the context and if it is available, it will switch to rendering only the blocks that are specified in the variable.

Advanced: Refreshing the CSRF Token after login

We have the following situation:

For different reasons, the time after a user is logged out is quite short (approximately one hour).
So, users could open a page, fill out a form, do something else and come back to the form only to submit and find that they have been logged out.
This is frustrating for the users because they lose the contents of their form.

One obvious way to prevent this is to make every endpoint that a form send data to not reject the request outright or send the user to the login form. We need to cache the form input, send the user to the login, and then use the form input from the cache. This is a lot of difficult work.

We think, our solution is easier to execute and does not require every form to follow certain rules. We do the following:

We show the user the login state prominently on the page and refresh this state in the background (every 60s and when the user comes back to the tab).

<div
    hx-get="{{ url('auth:login_status') }}"
    hx-trigger="load, every 60s, visibilitychange[document.visibilityState === 'visible'] from:document"
    hx-target="find span"
    hx-indicator=".login-status"
    class="login-status"
>
    <span></span><i class="htmx-indicator fa fa-spinner fa-spin"></i>
</div>

The view for the auto:login_status request looks like this:

def get_login_status(request):
    return TemplateResponse(
        request,
        "auth/login_status.html.j2",
        {
            "is_authenticated": request.user.is_authenticated,
            "login_url": your_function_to_get_login_url(request),
        },
    )

And finally, the template for this view looks like this:

{% if is_authenticated %}
    You are logged in. <i class="far fa-check-circle text-success-emphasis" title="Logged in."></i>
{% else %}
    You have been logged out. <i class="far fa-times-circle text-danger-emphasis" title="Your login has expired."></i> Please <a href="{{ login_url }}" target="_blank">login</a> again.
{% endif %}

<script type="text/javascript">
CSRF_TOKEN = '{{ csrf_token }}';
document.querySelectorAll("[name='csrfmiddlewaretoken']").forEach((elem) => {
    elem.setAttribute("value", CSRF_TOKEN);
});
</script>

This way, we send back the current CSRF token every time and update not only the CSRF token that is used for every HTMX request but also the token that is used whenever a form is sent.

Also, every time the status requests is sent, we check if the user is authenticated. If he is not authenticated, we send back a link to the login page that is opened in a new tab. The user can then login there and come back to the old tab. There, we will (because the visibilitychange event is triggered) send another login status request, update the CSRF token, and also update the information about the user's login state. The user will then be able to submit the form he was working on as if he reloaded the page and we saved his form inputs.

AAD auth for Plotly Dash

Alexander Kammerer — Sun, 03 Apr 2022 18:28:51 +0000

This article shows you how to use Azure Active Directory authentication to protect your dashboards. We will implement SSO using the OAuth 2.0 flow that is supported by AAD.

The rise of data science and dashboards

The usage of dashboards to visualize and explain data to a number of internal and external clients has increased heavily in the last few years. Together with the strong buzz around data science topics, this has resulted in a lot of interesting new software, including many open source libraries. On one side, Tableau is as a large provider of ready-made solutions around data visualizations. On the other side, there is also a strong DIY community that bets on open-source technology that is sometimes assisted by paid offerings or paid consulting.

One particular interesting case is Dash by Plotly. While I myself have previously used Bokeh, I quickly made the transition to Dash since I felt it was more ready for usage as a deployed application. Additionally, having access to Plotly as a charting library is a big plus because it is such a successful open-source project with a strong community and a fantastic library.

One important thing to being able to deploy our application is of course authentication. I wanted to use SSO via AAD to secure our Dash apps. The hurdles for that are:

There is both a huge number of articles detailing AAD integration and not very many useful articles. There are so many usecases for authentication (and authorization) with AAD that it is hard to find exactly what you are looking for.
There are multiple packages that claim to support AAD authentication for flask. But some of them are very heavily built around one usecase that may not fit yours.
Authentication with AAD uses the OpenID flow of the OAuth protocol. The OAuth procotol is quite difficult to understand already and it is not helped by the fact that every provider implements it a little different.

I want to present my usecase and then show you how I implemented it. What I wanted is:

Implement SSO using AAD so that a user can authenticate to the Dash app using his AAD account.
I want to read the user's name and email from the issued OpenID token so that I can use this in my app.

Using flask dance

The OpenID flow is not an easy thing to implement correctly and it usually is a good idea to rely on implementations that are used by many people instead of using your own. I have found flask dance to be a great choice for this integration. It will take care of the whole OAuth 2.0 protocol flow (the "dance") for you and all you need to do is configure the library the right way.

Let's look at the basic setup:

from flask import Flask
from flask_dance.contrib.azure import make_azure_blueprint

blueprint = make_azure_blueprint(
    client_id="your_client_id",
    client_secret="your_client_secret",
    tenant="your_tenant_id",
    scope=["openid", "email", "profile"],
)

app = Flask(__name__)
app.register_blueprint(blueprint, url_prefix="/login")

Here, we have a flask instance and the blueprint from the flask dance library to activate the AAD authentication. What you need to do now is the following:

You need to create an app registration in your AAD blade in the Azure portal. This will give you a client ID and a tenant ID. Microsoft has a great tutorial that will guide you through the process. Follow the part of the tutorial that is called Register an application.
You need to create a client secret for usage in your deployment or local development. This is described in the Add credentials section.
Replace the placeholders in the make_azure_blueprint() call with your values.
Add localhost:8050/login/azure/authorized to your redirect URIs in the app registration.

It is good to know what flask dance is doing in the background. And to understand that, you need to know that is necessary for the OpenID flow to work:

The first step is to send a request for authentication to Microsoft. For that, you redirect the user to some login site by Microsoft where he inputs his credentials and authenticates with his Microsoft account. Flask dance adds a route to your flask app that redirects you to the Microsoft endpoint for authentication.
Then, if the user logged in on the Microsoft page, you will eventually get different tokens (in the the form of JWTs) that serve the purpose to verify the authentication (open id token), allow access to Microsoft services (access token), and renew the access token (refresh token). If you are only interested in the login, only the open id token is of interest to you.
Flask dance will add a route to your flask server so that the response from Microsoft can be parsed and with the help of the response the tokens are retrieved and verified.
Optional: if you want, you can use one of these tokens to authenticate as the user to Microsoft APIs, if the user has given you access. If you only want SSO, make sure your scope is only scope=["openid", "email", "profile"] and nothing else.

Putting your Dash app behind authentication

The next step is to make sure that your users can only access the Dash app if they are logged in. This means, we need to verify the login and if the user is not logged-in, prompt him to login. This can be done using decorators for the flask routes that you want to protect. Let's take a look at the decorator that we want to use:

from flask import redirect, url_for
from flask_dance.contrib.azure import azure

def login_required(func):
    """Require a login for the given view function."""

    def check_authorization(*args, **kwargs):
        if not azure.authorized or azure.token.get("expires_in") < 0:
            return redirect(url_for("azure.login"))
        else:
            return func(*args, **kwargs)

    return check_authorization

We have done multiple things here:

We wrap the given function so that we first check if the user is authorized.
If the user is authorized, we call the original function and return the result.
If the user is not authorized or the token is expired, we redirect the user to login again.

Now, what is left is actually protecting the routes. For this, we assume that your flask instance is called app:

for view_func in app.view_functions:
    if not view_func.startswith("azure"):
        app.view_functions[view_func] = login_required(app.view_functions[view_func])

You need to protect all the routes of your dash server except for the login routes. To make sure to hit all the views that Dash adds without actually specifying them, we can simply cycle through all views and protect them with the login_required() decorator.

We iterate over all your view functions and protect the ones that do not start with azure. This way, we make sure that we do not protect the view functions that accept the authentication result from Microsoft and the login route that redirects the user to the Microsoft login site. This is important because the user will not be authenticated when we receive the result and only after we verified it.

Putting everything together

A full example of this would then be:

from dash import Dash, html
from werkzeug.middleware.proxy_fix import ProxyFix
from flask import Flask, redirect, url_for
from flask_dance.contrib.azure import azure, make_azure_blueprint


def login_required(func):
    """Require a login for the given view function."""

    def check_authorization(*args, **kwargs):
        if not azure.authorized or azure.token.get("expires_in") < 0:
            return redirect(url_for("azure.login"))
        else:
            return func(*args, **kwargs)

    return check_authorization

blueprint = make_azure_blueprint(
    client_id="your_client_id",
    client_secret="your_client_secret",
    tenant="your_tenant_id",
    scope=["openid", "email", "profile"],
)

app = Flask(__name__)
app.config["SECRET_KEY"] = "secretkey"
app.register_blueprint(blueprint, url_prefix="/login")

dash_app = Dash(__name__, server=app)

# use this in your production environment since you will otherwise run into problems
# https://flask-dance.readthedocs.io/en/v0.9.0/proxies.html#proxies-and-https
app.wsgi_app = ProxyFix(app.wsgi_app, x_proto=1, x_host=1)

# It is very important to run this AFTER you called Dash(..., server=app) because that will add the routes to your flask app.
# If you run this before adding the routes, nothing will be protected.
for view_func in app.view_functions:
    if not view_func.startswith("azure"):
        app.view_functions[view_func] = login_required(app.view_functions[view_func])

dash_app.layout = html.Div(children=[
  html.H1(children='Hello Dash'),
  html.Div(children="You are logged in!")
])

if __name__ == '__main__':
    dash_app.run_server(debug=True)

Note a few things that were previously missing that I have added now:

Make sure to use a long string for the SECRET_KEY and consult the flask documentation. You will need to set it because flask dance uses the session to store the tokens for the user.
You can only access the app via localhost (not 127.0.0.1) because otherwise the authorization will not work. This is due to the redirect URIs that Microsoft allows.
Make sure to apply the proxy fix if you are using this in production behind an SSL proxy.
In production, you will need to add the URL https://yourdomain.com/login/azure/authorized to your redirect URIs for your AD registered app. Of course, make sure to use the actual URL of your app instead of https://yourdomain.com.
To make the authentication work, you need to add OAUTHLIB_RELAX_TOKEN_SCOPE=1 to your environment variables configuration (for development and production). This is because AAD does not guarantee to return the scope in the same order as we requested it.
Only and only for local development will you need to add OAUTHLIB_INSECURE_TRANSPORT=1 to your environment variables to make the authentication work. This is because locally, the login token will not be transported via SSL.

If you have followed my instructions, you should be able to run the above code and when you visit localhost:8050, you should be forwarded to localhost:8050/login/azure that then redirects you to the login site for Microsoft. After you login, you are redirected to localhost:8050/login/azure/authorized and then finally to localhost:8050 where you should see the text You are logged in!.

Extracting the user information

One nice touch is to display the name of the user that is logged in. This is easily doable using the token that we get after the authentication flow. I am leaving it up to you how you want to integrate the token in your app, just make sure that you only use the azure variable in the context of a request as it is tied to a session that is only available when you have an actual request. This may for example interfere when you include the username directly in your Dash layout. The Dash layout is rendered before the request comes in and therefore you do not yet know who your user is. You have two options then:

Make the Dash layout a function that is then rendered each time a user visits the page (and therefore you have a request context).
Make the username the output of a callback. Callbacks live in request contexts and therefore you can use the azure variable there.

Now let us take a look at the code that decods the usename from the token. I am using the pyjwt library to work with the token.

import jwt
from flask_dance.contrib.azure import azure

# no need to verify the token here, that was already done before
id_claims = jwt.decode(azure.token.get("id_token"), options={"verify_signature": False})
name = id_claims.get("name")

There is more information available in the token like the email of the user. For that, just look at the claims of the token yourself or use the website jwt.ms by Microsoft to inspect your token (your token, which essentially is a credential, never leaves your browser).

Wrap-up

Making AAD authentication work with flask is basically all you need to do to also make it work with your Dash setup. For that, we have used flask dance and configured it to work with an AAD registered app. The final touches are some details to make the OAuth flow work locally and in production and you are all set. You can then extract further details from the token like the name or the email of the user.

DEV Community: Alexander Kammerer

TIL: How to reset all files in one folder in git

Copy Markdown to Teams

The Plan: Markdown to HTML to Teams

Nushell

Powershell

Creating a Shortcut (on Windows)

The New and Improved Usage

Clean up your Azure Container Registry

Converting Plotly charts into images in parallel

Rendering graphs with Kaleido

Rendering a lot of graphs

Creating a random graph

Running the image transformation in threads

How it works

Summary

Appendix: The Code

Connect to Azure SQL database in SQL Alchemy using Entra ID tokens

Session lifetime management

Connect via Entra ID tokens using managed identity

Summary

Acknowledgements

Connect to MSSQL database in Django using Tokens

Create the user or group in the database

Make sure you can acquire a token

Change your Django config

Summary

Azure Pipelines: uv Cache

Cache Task in the Pipeline

Prune the Cache

Summary

Connect to Azure SQL in R via Entra ID / AAD tokens from Azure CLI

Retrieve the token

Connect to the database

Multi-Stage Docker Builds for Python Projects using uv

Using uv to speed up Docker builds

Multi-stage builds

Example: Azure Function App

Example: Python app

Django and HTMX

Package: django-htmx

CSRF Token

Rendering parts of your template

Advanced: Refreshing the CSRF Token after login

AAD auth for Plotly Dash

The rise of data science and dashboards

Using flask dance

Putting your Dash app behind authentication

Putting everything together

Extracting the user information

Wrap-up

Package: `django-htmx`