5 min read
Using Codex Models for Azure Development Automation
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.