Skip to main content

Download notebook · View source · Open in Colab

Work with Audio

ApertureDB stores audio files as Blobs with metadata such as dish name, contributor, and cuisine. Queries filter that metadata and can return the audio bytes. This example stores, retrieves, updates, and deletes one audio blob; it does not generate embeddings or perform vector search. A unique run marker limits cleanup to the example audio.

The server computes the read-only _mime_type MIME property from the blob bytes (for this example, audio/wav). To override detection, pass mime_type directly in AddBlob; do not include _mime_type in AddBlob.properties.

Connect to ApertureDB​

ApertureDB Cloud The SDK connects with an API key from your Cloud instance. Obtain the key from Connect > Generate API Key and add it to a .env file in this directory:

APERTUREDB_KEY=your_key_here

Local evaluation: follow the deployment guide, then configure your connection profile.

See client configuration options for all connection methods and server setup options for deployment choices.

%pip install --quiet "aperturedb==0.5.1" "python-dotenv==1.1.1"
# Option A: ApertureDB Cloud
from dotenv import load_dotenv
load_dotenv()  # loads APERTUREDB_KEY from .env into the environment
# For local evaluation, configure the profile from your versioned deployment package.
# https://docs.aperturedata.io/administration/deployment/DockerCompose
from aperturedb.CommonLibrary import create_connector
from uuid import uuid4

client = create_connector()
run_id = "audio-notebook-" + uuid4().hex
scope = ["$documentation_run", "==", run_id]

def checked_query(query, blobs=None):
    response, output_blobs = client.query(query, blobs or [])
    assert len(response) == len(query), response
    for result in response:
        assert next(iter(result.values()))["status"] == 0, response
    return response, output_blobs

response, _ = checked_query([{"GetStatus": {}}])
print("Connected to ApertureDB", response[0]["GetStatus"]["version"])

Generate a Sample Audio Clip​

Python's built-in wave module generates a short WAV file containing a timer beep. An existing .wav, .mp3, or other audio file can be stored in the same way.

import wave, struct, math, os

os.makedirs("data", exist_ok=True)

sample_rate = 44100
duration    = 2       # seconds
frequency   = 880     # Hz — a short timer beep

# Apply a short fade-out to avoid clipping
samples = []
total = sample_rate * duration
for t in range(total):
    fade = max(0.0, 1.0 - t / total)  
    samples.append(int(32767 * fade * math.sin(2 * math.pi * frequency * t / sample_rate)))

with wave.open("data/cooking_timer.wav", "w") as f:
    f.setnchannels(1)
    f.setsampwidth(2)
    f.setframerate(sample_rate)
    f.writeframes(struct.pack("<" + "h" * len(samples), *samples))

print(f"Generated data/cooking_timer.wav ({os.path.getsize('data/cooking_timer.wav')} bytes)")

from pathlib import Path
import hashlib

with wave.open("data/cooking_timer.wav", "rb") as clip:
    assert (clip.getnchannels(), clip.getsampwidth(), clip.getframerate(), clip.getnframes()) == (1, 2, sample_rate, total)
fixture_sha256 = hashlib.sha256(Path("data/cooking_timer.wav").read_bytes()).hexdigest()
print("WAV SHA-256:", fixture_sha256)

Preview the audio clip before storing it:

from IPython.display import Audio
Audio("data/cooking_timer.wav")

Add an Audio Blob to ApertureDB​

AddBlob stores the audio bytes with metadata. The type: audio property distinguishes audio records in later filters.

The Python SDK loaders handle bulk ingestion.

query = [{
    "AddBlob": {
        "properties": {
            "documentation_run": run_id,
            "type":        "audio",
            "format":      "wav",
            "name":        "cooking_timer",
            "dish_name":   "Baked Potato",
            "cuisine":     "American",
            "contributor": "Gavin",
            "duration_s":  2,
        },
        "if_not_found": ['all', scope, ['$name', '==', 'cooking_timer']],
    }
}]

with open("data/cooking_timer.wav", "rb") as f:
    audio_bytes = f.read()

response, _ = checked_query(query, [audio_bytes])
client.print_last_response()

Find Audio Blobs by Metadata​

FindBlob selects audio records by their metadata. blobs: true includes the audio bytes in the response.

query = [{
    "FindBlob": {
        "constraints": ['all', scope, ['$type', '==', 'audio'], ['$cuisine', '==', 'American']],
        "blobs": True,
        "results": {"all_properties": True},
    }
}]

response, blobs = checked_query(query)
client.print_last_response()

assert response[0]["FindBlob"]["returned"] == 1, response
assert response[0]["FindBlob"]["entities"][0]["duration_s"] == duration
assert blobs == [audio_bytes], "Retrieved audio must exactly match the uploaded WAV."

Play back the retrieved audio:

from IPython.display import Audio, display

display(Audio(blobs[0]))

Update Audio Blob Metadata​

UpdateBlob changes existing metadata values or adds properties to matching audio blobs.

query = [{
    "UpdateBlob": {
        "constraints": ['all', scope, ['$name', '==', 'cooking_timer']],
        "properties":  {"verified": True},
    }
}]

response, _ = checked_query(query)
client.print_last_response()

response, _ = checked_query([{"FindBlob": {
    "constraints": scope, "blobs": False, "results": {"list": ["verified"]}
}}])
assert response[0]["FindBlob"]["returned"] == 1, response
assert response[0]["FindBlob"]["entities"][0]["verified"] is True, response

Delete the Audio Blob​

The cleanup query deletes only this run's audio blob. The following count query returns zero matches.

query = [{
    "DeleteBlob": {
        "constraints": scope,
    }
}]

response, _ = checked_query(query)
client.print_last_response()

response, _ = checked_query([{"FindBlob": {
    "constraints": scope, "blobs": False, "results": {"count": True}
}}])
assert response[0]["FindBlob"]["count"] == 0, response
print("Verified audio byte equality, metadata update, and scoped cleanup.")

What's Next?​