Our assistant's knowledge base is stored in Azure Blob Storage as a document set. Accordingly, to work with it, we need an API that allows us to perform the following operations:

Get a list of documents from a specified Azure Blob Storage container
Add files to the container
Delete the files from the container

In Python, Azure Blob Storage is handled by the azure.storage.blob library, which contains the storage client. Accordingly, we need to add a link to it in our application and import this client:

from azure.storage.blob import BlobServiceClient

! Note. Before using the client, it must be initialized, i.e., at least, you should specify a string for connecting to the storage. Since we will initialize each time any of our three possible storage operations are called, it makes sense to formalize it as a separate function:

def init_storage_client(): 
    
if ( 
        not app_settings.storage.connection_string and 
        not app_settings.storage.container 
    ): 

        raise ValueError( 
            "AZURE_STORAGE_CONNECTION_STRING or AZURE_STORAGE_CONTAINER is required" 
        ) 

    blob_service_client  = BlobServiceClient.from_connection_string( app_settings.storage.connection_string) 

    return blob_service_client

As you can see, our initialization uses two new environment variables, AZURE_STORAGE_CONNECTION_STRING and AZURE_STORAGE_CONTAINER, which are not in the .env file. Hence, we need to add them to the .env file, specifying the storage connection string and the container's name in which the knowledge base documents are stored. Next, we need to add the values of these variables to the application's global settings object.

! Note. Variables with fields of the settings object are mapped automatically, but you need to create the fields themselves. This process is done in the backend/settings.py file. Create a new class in it for the Azure Blob Storage configuration group:

class _AzureSearchSettings(_BaseAzureSearchSettings):
    model_config = SettingsConfigDict( 
        env_prefix="AZURE_SEARCH_", 
        env_file=DOTENV_PATH, 
        extra="ignore", 
        env_ignore_empty=True 
    )

Then, add this group to the main settings object:

class _AppSettings(BaseModel):
    base_settings: _BaseSettings = _BaseSettings()
    azure_openai: _AzureOpenAISettings = _AzureOpenAISettings()
    search: _SearchCommonSettings = _SearchCommonSettings()
    storage: _StorageSettings = _StorageSettings()
    msgraph: _MSGraphSettings = _MSGraphSettings()
    ui: Optional[_UiSettings] = _UiSettings()

Now, you can move on to implementing the basic features of Azure Blob Storage. Here is the function to add the files to the storage:

@bp.route('/files', methods=['POST']) 
async def upload_file():   
    files = await request.files
 
    file = files['file']

    if file is None: 
        return jsonify({"error": "No file part in the request"}), 400
 
    if file.filename == '': 
        return jsonify({"error": "No file selected for uploading"}), 400
 
    filename = secure_filename(file.filename) 
    success = await upload_file_to_blob(file, filename)
 
    if success: 
        return jsonify({"message": "File successfully uploaded"}), 200 
    else: 
        return jsonify({"error": "Failed to upload file"}), 500
 
async def upload_file_to_blob(file, filename): 
    blob_service_client = init_storage_client() 
    blob_client = blob_service_client.get_blob_client( container=app_settings.storage.container, blob=filename) 

    try: 
        blob_client.upload_blob(file, overwrite=True) 
        return True 
    except Exception as e: 
        print(f"Failed to upload to blob storage: {e}") 

return False

Here is the function to delete the files from the storage:

 @bp.route('/files', methods=['DELETE']) 
async def delete_file(): 
    try: 
        # Get the blob name from the request 
        blob_name = request.args.get('file_name')

        if not blob_name: 
            return jsonify({"error": "File name is required"}), 400        

        blob_service_client = init_storage_client()         

        # Get the container client 
        container_client = blob_service_client.get_container_client( app_settings.storage.container)     

        blob_client = container_client.get_blob_client(blob_name) 

        # Delete the blob 
        blob_client.delete_blob() 

        update_search_indexer()         

        return jsonify({"message": f"File '{blob_name}' deleted successfully"}), 200 

    except Exception as e: 
        return jsonify({"error": str(e)}), 500

Finally, here is the function to get a list of the files:

@bp.route('/files', methods=['GET'])
async def list_files():
    try:
        blob_service_client = init_storage_client()
        container_client = blob_service_client.get_container_client(app_settings.storage.container)
        # List all blobs in the container
        blob_list = container_client.list_blobs()

        blobs = []         

        for blob in blob_list:
            blob_details = {
                'name': blob.name,
                'size': blob.size,
                'last_modified': blob.last_modified.strftime('%Y-%m-%d %H:%M:%S') if blob.last_modified else None
            }
            blobs.append(blob_details)  

        # Return the list of blobs as JSON response
        return jsonify({'files': blobs}), 200

    except Exception as e:
        return jsonify({'error': str(e)}), 500