def get_all_files()

in code/embedding-function/utilities/helpers/azure_blob_storage_client.py [0:0]

• 49 lines of code
• 8 McCabe index (conditional complexity)

    def get_all_files(self):
        # Get all files in the container from Azure Blob Storage
        container_client = self.blob_service_client.get_container_client(
            self.container_name
        )
        blob_list = container_client.list_blobs(include="metadata")
        # sas = generate_blob_sas(account_name, container_name, blob.name,account_key=account_key,  permission="r", expiry=datetime.utcnow() + timedelta(hours=3))
        sas = generate_container_sas(
            self.account_name,
            self.container_name,
            user_delegation_key=self.user_delegation_key,
            account_key=self.account_key,
            permission="r",
            expiry=datetime.utcnow() + timedelta(hours=3),
        )
        files = []
        converted_files = {}
        for blob in blob_list:
            if not blob.name.startswith("converted/"):
                files.append(
                    {
                        "filename": blob.name,
                        "converted": (
                            blob.metadata.get("converted", "false") == "true"
                            if blob.metadata
                            else False
                        ),
                        "embeddings_added": (
                            blob.metadata.get("embeddings_added", "false") == "true"
                            if blob.metadata
                            else False
                        ),
                        "fullpath": f"{self.endpoint}{self.container_name}/{blob.name}?{sas}",
                        "converted_filename": (
                            blob.metadata.get("converted_filename", "")
                            if blob.metadata
                            else ""
                        ),
                        "converted_path": "",
                    }
                )
            else:
                converted_files[blob.name] = (
                    f"{self.endpoint}{self.container_name}/{blob.name}?{sas}"
                )

        for file in files:
            converted_filename = file.pop("converted_filename", "")
            if converted_filename in converted_files:
                file["converted"] = True
                file["converted_path"] = converted_files[converted_filename]

        return files