def upload_sample_documents()

in demo-python/code/community-integration/ragas/lib/utils.py [0:0]

• 16 lines of code
• 5 McCabe index (conditional complexity)

def upload_sample_documents(
        blob_connection_string: str,
        blob_container_name: str,
        use_user_identity: bool = True
    ):
    # Connect to Blob Storage
    blob_service_client = BlobServiceClient.from_connection_string(conn_str=blob_connection_string, credential=DefaultAzureCredential() if use_user_identity else None)
    container_client = blob_service_client.get_container_client(blob_container_name)
    if not container_client.exists():
        container_client.create_container()

    documents_directory = os.path.join("..", "..", "..", "..", "data", "benefitdocs")
    pdf_files = glob.glob(os.path.join(documents_directory, '*.pdf'))
    for file in pdf_files:
        with open(file, "rb") as data:
            name = os.path.basename(file)
            if not container_client.get_blob_client(name).exists():
                container_client.upload_blob(name=name, data=data)