Skip to content

[python-skills] the generated code mixes using sync/async #290

@xiangyan99

Description

@xiangyan99

Sample:

"""
Configuration module for Azure services.
Builds connections for Blob Storage and Key Vault using shared credentials.
"""
import os
from azure.identity import DefaultAzureCredential
from azure.keyvault.keys import KeyClient
from azure.keyvault.keys.aio import KeyClient as AsyncKeyClient
from azure.storage.blob import BlobServiceClient
from azure.storage.blob.aio import BlobServiceClient as AsyncBlobServiceClient


class AzureConfig:
    """Configuration manager for Azure services with shared credentials."""
    
    def __init__(self):
        """Initialize configuration from environment variables."""
        self.vault_url = os.getenv("AZURE_KEY_VAULT_URL")
        if not self.vault_url:
            raise ValueError("AZURE_KEY_VAULT_URL environment variable must be set")
        
        self.storage_account_url = os.getenv("AZURE_STORAGE_ACCOUNT_URL")
        if not self.storage_account_url:
            raise ValueError("AZURE_STORAGE_ACCOUNT_URL environment variable must be set")
        
        self.container_name = os.getenv("AZURE_STORAGE_CONTAINER_NAME", "encrypted-files")
        self.key_name = os.getenv("AZURE_KEY_VAULT_KEY_NAME", "encryption-key")
        
        # Single shared credential instance
        self._credential = DefaultAzureCredential()
    
    @property
    def credential(self):
        """Get the shared credential instance."""
        return self._credential
    
    def get_key_client(self) -> KeyClient:
        """Create a synchronous Key Vault Key client."""
        return KeyClient(vault_url=self.vault_url, credential=self._credential)
    
    def get_async_key_client(self) -> AsyncKeyClient:
        """Create an asynchronous Key Vault Key client."""
        return AsyncKeyClient(vault_url=self.vault_url, credential=self._credential)
    
    def get_blob_service_client(self) -> BlobServiceClient:
        """Create a synchronous Blob Service client."""
        return BlobServiceClient(account_url=self.storage_account_url, credential=self._credential)
    
    def get_async_blob_service_client(self) -> AsyncBlobServiceClient:
        """Create an asynchronous Blob Service client."""
        return AsyncBlobServiceClient(account_url=self.storage_account_url, credential=self._credential)

It uses sync DAC in async client.

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions