November 10, 2022 6 min read

Using Codex Models for Azure Development Automation

OpenAI’s Codex models, available through Azure OpenAI Service, are specifically trained for code generation. They power GitHub Copilot and can dramatically accelerate Azure development tasks.

Understanding Codex

Codex models (code-davinci-002, code-cushman-001) are descendants of GPT-3 fine-tuned on public code repositories. They understand:

Multiple programming languages
API patterns and conventions
Framework idioms
Infrastructure as Code

Setting Up for Azure Development

import openai
import os

openai.api_type = "azure"
openai.api_base = os.getenv("AZURE_OPENAI_ENDPOINT")
openai.api_key = os.getenv("AZURE_OPENAI_KEY")
openai.api_version = "2022-12-01"

def generate_code(prompt: str, language: str = "python") -> str:
    """Generate code using Codex."""

    full_prompt = f"# Language: {language}\n{prompt}\n\n"

    response = openai.Completion.create(
        engine="code-davinci-002",
        prompt=full_prompt,
        max_tokens=2000,
        temperature=0,
        stop=["# End", "```"]
    )

    return response.choices[0].text.strip()

Generating Azure Infrastructure Code

Terraform Generation

prompt = """# Create Terraform code for:
# - Azure Resource Group in East US
# - Azure Storage Account with blob container
# - Private endpoint for storage
# - Azure Key Vault for secrets
# - All resources tagged with environment=dev

terraform {
  required_providers {
    azurerm = {
      source  = "hashicorp/azurerm"
      version = "~>3.0"
    }
  }
}
"""

terraform_code = generate_code(prompt, "terraform")
print(terraform_code)

Output:

provider "azurerm" {
  features {}
}

locals {
  tags = {
    environment = "dev"
  }
}

resource "azurerm_resource_group" "main" {
  name     = "rg-dev-eastus"
  location = "East US"
  tags     = local.tags
}

resource "azurerm_storage_account" "main" {
  name                     = "stdeveastus001"
  resource_group_name      = azurerm_resource_group.main.name
  location                 = azurerm_resource_group.main.location
  account_tier             = "Standard"
  account_replication_type = "LRS"

  network_rules {
    default_action = "Deny"
    bypass         = ["AzureServices"]
  }

  tags = local.tags
}

resource "azurerm_storage_container" "main" {
  name                  = "data"
  storage_account_name  = azurerm_storage_account.main.name
  container_access_type = "private"
}

resource "azurerm_private_endpoint" "storage" {
  name                = "pe-storage-dev"
  location            = azurerm_resource_group.main.location
  resource_group_name = azurerm_resource_group.main.name
  subnet_id           = azurerm_subnet.private.id

  private_service_connection {
    name                           = "storage-connection"
    private_connection_resource_id = azurerm_storage_account.main.id
    subresource_names              = ["blob"]
    is_manual_connection           = false
  }

  tags = local.tags
}

resource "azurerm_key_vault" "main" {
  name                = "kv-dev-eastus-001"
  location            = azurerm_resource_group.main.location
  resource_group_name = azurerm_resource_group.main.name
  tenant_id           = data.azurerm_client_config.current.tenant_id
  sku_name            = "standard"

  purge_protection_enabled = true

  tags = local.tags
}

Bicep Generation

prompt = """# Create Bicep template for:
# - Azure Container Registry with admin enabled
# - Azure Container Apps Environment
# - Container App with auto-scaling
# Parameters for environment name and location

@description('Environment name')
param environmentName string

@description('Location for resources')
param location string = resourceGroup().location
"""

bicep_code = generate_code(prompt, "bicep")

Generating Azure SDK Code

Python SDK for Cosmos DB

prompt = """# Python class for Azure Cosmos DB operations
# Requirements:
# - Async support using aiohttp
# - CRUD operations for a "products" container
# - Pagination support
# - Change feed processing
# - Bulk operations

from azure.cosmos.aio import CosmosClient
from azure.identity.aio import DefaultAzureCredential
import asyncio
from typing import AsyncIterator, List, Optional
"""

cosmos_code = generate_code(prompt, "python")

C# SDK for Service Bus

prompt = """// C# class for Azure Service Bus messaging
// Requirements:
// - Send messages to queue
// - Receive messages with sessions
// - Dead letter handling
// - Message scheduling
// - Batch operations

using Azure.Messaging.ServiceBus;
using Microsoft.Extensions.Logging;
"""

servicebus_code = generate_code(prompt, "csharp")

Generating Azure Functions

prompt = """# Azure Function in Python that:
# 1. Triggers on Blob Storage upload
# 2. Processes image using Azure Computer Vision
# 3. Stores results in Cosmos DB
# 4. Sends notification via Event Grid

import azure.functions as func
import logging
from azure.cognitiveservices.vision.computervision import ComputerVisionClient
from azure.cosmos import CosmosClient
from azure.eventgrid import EventGridPublisherClient
"""

function_code = generate_code(prompt, "python")

Output:

import azure.functions as func
import logging
from azure.cognitiveservices.vision.computervision import ComputerVisionClient
from azure.cognitiveservices.vision.computervision.models import VisualFeatureTypes
from azure.cosmos import CosmosClient, PartitionKey
from azure.eventgrid import EventGridPublisherClient
from azure.core.credentials import AzureKeyCredential
from msrest.authentication import CognitiveServicesCredentials
import os
import json
from datetime import datetime

# Configuration
CV_ENDPOINT = os.environ["COMPUTER_VISION_ENDPOINT"]
CV_KEY = os.environ["COMPUTER_VISION_KEY"]
COSMOS_ENDPOINT = os.environ["COSMOS_ENDPOINT"]
COSMOS_KEY = os.environ["COSMOS_KEY"]
EVENTGRID_ENDPOINT = os.environ["EVENTGRID_ENDPOINT"]
EVENTGRID_KEY = os.environ["EVENTGRID_KEY"]

# Clients
cv_client = ComputerVisionClient(CV_ENDPOINT, CognitiveServicesCredentials(CV_KEY))
cosmos_client = CosmosClient(COSMOS_ENDPOINT, COSMOS_KEY)
database = cosmos_client.get_database_client("images")
container = database.get_container_client("analysis")
eg_client = EventGridPublisherClient(EVENTGRID_ENDPOINT, AzureKeyCredential(EVENTGRID_KEY))


def main(blob: func.InputStream):
    logging.info(f"Processing blob: {blob.name}, Size: {blob.length} bytes")

    try:
        # Analyze image with Computer Vision
        analysis = cv_client.analyze_image_in_stream(
            blob,
            visual_features=[
                VisualFeatureTypes.categories,
                VisualFeatureTypes.description,
                VisualFeatureTypes.tags,
                VisualFeatureTypes.objects
            ]
        )

        # Prepare result document
        result = {
            "id": blob.name.replace("/", "_"),
            "partitionKey": "image-analysis",
            "blobName": blob.name,
            "blobSize": blob.length,
            "processedAt": datetime.utcnow().isoformat(),
            "description": analysis.description.captions[0].text if analysis.description.captions else None,
            "tags": [tag.name for tag in analysis.tags],
            "categories": [cat.name for cat in analysis.categories],
            "objects": [{"name": obj.object_property, "confidence": obj.confidence}
                       for obj in analysis.objects]
        }

        # Store in Cosmos DB
        container.upsert_item(result)
        logging.info(f"Stored analysis result for {blob.name}")

        # Send Event Grid notification
        event = {
            "id": result["id"],
            "eventType": "ImageAnalysis.Completed",
            "subject": f"images/{blob.name}",
            "eventTime": datetime.utcnow().isoformat() + "Z",
            "data": {
                "blobName": blob.name,
                "tags": result["tags"][:5],  # Top 5 tags
                "description": result["description"]
            },
            "dataVersion": "1.0"
        }

        eg_client.send([event])
        logging.info(f"Sent Event Grid notification for {blob.name}")

    except Exception as e:
        logging.error(f"Error processing blob {blob.name}: {str(e)}")
        raise

Building a Development Assistant

class AzureCodeAssistant:
    """AI-powered assistant for Azure development."""

    def __init__(self, engine: str = "code-davinci-002"):
        self.engine = engine

    def explain_code(self, code: str, language: str) -> str:
        """Explain what code does."""
        prompt = f"""# Explain this {language} code:
```{language}
{code}

Explanation:"""

    return self._generate(prompt, max_tokens=500, temperature=0.3)

def find_bugs(self, code: str, language: str) -> str:
    """Find potential bugs in code."""
    prompt = f"""# Find bugs and issues in this {language} code:

{code}

Bugs and Issues:"""

    return self._generate(prompt, max_tokens=500, temperature=0.2)

def optimize(self, code: str, language: str) -> str:
    """Suggest optimizations."""
    prompt = f"""# Optimize this {language} code for performance and readability:

{code}

Optimized code:"""

    return self._generate(prompt, max_tokens=1500, temperature=0)

def convert_language(self, code: str, from_lang: str, to_lang: str) -> str:
    """Convert code between languages."""
    prompt = f"""# Convert this {from_lang} code to {to_lang}:

{code}

{to_lang} code:


        return self._generate(prompt, max_tokens=1500, temperature=0)

    def generate_tests(self, code: str, language: str, framework: str = "pytest") -> str:
        """Generate unit tests for code."""
        prompt = f"""# Generate {framework} tests for this {language} code:
```{language}
{code}

Tests:


        return self._generate(prompt, max_tokens=1500, temperature=0.2)

    def _generate(self, prompt: str, max_tokens: int, temperature: float) -> str:
        response = openai.Completion.create(
            engine=self.engine,
            prompt=prompt,
            max_tokens=max_tokens,
            temperature=temperature,
            stop=["```\n\n", "# End"]
        )
        return response.choices[0].text.strip()


# Usage example
assistant = AzureCodeAssistant()

# Generate tests
code = """
def calculate_cosmos_ru(item_size_kb: float, reads_per_sec: int, writes_per_sec: int) -> int:
    read_rus = reads_per_sec * (item_size_kb / 4) * 1
    write_rus = writes_per_sec * (item_size_kb / 4) * 5
    return int(read_rus + write_rus)
"""

tests = assistant.generate_tests(code, "python", "pytest")
print(tests)

Best Practices for Codex

Be specific: Include language, framework, and requirements
Provide context: Add relevant imports and class signatures
Use temperature 0: For deterministic code generation
Include examples: Show expected input/output
Review carefully: Always review and test generated code

Conclusion

Codex models are powerful tools for accelerating Azure development. From infrastructure as code to SDK implementations, they can generate production-quality code with proper prompting. Integrate them into your development workflow to boost productivity while maintaining oversight.