Store and load products

In this step you implement the two simplest data-access methods on KeyValueServiceNewClient. The store_product method writes a Product to a set, and get_product reads one back by its key. Before you implement them, you walk through the connect() method that wires up the SDK’s ClusterDefinition, Behavior, and Session.

Open the file

1. Stop the FastAPI application from the previous step with Ctrl+C. Leave Voyager open and connected.

2. Open python-server/src/aerospikeworkshop/services/key_value_service_new_client.py in your IDE.

   Every code change in this section, and in the next section, is made in this single file.

Walk through the SDK connection code

The SDK connection code is already written for you in the connect() method. Find it under the TODO: STEP 1: VALIDATE THE CONNECTION header near the top of the class:

async def connect(self) -> None:
    # TODO: STEP 1: VALIDATE THE CONNECTION
    cluster_def = ClusterDefinition(
        self._settings.aerospike_host, self._settings.aerospike_port
    )
    if self._settings.aerospike_username:
        cluster_def = cluster_def.with_native_credentials(
            self._settings.aerospike_username,
            self._settings.aerospike_password or "",
        )
    self._cluster = await cluster_def.connect()
    self._session = self._cluster.create_session(Behavior.DEFAULT)

You do not change this code in this tutorial, but reading it top to bottom helps you understand the SDK pieces you use later:

• ClusterDefinition describes how to find and authenticate against your cluster. It accepts the seed hostname and port, plus optional credentials, TLS settings, and other connection options. The hostname and port come from environment variables (AEROSPIKE_PORT=3000 for the local Docker install).
• await cluster_def.connect() opens the cluster connection and returns a Cluster object that owns the underlying network resources.
• A Cluster cannot run data operations on its own. The SDK requires a Behavior, which centralizes timeouts, retry policy, replica selection, and other per-call settings in one place. If you have used the legacy Aerospike Python client, Behavior replaces the per-method policy objects.
• create_session(Behavior.DEFAULT) returns a Session that you use for inserts, queries, and updates. This tutorial uses Behavior.DEFAULT to keep the focus on the data API.

FastAPI calls connect() during application startup. You use self._session (with _require_session()) for every database call from this point on.

Implement store_product

The auto-loader calls store_product(product) once per record in the data/styles folder. The method body is a pass placeholder.

1. Find the store_product method.

   async def store_product(self, product: Product) -> None:
       # TODO: STEP 2: STORE A PRODUCT OBJECT
       pass

2. Replace pass with the SDK call.

   session = self._require_session()
   await (
       session.insert(self.product_dataset.id(product.id))
       .put(product.to_bins())
       .execute()
   )

   self._require_session() returns the Session that connect() created at startup, or raises if the application is not connected yet.

   This is the SDK’s fluent style: each method call adds one piece of the operation to a builder, and the chain reads top to bottom:

   • insert writes the record and fails if a record with the same key already exists. The other write verbs include update (fails if the record does not exist) and upsert (writes the record whether or not it exists). The corresponding read verb, query, is what you use in get_product later in this step.
   • self.product_dataset.id(product.id) addresses one record in the test.products set. A DataSet is a (namespace, set) pair.
   • put(product.to_bins()) passes the product as a dictionary of bin names and values. The workshop’s Product model provides to_bins() and from_bins() helpers for this conversion.
   • await ...execute() is the terminal call that ships the operation to the cluster. Every fluent chain in the SDK ends with a terminal call. If you forget it, the chain only describes the operation and no request reaches the cluster.

Confirm store_product works

Save your changes to key_value_service_new_client.py, then restart the server so auto-load calls your new store_product implementation. Uvicorn does not reload edited Python files unless you started it with --reload.

1. Stop the FastAPI application if it is still running.

   In the terminal where the application is running, press Ctrl+C and wait for the shell prompt to return.

   If the previous process is no longer running but port 8080 is still bound, find the holding process with lsof -nP -iTCP:8080 -sTCP:LISTEN and stop it with kill <PID>. Replace <PID> with the process ID from the lsof output.

2. Change into the python-server directory if you are not already there.

   cd python-server

   Every command in this tutorial that starts the server runs from aerospike-client-sdk-workshop/python-server. If you open a new terminal between steps, activate your virtual environment (if you use one), then run this cd first.

3. Restart the application with the new-client profile.

   AEROSPIKE_CLIENT_PROFILE=new-client AEROSPIKE_PORT=3000 \
     uvicorn aerospikeworkshop.main:app --host 0.0.0.0 --port 8080

   If the database already contains product records from a previous attempt, clear them first from another terminal:

   curl -X DELETE "http://localhost:8080/rest/v1/data/clear?confirm=yes-delete-all"

   Then restart the server so the auto-loader runs again. The auto-load line in the logs should now report 200 products in products set (along with processed 200/200 sample files).

4. In Voyager, select Refresh at the top of the cluster details panel.

   The test namespace now reports 200 records.

   

5. Drill into the products set to confirm that the records contain product data.

   

Tip

The fluent verb / execute() pattern reads like a sentence, so AI coding assistants in your IDE can often suggest the rest of a chain after you type the first call. Type each chain yourself the first time to understand what every part does, then accept the assistant’s suggestions on later methods if you find them helpful.

Implement get_product

The get_product(product_id) method returns a Product if one exists, or None if not.

1. Find the get_product method. The body is a single line, return None.

2. Replace return None with the following implementation:

   session = self._require_session()
   stream = await session.query(self.product_dataset.id(product_id)).execute()
   result = await stream.first()
   stream.close()
   if result is None or not result.is_ok:
       return None
   return Product.from_bins(result.record.bins)

   The implementation has four SDK calls plus a result guard:

   • session.query(self.product_dataset.id(product_id)) — In the Python SDK, every read goes through query(), whether you fetch one record by key or scan a set with a filter. Passing .id(product_id) limits the query to a single key (a point read). execute() still returns a result stream that holds at most one row.
   • await ...execute() ships the operation and returns that result stream.
   • await stream.first() reads at most one record from the stream.
   • stream.close() releases stream resources. This call is synchronous (do not await it). Always close streams when you finish with them, including after a point read.
   • Product.from_bins(result.record.bins) converts the bin dictionary back into a Product.

   Before you call from_bins, the if line checks two conditions. You reuse this guard on every point read in this tutorial:

   • result is None — the stream had no row (stream.first() on an empty result).
   • not result.is_ok — the read failed or returned an error status.

Confirm get_product works

1. Stop the FastAPI application if it is still running, then restart it from the python-server directory.

   cd python-server
   AEROSPIKE_CLIENT_PROFILE=new-client AEROSPIKE_PORT=3000 \
     uvicorn aerospikeworkshop.main:app --host 0.0.0.0 --port 8080

2. In your browser, reload http://localhost:8080.

   The home page is still empty because you implement the listing query in the next step, but you can confirm that point reads work by visiting a product detail page directly.

3. In Voyager, open the products set and copy any value from the Key column at the start of a row. This is the record key you query against.

4. In your browser, visit the product detail URL, replacing PRODUCT_ID with the key you copied:

   http://localhost:8080/product/PRODUCT_ID

   The page renders with the product’s details. If you visit a URL with a key that does not exist (for example, append xyz to a real key), the page renders an Oops... looks like that page doesn't exist screen instead of throwing.

Outcomes

You now have:

• Two hundred product records in the test.products set.
• The ability to look up any product by key from the running application.
• A working understanding of ClusterDefinition, Behavior, Session, and the SDK’s fluent verb / execute() pattern.

You cannot yet:

• See products listed on the home page or in category dropdowns.
• Filter products by category, article type, usage, or brand.

In the next step you implement the secondary-index query that powers product listings, then build a multi-condition query for the advanced search filters.

Checklist

• I understand how the application connects to Aerospike with ClusterDefinition, Behavior, and Session.
• I’ve implemented store_product and confirmed all 200 records are written.
• I’ve implemented get_product and confirmed product detail pages load.