AnalyticDB:Use the API operations of AnalyticDB for PostgreSQL to search for images

Initialize the client

All examples in this section use the same client initialization. Create the client once and reuse it across operations:

# -*- coding: utf-8 -*-
import os
from alibabacloud_gpdb20160503.client import Client as gpdb20160503Client
from alibabacloud_tea_OpenAPI import models as open_api_models
from alibabacloud_gpdb20160503 import models as gpdb_20160503_models
from alibabacloud_tea_util import models as util_models

def create_client() -> gpdb20160503Client:
    # Credentials are read from environment variables—never hardcode them.
    config = open_api_models.Config(
        access_key_id=os.environ["ALIBABA_CLOUD_ACCESS_KEY_ID"],
        access_key_secret=os.environ["ALIBABA_CLOUD_ACCESS_KEY_SECRET"]
    )
    config.endpoint = "gpdb.aliyuncs.com"
    return gpdb20160503Client(config)

Upload a single image

Use UploadDocumentAsyncAdvanceRequest for local files (pass a file object) and UploadDocumentAsyncRequest for remote images (pass a URL string). The remaining parameters are identical.

Upload a local image:

client = create_client()
with open("<image_file_path>", "rb") as f:
    # image_file_path: absolute path to the local image file
    request = gpdb_20160503_models.UploadDocumentAsyncAdvanceRequest(
        region_id="<your-instance-region-id>",
        dbinstance_id="<your-instance-id>",
        namespace="<your-namespace-name>",
        namespace_password="<your-namespace-password>",
        collection="<your-collection-name>",
        file_name="<filename-with-extension>",  # e.g., photo.jpg
        file_url_object=f,
        dry_run=False,
        metadata={"caption": "sample image", "category": "nature"},  # dict format
    )
    runtime = util_models.RuntimeOptions()
    try:
        response = client.upload_document_async_advance(request, runtime)
        print("Job ID:", response.body.job_id)
    except Exception as error:
        print(error)

Upload a remote image:

client = create_client()
request = gpdb_20160503_models.UploadDocumentAsyncRequest(
    region_id="<your-instance-region-id>",
    dbinstance_id="<your-instance-id>",
    namespace="<your-namespace-name>",
    namespace_password="<your-namespace-password>",
    collection="<your-collection-name>",
    file_name="<filename-with-extension>",  # e.g., photo.jpg
    file_url="<image_file_url>",            # publicly accessible URL
    dry_run=False,
    metadata={"caption": "sample image", "category": "nature"},  # dict format
)
runtime = util_models.RuntimeOptions()
try:
    response = client.upload_document_async_with_options(request, runtime)
    print("Job ID:", response.body.job_id)
except Exception as error:
    print(error)

Parameter reference:

Upload multiple images

To upload multiple images at once, pack them into a compressed archive and upload the archive. The UploadDocumentAsync operation extracts and processes each image in the archive.

client = create_client()
with open("<compress_file_path>", "rb") as f:
    # compress_file_path: absolute path to the local archive (.tar, .gz, or .zip)
    request = gpdb_20160503_models.UploadDocumentAsyncAdvanceRequest(
        region_id="<your-instance-region-id>",
        dbinstance_id="<your-instance-id>",
        namespace="<your-namespace-name>",
        namespace_password="<your-namespace-password>",
        collection="<your-collection-name>",
        file_name="<archive-filename-with-extension>",  # e.g., images.zip
        file_url_object=f,
        dry_run=False,
        metadata={"batch": "upload-batch-1"},
    )
    runtime = util_models.RuntimeOptions()
    try:
        response = client.upload_document_async_advance(request, runtime)
        print("Job ID:", response.body.job_id)
    except Exception as error:
        print(error)

Check upload progress

UploadDocumentAsync is asynchronous. Poll GetUploadDocumentJob with the job ID until the status is Success.

client = create_client()
request = gpdb_20160503_models.GetUploadDocumentJobRequest(
    region_id="<your-instance-region-id>",
    dbinstance_id="<your-instance-id>",
    namespace="<your-namespace-name>",
    namespace_password="<your-namespace-password>",
    collection="<your-collection-name>",
    job_id="<job_id>",  # job_id returned by UploadDocumentAsync
)
runtime = util_models.RuntimeOptions()
try:
    response = client.get_upload_document_job_with_options(request, runtime)
    print("Status:", response.body.job.status)
except Exception as error:
    print(error)

When job.status returns Success, all images in the upload job are indexed and ready for search. For more information, see GetUploadDocumentJob.

Search by text

QueryContent accepts a text string, converts it into a feature vector, and returns the top-k most similar images.

# -*- coding: utf-8 -*-
import os
from urllib.request import urlopen
from PIL import Image
from alibabacloud_gpdb20160503.client import Client as gpdb20160503Client
from alibabacloud_tea_OpenAPI import models as open_api_models
from alibabacloud_gpdb20160503 import models as gpdb_20160503_models
from alibabacloud_tea_util import models as util_models

client = create_client()  # create_client() defined in the "Initialize the client" section

request = gpdb_20160503_models.QueryContentRequest(
    region_id="<your-instance-region-id>",
    dbinstance_id="<your-instance-id>",
    namespace="<your-namespace-name>",
    namespace_password="<your-namespace-password>",
    collection="<your-collection-name>",
    content="Dog",  # the text query
    top_k=3,
)
runtime = util_models.RuntimeOptions()
try:
    response = client.query_content_with_options(request, runtime)
    if response.status_code != 200:
        raise Exception(f"QueryContent failed: {response.body}")

    for match in response.body.matches.match_list:
        url = match.file_url
        caption = match.metadata.get("caption")
        print(f"URL: {url}, Caption: {caption}")
        Image.open(urlopen(url)).show()
except Exception as error:
    print(error)

Each item in match_list has a file_url pointing to the matched image and a metadata dict containing fields you set during upload (such as caption or category).

Parameter reference:

Sample output (querying "Dog"):

Results vary based on the images in your collection.

Search by image

To search using a local image as the query, use QueryContentAdvanceRequest with a file object and filename.

# -*- coding: utf-8 -*-
import os
from urllib.request import urlopen
from PIL import Image
from alibabacloud_gpdb20160503.client import Client as gpdb20160503Client
from alibabacloud_tea_OpenAPI import models as open_api_models
from alibabacloud_gpdb20160503 import models as gpdb_20160503_models
from alibabacloud_tea_util import models as util_models

client = create_client()  # create_client() defined in the "Initialize the client" section

query_file_path = "<image_file_path>"  # absolute path to the query image
with open(query_file_path, "rb") as f:
    request = gpdb_20160503_models.QueryContentAdvanceRequest(
        region_id="<your-instance-region-id>",
        dbinstance_id="<your-instance-id>",
        namespace="<your-namespace-name>",
        namespace_password="<your-namespace-password>",
        collection="<your-collection-name>",
        file_url_object=f,
        file_name=os.path.basename(query_file_path),
        top_k=3,
    )
    runtime = util_models.RuntimeOptions()
    try:
        response = client.query_content_advance(request, runtime)
        if response.status_code != 200:
            raise Exception(f"QueryContent failed: {response.body}")

        for match in response.body.matches.match_list:
            url = match.file_url
            caption = match.metadata.get("caption")
            print(f"URL: {url}, Caption: {caption}")
            Image.open(urlopen(url)).show()
    except Exception as error:
        print(error)

Parameter reference:

Sample output (querying with a bicycle image):

Results vary based on the images in your collection.