from typing import Any, Dict, Optional, List
import asyncio
from ..clients.client_factory import AzureClientFactory
class AzureWebAppManager:
"""Class for managing Azure Web Apps."""
def __init__(self, client_factory: AzureClientFactory):
self.client_factory = client_factory
async def create_web_app(
self,
resource_group: str,
name: str,
app_service_plan: str,
location: str = "westeurope",
runtime_stack: str = "python:3.9",
os_type: str = "Linux",
https_only: bool = True,
client_affinity_enabled: bool = False,
tags: Optional[Dict[str, str]] = None
) -> Dict[str, Any]:
"""Create a new Web App."""
web_client = self.client_factory.get_web_client()
loop = asyncio.get_event_loop()
try:
# Parse runtime stack into framework and version
if ":" in runtime_stack:
framework, version = runtime_stack.split(":", 1)
else:
framework, version = runtime_stack, ""
# Build site configuration based on runtime stack
if framework.lower() == "python":
site_config = {
"linux_fx_version": f"PYTHON|{version}",
"app_command_line": f"gunicorn --bind=0.0.0.0 --timeout 600 app:app"
}
elif framework.lower() == "node":
site_config = {
"linux_fx_version": f"NODE|{version}",
"app_command_line": "npm start"
}
elif framework.lower() == "php":
site_config = {
"linux_fx_version": f"PHP|{version}"
}
elif framework.lower() == "dotnetcore":
site_config = {
"linux_fx_version": f"DOTNETCORE|{version}"
}
elif framework.lower() == "java":
site_config = {
"linux_fx_version": f"JAVA|{version}"
}
else:
# Default case
site_config = {
"linux_fx_version": runtime_stack
}
# Add general configuration options
site_config.update({
"always_on": True,
"http_logging_enabled": True,
"detailed_error_logging_enabled": True,
"min_tls_version": "1.2"
})
def create_app():
# First, validate that the app service plan exists
try:
app_service_plan_info = web_client.app_service_plans.get(
resource_group_name=resource_group,
name=app_service_plan
)
except Exception as e:
raise Exception(f"App Service Plan '{app_service_plan}' not found: {str(e)}")
# Create the web app
poller = web_client.web_apps.begin_create_or_update(
resource_group_name=resource_group,
name=name,
site_envelope={
"location": location,
"server_farm_id": app_service_plan_info.id,
"site_config": site_config,
"https_only": https_only,
"client_affinity_enabled": client_affinity_enabled,
"tags": tags,
"kind": "app,linux" if os_type.lower() == "linux" else "app"
}
)
return poller.result()
web_app = await loop.run_in_executor(None, create_app)
# Build response with web app details
return {
"id": web_app.id,
"name": web_app.name,
"resource_group": resource_group,
"location": location,
"default_host_name": web_app.default_host_name,
"url": f"https://{web_app.default_host_name}",
"kind": web_app.kind,
"state": web_app.state,
"runtime_stack": runtime_stack,
"app_service_plan": app_service_plan,
"https_only": https_only
}
except Exception as e:
return {
"error": str(e),
"status": "failed",
"resource_group": resource_group,
"name": name
}
async def deploy_github_to_web_app(
self,
resource_group: str,
name: str,
repo_url: str,
branch: str = "main"
) -> Dict[str, Any]:
"""Deploy a GitHub repository to a Web App using Azure's Python API."""
web_client = self.client_factory.get_web_client()
loop = asyncio.get_event_loop()
def deploy():
deployment_operation = web_client.web_apps.begin_create_or_update_source_control(
resource_group_name=resource_group,
name=name,
site_source_control={
"repo_url": repo_url,
"branch": branch,
"is_manual_integration": True
}
)
return deployment_operation.result()
try:
deployment_result = await loop.run_in_executor(None, deploy)
details = deployment_result.as_dict() if hasattr(deployment_result, "as_dict") else str(deployment_result)
return {
"result": "Deployment triggered successfully",
"deployment_details": details,
"status": "running",
"url": f"https://{name}.azurewebsites.net"
}
except Exception as e:
return {
"result": "Deployment failed",
"error": str(e),
"status": "error"
}
async def connect_custom_domain(
self,
resource_group: str,
name: str,
custom_domain: str
) -> Dict[str, Any]:
"""Connect a custom top-level domain to an existing Web App."""
web_client = self.client_factory.get_web_client()
loop = asyncio.get_event_loop()
def connect_domain():
update_operation = web_client.web_apps.begin_update(
resource_group_name=resource_group,
name=name,
site_envelope={
"hostNames": [custom_domain]
}
)
return update_operation.result()
try:
update_result = await loop.run_in_executor(None, connect_domain)
details = update_result.as_dict() if hasattr(update_result, "as_dict") else str(update_result)
return {
"result": "Custom domain connected successfully.",
"custom_domain": custom_domain,
"details": details
}
except Exception as e:
return {
"result": "Failed to connect custom domain.",
"error": str(e)
}
async def create_app_service_plan(
self,
resource_group: str,
name: str,
location: str = "westeurope",
sku: str = "B1",
os_type: str = "Linux",
per_site_scaling: bool = False,
maximum_elastic_worker_count: int = 1,
tags: Optional[Dict[str, str]] = None
) -> Dict[str, Any]:
"""Create an App Service Plan in Azure."""
web_client = self.client_factory.get_web_client()
loop = asyncio.get_event_loop()
try:
# Parse the SKU into its components
sku_parts = sku.split("_") if "_" in sku else [sku]
tier = sku_parts[0]
# Map tier to reserved status (Linux requires reserved=True)
reserved = os_type.lower() == "linux"
def create_plan():
# Create the app service plan
poller = web_client.app_service_plans.begin_create_or_update(
resource_group_name=resource_group,
name=name,
app_service_plan={
"location": location,
"sku": {
"name": sku,
"tier": tier,
"capacity": 1 # Default to 1 instance
},
"kind": "linux" if os_type.lower() == "linux" else "windows",
"reserved": reserved, # Required for Linux plans
"per_site_scaling": per_site_scaling,
"maximum_elastic_worker_count": maximum_elastic_worker_count,
"tags": tags,
"is_spot": False, # Not using spot instances by default
}
)
return poller.result()
app_service_plan = await loop.run_in_executor(None, create_plan)
# Build response with app service plan details
return {
"id": app_service_plan.id,
"name": app_service_plan.name,
"resource_group": resource_group,
"location": location,
"sku": {
"name": app_service_plan.sku.name,
"tier": app_service_plan.sku.tier,
"capacity": app_service_plan.sku.capacity
},
"os_type": os_type,
"status": app_service_plan.status,
"per_site_scaling": per_site_scaling,
"maximum_elastic_worker_count": maximum_elastic_worker_count
}
except Exception as e:
return {
"error": str(e),
"status": "failed",
"resource_group": resource_group,
"name": name
}
async def deploy_docker_to_web_app(
self,
resource_group: str,
name: str,
image: str,
registry_url: Optional[str] = None,
registry_username: Optional[str] = None,
registry_password: Optional[str] = None,
startup_command: Optional[str] = None,
tags: Optional[Dict[str, str]] = None
) -> Dict[str, Any]:
"""Deploy a Docker container to an Azure Web App."""
web_client = self.client_factory.get_web_client()
loop = asyncio.get_event_loop()
try:
# First, get the current web app to ensure it exists
def get_webapp():
return web_client.web_apps.get(
resource_group_name=resource_group,
name=name
)
webapp = await loop.run_in_executor(None, get_webapp)
# Prepare container settings
container_settings = {
"linux_fx_version": f"DOCKER|{image}"
}
# Add startup command if provided
if startup_command:
container_settings["app_command_line"] = startup_command
# Create site config with container settings
site_config = {
**container_settings,
"always_on": True,
"http_logging_enabled": True,
"detailed_error_logging_enabled": True
}
# Prepare app settings for container registry credentials
app_settings = []
if registry_url and registry_username and registry_password:
app_settings = [
{"name": "DOCKER_REGISTRY_SERVER_URL", "value": registry_url},
{"name": "DOCKER_REGISTRY_SERVER_USERNAME", "value": registry_username},
{"name": "DOCKER_REGISTRY_SERVER_PASSWORD", "value": registry_password},
]
# Update the web app with Docker configuration
def update_webapp():
# Update site configuration with Docker container
config_poller = web_client.web_apps.begin_update_configuration(
resource_group_name=resource_group,
name=name,
site_config=site_config
)
config_result = config_poller.result()
# If we have registry credentials, update app settings
if app_settings:
settings_poller = web_client.web_apps.begin_update_application_settings(
resource_group_name=resource_group,
name=name,
app_settings_info={
"properties": {setting["name"]: setting["value"] for setting in app_settings}
}
)
settings_result = settings_poller.result()
# Restart the web app to apply changes
restart_poller = web_client.web_apps.begin_restart(
resource_group_name=resource_group,
name=name
)
restart_result = restart_poller.result()
return {
"config": config_result,
"settings": settings_result if app_settings else None,
"restart": restart_result
}
update_result = await loop.run_in_executor(None, update_webapp)
# Return success response
return {
"result": "Docker container deployed successfully",
"url": f"https://{webapp.default_host_name}",
"container_image": image,
"registry_url": registry_url,
"status": "running"
}
except Exception as e:
return {
"error": str(e),
"status": "failed",
"resource_group": resource_group,
"name": name,
"image": image
}
async def get_web_app_info(
self,
resource_group: str,
name: str
) -> Dict[str, Any]:
"""Get detailed information about a Web App."""
web_client = self.client_factory.get_web_client()
# Simulated response for checking deployment status.
return {
"name": name,
"resource_group": resource_group,
"url": f"https://{name}.azurewebsites.net",
"status": "stopped",
"message": "Web app appears to be stopped or not responding."
}
async def update_web_app_settings(
self,
resource_group: str,
name: str,
settings: Dict[str, str],
tags: Optional[Dict[str, str]] = None
) -> Dict[str, Any]:
"""Update application settings for a Web App."""
web_client = self.client_factory.get_web_client()
loop = asyncio.get_event_loop()
try:
# Update app settings
def update_settings():
# Update application settings
settings_poller = web_client.web_apps.begin_update_application_settings(
resource_group_name=resource_group,
name=name,
app_settings_info={
"properties": settings
}
)
settings_result = settings_poller.result()
# If tags are provided, update them as well
if tags:
# First get the current web app to ensure it exists
webapp = web_client.web_apps.get(
resource_group_name=resource_group,
name=name
)
# Update the tags
tags_poller = web_client.web_apps.begin_update(
resource_group_name=resource_group,
name=name,
site_envelope={
"tags": tags
}
)
tags_result = tags_poller.result()
return {
"settings": settings_result,
"tags": tags_result
}
return {
"settings": settings_result
}
update_result = await loop.run_in_executor(None, update_settings)
# Get the updated settings to confirm they were applied
def get_updated_settings():
return web_client.web_apps.list_application_settings(
resource_group_name=resource_group,
name=name
)
updated_settings = await loop.run_in_executor(None, get_updated_settings)
return {
"result": "Web app settings updated successfully",
"settings_count": len(settings),
"updated_settings": {k: v for k, v in updated_settings.properties.items() if k in settings},
"tags_updated": tags is not None,
"web_app": name,
"resource_group": resource_group
}
except Exception as e:
return {
"error": str(e),
"status": "failed",
"resource_group": resource_group,
"name": name
}
async def list_web_app_logs(
self,
resource_group: str,
name: str,
log_type: str = "application",
start_time: Optional[str] = None,
end_time: Optional[str] = None
) -> Dict[str, Any]:
"""List logs from a Web App."""
web_client = self.client_factory.get_web_client()
loop = asyncio.get_event_loop()
try:
# Validate log type
valid_log_types = ["application", "deployment", "detailed_errors", "failed_request_tracing", "web_server"]
if log_type.lower() not in valid_log_types:
return {
"error": f"Invalid log_type. Must be one of: {', '.join(valid_log_types)}",
"status": "failed"
}
# For application logs, we need to get the actual log entries
if log_type.lower() == "application":
def get_app_logs():
# Get the kudu scm site URL for accessing logs
publish_profile = web_client.web_apps.list_publishing_profile_xml_with_secrets(
resource_group_name=resource_group,
name=name
)
# Parse publishing profile to get scm URL
import xml.etree.ElementTree as ET
from io import StringIO
# Convert bytes to string if needed
if isinstance(publish_profile, bytes):
publish_profile = publish_profile.decode('utf-8')
root = ET.parse(StringIO(publish_profile))
scm_url = None
for profile in root.findall('.//publishProfile'):
if profile.get('publishMethod') == 'MSDeploy':
scm_url = profile.get('publishUrl')
username = profile.get('userName')
password = profile.get('userPWD')
break
if not scm_url:
return {
"error": "Could not find SCM URL in publishing profile",
"status": "failed"
}
# Now use the SCM URL to get logs
import requests
from datetime import datetime, timedelta
import base64
# Format: https://username:password@sitename.scm.azurewebsites.net
auth_header = f"Basic {base64.b64encode(f'{username}:{password}'.encode()).decode()}"
# Set up time parameters
if start_time:
start = start_time
else:
# Default to 1 hour ago
start = (datetime.utcnow() - timedelta(hours=1)).strftime("%Y-%m-%dT%H:%M:%SZ")
if end_time:
end = end_time
else:
# Default to now
end = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
# Construct log URL
log_url = f"https://{scm_url}/api/logs/application"
# Get logs
response = requests.get(
log_url,
headers={
"Authorization": auth_header
},
params={
"startTime": start,
"endTime": end
}
)
if response.status_code != 200:
return {
"error": f"Failed to retrieve logs: {response.text}",
"status": "failed",
"status_code": response.status_code
}
# Parse log entries
logs = response.json()
return {
"log_entries": logs,
"count": len(logs),
"start_time": start,
"end_time": end
}
logs_result = await loop.run_in_executor(None, get_app_logs)
# Add metadata to the result
return {
"web_app": name,
"resource_group": resource_group,
"log_type": log_type,
"logs": logs_result
}
# For other log types, just return the diagnostic settings
else:
def get_diagnostic_settings():
# Get web app diagnostics settings
diagnostics = web_client.web_apps.get_diagnostic_logs_configuration(
resource_group_name=resource_group,
name=name
)
return diagnostics
diagnostics = await loop.run_in_executor(None, get_diagnostic_settings)
# Extract relevant information based on log_type
log_info = {}
if log_type.lower() == "deployment":
if hasattr(diagnostics, 'application_logs') and hasattr(diagnostics.application_logs, 'file_system'):
log_info = {
"level": diagnostics.application_logs.file_system.level,
"retention_in_days": diagnostics.application_logs.file_system.retention_in_days,
"retention_in_mb": diagnostics.application_logs.file_system.retention_in_mb
}
elif log_type.lower() == "detailed_errors":
if hasattr(diagnostics, 'detailed_error_messages') and diagnostics.detailed_error_messages:
log_info = {
"enabled": diagnostics.detailed_error_messages.enabled
}
elif log_type.lower() == "failed_request_tracing":
if hasattr(diagnostics, 'failed_requests_tracing') and diagnostics.failed_requests_tracing:
log_info = {
"enabled": diagnostics.failed_requests_tracing.enabled
}
elif log_type.lower() == "web_server":
if hasattr(diagnostics, 'http_logs') and hasattr(diagnostics.http_logs, 'file_system'):
log_info = {
"retention_in_days": diagnostics.http_logs.file_system.retention_in_days,
"retention_in_mb": diagnostics.http_logs.file_system.retention_in_mb
}
return {
"web_app": name,
"resource_group": resource_group,
"log_type": log_type,
"log_settings": log_info,
"diagnostics_enabled": log_info.get("enabled", False)
if "enabled" in log_info else (log_info.get("level", "Off") != "Off")
}
except Exception as e:
return {
"error": str(e),
"status": "failed",
"resource_group": resource_group,
"name": name,
"log_type": log_type
}
async def restart_web_app(
self,
resource_group: str,
name: str,
soft_restart: bool = False
) -> Dict[str, Any]:
"""Restart a Web App using Azure API."""
web_client = self.client_factory.get_web_client()
loop = asyncio.get_event_loop()
def restart():
restart_operation = web_client.web_apps.begin_restart(
resource_group_name=resource_group,
name=name
)
return restart_operation.result()
try:
restart_result = await loop.run_in_executor(None, restart)
details = restart_result.as_dict() if hasattr(restart_result, "as_dict") else str(restart_result)
return {
"result": "Web app restarted successfully.",
"details": details
}
except Exception as e:
return {
"result": "Web app restart failed.",
"error": str(e)
}
