Conjure Your Azure Infrastructure With Pydantic Like a Real Houdini

Harry Houdini was a great magician. Perhaps even greater than you think! He could get out of any trouble. No matter if he was handcuffed inside a container filled with water, or burried alive under several feet of dirt. He would get away with anything and pull rabbits out of hats just for lunch.

You can be like him, if you want. Let’s find out how you can conjure your Azure Cloud infrastructure out of the hat with Python and Pydantic and save your neck from any predicament, whether it’s about difficult project requirements or crushing deadlines. Let’s do it like a real Houdini!

All Magic Begins With Abstraction

Abstractions are the key to understanding and managing your Azure Cloud infrastructure. Like anything else! It is really wise to model your infrastructure in an object-oriented way. That way, you can truly understand the relationships between the different items.

By modeling your cloud architecture as objects using Python and the Pydantic library, not only will you better understand how your infrastructure works, but you’ll also be able to execute key commands for its development and maintenance. You can also use your models to automate tasks. Python models themselves serve as documentation for what kind of system you have configured.

I will show you how you can use Pydantic to make meaningful abstractions of your infrastructure in a way that serves your business needs.

So let’s get started and model your Azure infrastucture to the extent that you can go ahead deploying awesome blobs to the Azure blob storage.

Why Exactly Azure Blob Storage?

If you want to store and retrieve unstructured data in the cloud, for example images of rabbits and hats, like me right now, Azure Blob Storage is the logical choice. So it wil serve as our example today. Azure Blob Storage is an essential service for storing and retrieving unstructured data in the cloud, like images, videos, and documents. More importantly, if you have your blobs in Azure Blob Storage and someone makes fun of you for whatever reason, it won’t remove the fact that you still have your blobs in the Azure Blob Storage!

Let’s get started!

First things first. If you want to configure an Azure Blob Storage (or any other Azure service), you need to have an Azure subscription and an account. Let’s abstract these entities into Python classes, starting with the subscription. I want to list my subscriptions first in order to know which subscription I am using for my blob storage account. Let’s write a minimal abstraction for the subscription:

from pydantic import BaseModel

class AzSubscription(BaseModel):

    id: str
    subscription_id: str
    authorization_source: str
    display_name: str

Our goal is now to list the subscriptions in Azure and deserialize the JSON response into a list of subscriptions. The basic command to list yor subscriptions “manually” from the shell is:

az account subscription list

We need to understand how to execute this command from Python. The right spell for that is:

import subprocess

result = subprocess.run("az account subscription list", shell=True, capture_output=True, text=True)

Perhaps you don’t want to be that verbose every time when you invoke the shell commands. Let’s create a wrapper function that will execute shell commands:

def x(command: str) -> str:
    result = subprocess.run(command, shell=True, capture_output=True, text=True)
    return result.stdout

Requesting the subscriptions will return a JSON string. I get the following response:

[
  {
    "authorizationSource": "RoleBased",
    "displayName": "Azure subscription 1",
    "id": "/subscriptions/7b0e90e9-eedf-43c4-2394-916ce8f385fd",
    "state": "Enabled",
    "subscriptionId": "7b0e90e9-eedf-43c4-2394-916ce8f385fd",
    "subscriptionPolicies": {
      "locationPlacementId": "Public_2014-09-01",
      "quotaId": "PayAsYouGo_2014-09-01",
      "spendingLimit": "Off"
    }
  }
]

Note that the JSON object describing the subscription contains a lot more information than just the few fields that we defined in our model. We are missing out quite a bit. So if we are going this path and consider how daunting immensely huge the full Azure API is, is this what we are doing right now doomed to be a just a flawed attempt to model the Great Azure API?

Shouldn’t We Model Everything If We Take This Path?

No, no, and once again, no! Namely, when we are modeling our cloud infrastructure to understand and automate it, our purpose is NOT to create an all covering generic library or API for everything that Azure offers. Our model just should contain the key information that is relevant for our infrastructure. So the model space will cover just a tiny fraction of the Azure API surface. That way, it will be a perfect abstraction of our system, nothing less and nothing more. If we covered more than our infrastructure actually is, we would be just going terribly wrong.

Deserializing The JSON Response Into an AzSubscription Object

Okay, back from the meta level to the actual hands-on work. Let’s figure out how to map a JSON object to a Pydantic model so we have a list of subscription objects. First, let’s upgrade our model to deserialize a JSON response into the Pydantic AzSubscription object and add from_json class method:

@classmethod
def from_json(cls, json: Dict[str, str]) -> 'AzSubscription':
    return cls(**{
        'authorization_source':json['authorizationSource'],
        'id': json['id'],
        'subscription_id': json['subscriptionId'],
        'display_name': json['displayName'],
    })

The above listed method deserializes one JSON subscription to one AzSubscription object. But since the response contains an array of those, not just one item, we need to have an array conversion method:

@classmethod
def from_json_list(cls, subscriptions: List[Dict[str, str]]) -> List['AzSubscription']:
    return list(map(cls.from_json, subscriptions))

To actually go Houdini and combine all these pieces into an amazing trick that actually fetches all subscriptions from the Azure hat and maps them into a list of AzSubscription objects in a whim, let’s do some magic:

@classmethod
def conjure_all(cls) -> List['AzSubscription']:
    subscriptions = x('az account subscription list')
    return cls.from_json_list(json.loads(subscriptions))

Method conjure_all will return a list of AzSubscription objects. But we need only one subscription to proceed. Since we have only one subscription in this example, we can safely assume that the first and only subscription in the stack ist the default one that we need. Let’s add a getter for the default subscription.

@classmethod
def fetch_default(cls) -> 'AzSubscription':
    return cls.fetch_all()[0]

To create an account on top of the subscription, we actually really need the ID of the subscription. Let’s add a getter for the subscription ID:

@classmethod
def get_default_subscription_id(cls) -> str:
    return cls.fetch_default().subscription_id

The Complete Example

Putting all the pieces together, we get the following code:

from pydantic import BaseModel
import subprocess
import json

def x(command: str) -> str:
    result = subprocess.run(command, shell=True, capture_output=True, text=True)
    return result.stdout

class AzSubscription(BaseModel):

    id: str
    subscription_id: str
    authorization_source: str
    display_name: str

    @classmethod
    def from_json(cls, json: Dict[str, str]) -> 'AzSubscription':
        return cls(**{
            'authorization_source': json['authorizationSource'],
            'id': json['id'],
            'subscription_id': json['subscriptionId'],
            'display_name': json['displayName'],
        })

    @classmethod
    def from_json_list(cls, subscriptions: List[Dict[str, str]]) -> List['AzSubscription']:
        return list(map(cls.from_json, subscriptions))

    @classmethod
    def conjure_all(cls) -> List['AzSubscription']:
        subscriptions = x('az account subscription list')
        return cls.from_json_list(json.loads(subscriptions))

    @classmethod
    def fetch_default(cls) -> 'AzSubscription':
        return cls.fetch_all()[0]

    @classmethod
    def get_default_subscription_id(cls) -> str:
        return cls.fetch_default().subscription_id

The Account

When we have a solution to fetch all subscriptions from the Azure in an object-oriented way, we can now continue building on top of that. We need an account on top of the actual subscription. So let’s model the account as a Pydantic object:

class AzAccount(BaseModel):

    subscription: AzSubscription

    @classmethod
    def from_subscription(cls, subscription: 'AzSubscription') -> 'AzAccount':
        return cls(subscription=subscription)

We have now a abstraction for our Azure account. It can be instantiated from a subscription. So we have to get first the subscription and then create an account from that subscription. That is a bit cumbersome, given that we only have one subscription anyway. So let’s automate that process by adding a default fetcher also for the account object.

@classmethod
def fetch_default(cls) -> 'AzAccount':
    return cls.from_subscription(AzSubscription.fetch_default())

What if we want to tell Azure that we want to bind our account to the default subscription? No problem!

def set_subscription(self):
    subscription_id = self.subscription.get_id()
    x(f'az account set --subscription {subscription_id}')

That would be it for now, folks. Be free to complete the rest of this code example to fit your needs. Read this article to model the rest of the process.

Long live Harry Houdini!