Image Intelligence Pipeline on Azure

Analyze images with Azure AI Vision and store results in a SQL database.

Introduction

30 Second Summary

Every day, apps analyze millions of images to sort your photos, detect objects in security footage, and read text from street signs. Behind the scenes, cloud AI services do this heavy lifting through simple API calls.

In this project, you will build a Python pipeline that sends images to Azure AI Vision for analysis, then stores the structured results in a cloud SQL database you can query for patterns.

What You'll Build

A Python-powered image intelligence pipeline that connects Azure AI Vision's computer vision models to an Azure SQL Database for persistent, queryable storage.

By the end of this project, you'll have:

  • A Python script that connects to Azure AI Vision and extracts captions, tags, objects, and text from any image URL.
  • An Azure SQL Database with a schema designed to store image analysis results.
  • SQL queries that mine your analysis data for patterns like "which images contain people?" and "what are the most common tags?".
  • Secret Mission: Add a batch processing feature that analyzes multiple images and generates a summary report of the most frequently detected objects and tags.

Are there any prerequisites?

This project assumes basic Python knowledge (variables, functions, loops) and basic SQL knowledge (SELECT, INSERT, CREATE TABLE). You do not need prior experience with Azure or computer vision.

Create the Azure AI Vision Resource

Your Python script needs an AI service to look at images and describe what it sees. Azure AI Vision is Microsoft's cloud-based computer vision service that analyzes images and returns structured data like captions, tags, detected objects, and extracted text.

In this step, you'll create a Computer Vision resource in the Azure Portal. This gives your script an API endpoint and key to send images for analysis.

In this step, get ready to:
  • Create a Computer Vision resource with the Free (F0) pricing tier.
  • Copy the endpoint URL and API key for your Python script.
Create the Computer Vision resource
  • Open your browser and go to portal.azure.com.
  • Sign in with your Azure account.
  • Click Create a resource (the "+" icon in the top-left area of the portal).
  • In the search bar, type Computer Vision and press Enter.
  • Select Computer Vision from the results.
  • Click Create.

You'll now fill in the resource details.

  • Select your Subscription from the dropdown.
  • Under Resource group, click Create new and enter a name like image-pipeline-rg.
  • For Region, select East US.
  • Enter a unique name for your resource (e.g. yourname-vision).
  • Under Pricing tier, select Free F0.
  • Check the Responsible AI Notice acknowledgment box.
  • Click Review + create.
  • Review your settings, then click Create.

The deployment takes about 30 seconds. Once it completes, click Go to resource.

What is the F0 tier?

The F0 tier is Azure AI Vision's free pricing tier. It gives you 5,000 free transactions per month, which is more than enough for this project.

You won't be charged anything as long as you stay within that limit.

Why East US?

Some Azure AI Vision features (like captions) have full support in specific regions. East US is a safe choice that supports all the features your pipeline uses.

Don't see the Free F0 tier?

Azure only allows one F0 Computer Vision resource per subscription. If you already have one from a previous project, you'll need to delete it first or use that existing resource instead.

help me find or delete my existing Computer Vision resource

Copy the endpoint URL and key

Your Python script needs two pieces of information to authenticate with this service: the endpoint URL and an API key.

  • On your Computer Vision resource page, click Keys and Endpoint in the left sidebar under Resource Management.
  • Copy the Endpoint value and paste it somewhere safe (like a text file or sticky note).
  • Click Show next to KEY 1, then copy the key value and paste it in the same place.

What are these values for?

The endpoint is the URL your script sends image analysis requests to. It looks something like https://yourname-vision.cognitiveservices.azure.com/.

The key proves your script is authorized to use this resource. You'll store both values in a .env file in the next step so your code can access them securely.

Don't see the Keys and Endpoint page?

Make sure your deployment is complete. If you navigated away, search for your resource name in the portal search bar at the top of the page.

help me find my Computer Vision resource in Azure Portal

Your AI Vision resource is live and ready to receive images. Next up, you'll create the Azure SQL Database where your pipeline will store analysis results.

Create the Azure SQL Database

Your AI Vision resource is ready to analyze images. Now you need somewhere to store the analysis results so you can query them later.

In this step, you'll create an Azure SQL Database using Azure's free offer. This gives you a fully managed cloud database that your Python script will write results to.

In this step, get ready to:
  • Create an Azure SQL Database with the free offer.
  • Configure the server with SQL authentication.
  • Add your local IP address to the server's firewall rules.
Create a new Azure SQL Database using the free offer

Azure SQL Database's free offer gives you 100,000 vCore seconds of compute per month for the lifetime of your subscription. That is more than enough for this project.

  • In the Azure Portal search bar at the top, type SQL databases and select SQL databases from the results.
  • Click + Create at the top left.
  • Select SQL database from the dropdown.

Where is the free offer?

Look for a banner at the top of the Create SQL Database page that says Free offer applied! or a Start free link in the creation pane. If you don't see it, try clicking the Server dropdown and re-selecting your server. This sometimes triggers the banner to appear.

  • On the Basics tab, under Project details, select your subscription.
  • For Resource group, select the same resource group you used for your Computer Vision resource (or click Create new if you need one).
  • For Database name, enter a name like image-intelligence-db.
  • For Server, click Create new. This opens the New server form.
Configure the server with SQL authentication

A logical server is the gateway that manages connections and authentication for your databases. You need to set up an admin account that your Python script will use to connect.

  • For Server name, enter a globally unique name (for example, imgpipeline-yourname). Azure will tell you if the name is already taken.
  • For Location, select East US from the dropdown.
  • For Authentication method, select Use SQL authentication.
  • Enter a Server admin login (for example, sqladmin). Write this down. You will need it later.
  • Enter a strong Password and confirm it. Write this down too.
  • Click OK to close the server form.

Keep these credentials safe

Write down your server name, admin username, and password somewhere secure. You will add these to a .env file in a later step. The server hostname follows the format your-server-name.database.windows.net.

  • Back on the Basics tab, leave Want to use SQL elastic pool set to No.
  • Click Next: Networking at the bottom of the page.
Add your local IP address to the server's firewall rules

Azure SQL Database blocks all external connections by default. You need to allow your computer's IP address through the firewall so your Python script can reach the database.

  • On the Networking tab, for Connectivity method, select Public endpoint.
  • Under Firewall rules, set Add current client IP address to Yes.
  • Leave Allow Azure services and resources to access this server set to No.
  • Click Next: Security. Leave the defaults on this page.
  • Click Next: Additional settings. Leave the defaults on this page as well.
  • Click Review + create.
  • Review the summary and click Create.

What is a firewall rule?

Azure SQL Database uses IP-based firewall rules at the server level. By setting Add current client IP address to Yes, you create a rule that allows connections from the IP address you are currently using. Without this, your Python script would get a connection refused error.

Deployment takes 1-2 minutes. Wait for the deployment to complete, then click Go to resource.

  • On the database Overview page, locate the Server name field. It shows the fully qualified hostname (for example, imgpipeline-yourname.database.windows.net).
  • Copy and save this server hostname. You will need it along with the database name, admin username, and password in the next step.

Can't connect later?

If your IP address changes (common on home Wi-Fi), you will need to update the firewall rule. Navigate to your SQL server in the portal, select Networking under Security in the left menu, click Add your client IP, and click Save. Changes can take up to 5 minutes to take effect.

help me troubleshoot my Azure SQL Database firewall connection

Your cloud database is live and your firewall is configured. Next up, you will build the Python script that connects to Azure AI Vision and starts analyzing images.

Build the Image Analysis Script

You have Azure AI Vision and a SQL Database ready to go in the cloud. Now you need a Python script that can actually talk to them.

In this step, you will create the project files and write a function that sends an image URL to Azure AI Vision and receives structured analysis results. Think captions, tags, detected objects, and extracted text.

In this step, get ready to:
  • Create the project folder with a .env credentials file and a pipeline.py script.
  • Write the analyze_image function that extracts captions, tags, objects, and OCR text from any image URL.
  • Test the full script against a sample image and see AI-generated results printed to your terminal.
Create the project structure

Your project needs two files. A .env file holds all your Azure credentials. A pipeline.py file holds the Python code that uses them.

  • Open the terminal in VS Code by pressing Ctrl+` (macOS and Windows).
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir image-pipeline
cd image-pipeline
  • Create the two project files by running:
touch .env pipeline.py
  • Confirm the files exist by running:
ls -a

You should see .env and pipeline.py in the output.

  • Open the image-pipeline folder in VS Code by clicking File in the top menu bar.
  • Click Open Folder.
  • Select the image-pipeline folder on your Desktop, then click Open.

Now add your Azure credentials to the .env file.

  • Open .env in the VS Code Explorer sidebar and paste the following configuration:
VISION_ENDPOINT=your-vision-endpoint-here
VISION_KEY=your-vision-key-here
SQL_SERVER=your-server-name.database.windows.net
SQL_DATABASE=your-database-name
SQL_USERNAME=your-admin-username
SQL_PASSWORD=your-admin-password
  • Replace each placeholder with the real values you noted from the Azure Portal in previous steps.
  • For VISION_ENDPOINT and VISION_KEY: use the values from your Computer Vision resource's Keys and Endpoint page.
  • For SQL_SERVER: use your server hostname (e.g. yourname.database.windows.net).
  • For SQL_DATABASE: use the database name you chose when creating it.
  • For SQL_USERNAME and SQL_PASSWORD: use the SQL admin credentials you set during server creation.

What does this file do?

The .env file stores sensitive credentials outside your code. The python-dotenv package loads these values into environment variables at runtime so your script can read them without hardcoding secrets.

  • Save .env.

Next, set up the imports and configuration in pipeline.py.

  • Open pipeline.py in the editor and paste the following code:
import os
import pyodbc
from datetime import datetime
from dotenv import load_dotenv
from azure.ai.vision.imageanalysis import ImageAnalysisClient
from azure.ai.vision.imageanalysis.models import VisualFeatures
from azure.core.credentials import AzureKeyCredential

# Load environment variables from .env file
load_dotenv()

# Azure AI Vision configuration
VISION_ENDPOINT = os.environ["VISION_ENDPOINT"]
VISION_KEY = os.environ["VISION_KEY"]

# Azure SQL Database configuration
SQL_SERVER = os.environ["SQL_SERVER"]
SQL_DATABASE = os.environ["SQL_DATABASE"]
SQL_USERNAME = os.environ["SQL_USERNAME"]
SQL_PASSWORD = os.environ["SQL_PASSWORD"]

What does this code do?

The import block pulls in the Azure AI Vision SDK, the pyodbc database driver, and python-dotenv for loading credentials.

The load_dotenv() call reads your .env file and makes each line available as an environment variable. Then os.environ["VISION_ENDPOINT"] retrieves those values so the rest of the script can use them.

  • Save pipeline.py.

Seeing a red underline on the imports?

If VS Code shows import errors, make sure you installed the Python packages in a previous step. You can verify by running pip list in your terminal and checking for azure-ai-vision-imageanalysis and pyodbc.

Help me fix import errors in my pipeline.py file.

Write the analyze_image function

The core of your pipeline is a function that takes any image URL, sends it to Azure AI Vision, and returns structured data about what the AI detected.

  • In pipeline.py, add the following function below the SQL_PASSWORD line:
def get_vision_client():
    """Create and return an Azure AI Vision client."""
    return ImageAnalysisClient(
        endpoint=VISION_ENDPOINT,
        credential=AzureKeyCredential(VISION_KEY)
    )

What does this function do?

The get_vision_client() function creates an authenticated connection to your Azure AI Vision resource. It combines your endpoint URL with an AzureKeyCredential wrapping your API key so the SDK can sign each request.

Now add the main analysis function that calls the Vision API and structures the results.

  • Below get_vision_client(), add the analyze_image function:
def analyze_image(image_url):
    """Analyze an image URL and return structured results."""
    client = get_vision_client()

    # Request four types of visual analysis from the API
    result = client.analyze_from_url(
        image_url=image_url,
        visual_features=[
            VisualFeatures.CAPTION,
            VisualFeatures.TAGS,
            VisualFeatures.OBJECTS,
            VisualFeatures.READ
        ]
    )

    # Build a structured dictionary to hold the results
    analysis = {
        "image_url": image_url,
        "caption": None,
        "caption_confidence": None,
        "tags": [],
        "objects": [],
        "text_lines": []
    }

    # Extract the AI-generated caption
    if result.caption is not None:
        analysis["caption"] = result.caption.text
        analysis["caption_confidence"] = result.caption.confidence

    # Extract content tags with confidence scores
    if result.tags is not None:
        for tag in result.tags.list:
            analysis["tags"].append({"name": tag.name, "confidence": tag.confidence})

    # Extract detected objects
    if result.objects is not None:
        for obj in result.objects.list:
            analysis["objects"].append({"name": obj.tags[0].name if obj.tags else "unknown", "confidence": obj.tags[0].confidence if obj.tags else 0})

    # Extract any text found in the image (OCR)
    if result.read is not None:
        for block in result.read.blocks:
            for line in block.lines:
                analysis["text_lines"].append(line.text)

    return analysis

What does this code do?

The analyze_from_url method sends the image URL to Azure AI Vision and requests four types of analysis: CAPTION (a sentence describing the image), TAGS (keywords like "outdoor" or "building"), OBJECTS (specific things detected with bounding boxes), and READ (text extracted via OCR).

The function then unpacks each result section into a clean Python dictionary. Each if block checks whether the API returned data for that feature before trying to access it.

  • Save pipeline.py.

Seeing a NameError for VisualFeatures?

Make sure your imports at the top of the file include from azure.ai.vision.imageanalysis.models import VisualFeatures. This enum defines the feature types the API accepts.

Help me fix a NameError in my analyze_image function.

Test with a sample image

Your function is ready. Time to send a real image through it and see what Azure AI Vision detects.

  • At the bottom of pipeline.py, add the following test code:
if __name__ == "__main__":
    # Test with a sample image from Microsoft's docs
    sample_url = "https://learn.microsoft.com/azure/ai-services/computer-vision/media/quickstarts/presentation.png"
    analysis = analyze_image(sample_url)

    print(f"Caption: {analysis['caption']}")
    print(f"Tags: {', '.join(t['name'] for t in analysis['tags'][:5])}")
    print(f"Objects: {', '.join(o['name'] for o in analysis['objects'])}")
    print(f"Text lines: {len(analysis['text_lines'])}")

What does this code do?

The if __name__ == "__main__" block only runs when you execute this file directly. It sends a sample presentation image to your analyze_image function and prints the caption, first five tags, detected objects, and how many lines of text the OCR found.

  • Save pipeline.py.
  • Run the script by entering this command in your terminal:
python pipeline.py

You should see output showing an AI-generated caption, a list of tags, any detected objects, and the number of text lines found in the image.

Getting an authentication error?

Double-check your .env file. Make sure VISION_ENDPOINT includes the full URL (starting with https://) and VISION_KEY is the key value with no extra spaces.

If you see a FileNotFoundError for the .env file, make sure you are running the command from inside the image-pipeline folder.

Help me fix an authentication error when running my pipeline.py script.

✔️ Awesome, I've got everything!

Great. Double check you have saved both .env and pipeline.py.

ⓧ I'd like to double check the full code

Here is the complete .env file:

VISION_ENDPOINT=your-vision-endpoint-here
VISION_KEY=your-vision-key-here
SQL_SERVER=your-server-name.database.windows.net
SQL_DATABASE=your-database-name
SQL_USERNAME=your-admin-username
SQL_PASSWORD=your-admin-password

Here is the complete pipeline.py file:

import os
import pyodbc
from datetime import datetime
from dotenv import load_dotenv
from azure.ai.vision.imageanalysis import ImageAnalysisClient
from azure.ai.vision.imageanalysis.models import VisualFeatures
from azure.core.credentials import AzureKeyCredential

load_dotenv()

VISION_ENDPOINT = os.environ["VISION_ENDPOINT"]
VISION_KEY = os.environ["VISION_KEY"]
SQL_SERVER = os.environ["SQL_SERVER"]
SQL_DATABASE = os.environ["SQL_DATABASE"]
SQL_USERNAME = os.environ["SQL_USERNAME"]
SQL_PASSWORD = os.environ["SQL_PASSWORD"]


def get_vision_client():
    return ImageAnalysisClient(
        endpoint=VISION_ENDPOINT,
        credential=AzureKeyCredential(VISION_KEY)
    )


def analyze_image(image_url):
    client = get_vision_client()
    result = client.analyze_from_url(
        image_url=image_url,
        visual_features=[
            VisualFeatures.CAPTION,
            VisualFeatures.TAGS,
            VisualFeatures.OBJECTS,
            VisualFeatures.READ
        ]
    )
    analysis = {
        "image_url": image_url,
        "caption": None,
        "caption_confidence": None,
        "tags": [],
        "objects": [],
        "text_lines": []
    }
    if result.caption is not None:
        analysis["caption"] = result.caption.text
        analysis["caption_confidence"] = result.caption.confidence
    if result.tags is not None:
        for tag in result.tags.list:
            analysis["tags"].append({"name": tag.name, "confidence": tag.confidence})
    if result.objects is not None:
        for obj in result.objects.list:
            analysis["objects"].append({"name": obj.tags[0].name if obj.tags else "unknown", "confidence": obj.tags[0].confidence if obj.tags else 0})
    if result.read is not None:
        for block in result.read.blocks:
            for line in block.lines:
                analysis["text_lines"].append(line.text)
    return analysis


if __name__ == "__main__":
    sample_url = "https://learn.microsoft.com/azure/ai-services/computer-vision/media/quickstarts/presentation.png"
    analysis = analyze_image(sample_url)
    print(f"Caption: {analysis['caption']}")
    print(f"Tags: {', '.join(t['name'] for t in analysis['tags'][:5])}")
    print(f"Objects: {', '.join(o['name'] for o in analysis['objects'])}")
    print(f"Text lines: {len(analysis['text_lines'])}")

Your image analysis function is working. Next up, you will design the database schema and write the code that stores these results in your Azure SQL Database.

Design the Database Schema and Store Results

Your pipeline can analyze images, but the results vanish the moment the script finishes. To turn this into a real intelligence system, you need a persistent store where results accumulate and become queryable.

In this step, you'll connect your Python script to the Azure SQL Database you created earlier, design a schema to hold image analysis results, and write a function that saves everything to the database.

In this step, get ready to:
  • Connect to your Azure SQL Database from Python using pyodbc.
  • Create tables to store image analyses (an images table and a tags table).
  • Write a save_results function that inserts analysis results into the database.
Connect to Azure SQL Database from Python

Your .env file already holds the SQL Server credentials. Now you need a function that builds the connection string and opens a live connection to your database.

  • Switch back to pipeline.py in VS Code and scroll to the bottom of the analyze_image function.
  • Below the analyze_image function (after its return analysis line), add the following get_db_connection function:
def get_db_connection():
    """Create and return a database connection."""
    conn_str = (
        f"Driver={{ODBC Driver 18 for SQL Server}};"
        f"Server=tcp:{SQL_SERVER},1433;"
        f"Database={SQL_DATABASE};"
        f"Uid={SQL_USERNAME};"
        f"Pwd={SQL_PASSWORD};"
        "Encrypt=yes;"
        "TrustServerCertificate=no;"
        "Connection Timeout=30;"
    )
    return pyodbc.connect(conn_str)

What does this code do?

This function builds a connection string with the credentials from your .env file and opens a connection to your Azure SQL Database.

  • Driver={{ODBC Driver 18 for SQL Server}} tells pyodbc which driver to use (the double braces escape the curly brackets in the f-string).
  • Encrypt=yes ensures all traffic between your machine and Azure is encrypted in transit.
  • Connection Timeout=30 gives Azure 30 seconds to respond before timing out.
  • Save pipeline.py.
  • You won't see a visible change yet. The next chunk uses this function to create your database tables.

Seeing a driver not found error later?

If you eventually see an error mentioning the ODBC Driver 18 not being found, confirm it is installed by running:

python -c "import pyodbc; print(pyodbc.drivers())"

You should see ODBC Driver 18 for SQL Server in the output list. If it is missing, revisit Step 1 to reinstall the driver.

Help me debug my ODBC driver issue.

Create tables to store image analyses

You need two tables: one for the image-level data (caption, URL, extracted text) and one for the tags and objects detected in each image. Separating tags into their own table lets you query across images later (e.g., "which images contain people?").

  • Below get_db_connection, add the create_tables function:
def create_tables():
    """Create the database tables if they don't exist."""
    conn = get_db_connection()
    cursor = conn.cursor()

    cursor.execute("""
        IF NOT EXISTS (SELECT * FROM sys.tables WHERE name = 'images')
        CREATE TABLE images (
            id INT IDENTITY(1,1) PRIMARY KEY,
            image_url NVARCHAR(2048) NOT NULL,
            caption NVARCHAR(500),
            caption_confidence FLOAT,
            extracted_text NVARCHAR(MAX),
            analyzed_at DATETIME2 DEFAULT GETDATE()
        )
    """)

    cursor.execute("""
        IF NOT EXISTS (SELECT * FROM sys.tables WHERE name = 'image_tags')
        CREATE TABLE image_tags (
            id INT IDENTITY(1,1) PRIMARY KEY,
            image_id INT NOT NULL,
            tag_name NVARCHAR(100) NOT NULL,
            confidence FLOAT,
            tag_type NVARCHAR(20) NOT NULL,
            FOREIGN KEY (image_id) REFERENCES images(id)
        )
    """)

    conn.commit()
    cursor.close()
    conn.close()
    print("Database tables created successfully.")

What does this code do?

This function creates two tables in your Azure SQL Database, but only if they don't already exist (thanks to the IF NOT EXISTS guard).

  • images stores one row per analyzed image: the URL, the AI-generated caption with its confidence score, any extracted text, and a timestamp.
  • image_tags stores individual tags and objects, linked back to images via a foreign key. The tag_type column distinguishes between content tags and detected objects.
  • conn.commit() finalizes the table creation. Without it, the changes would be rolled back when the connection closes.
  • Save pipeline.py.
  • Test the database connection by running this in your terminal:
python -c "from pipeline import create_tables; create_tables()"

You should see Database tables created successfully. printed in your terminal.

Seeing a login failed or connection refused error?

  • Double check that the values in your .env file match the admin username and password you set when creating the SQL Server.
  • Confirm your local IP is in the server's firewall rules. Return to the Azure Portal, open your SQL Server (not database), click Networking, and verify your IP appears under Firewall rules.
  • Make sure SQL_SERVER in your .env includes the full hostname (e.g., yourserver.database.windows.net), not just the server name.

Help me debug my Azure SQL Database connection error.

Write the save_results function

Now that your tables exist, you need a function that takes the analysis dictionary from analyze_image and inserts it across both tables.

  • Below create_tables, add the save_results function:
def save_results(analysis):
    """Save analysis results to the database."""
    conn = get_db_connection()
    cursor = conn.cursor()

    extracted_text = " | ".join(analysis["text_lines"]) if analysis["text_lines"] else None

    cursor.execute(
        "INSERT INTO images (image_url, caption, caption_confidence, extracted_text) "
        "OUTPUT INSERTED.id "
        "VALUES (?, ?, ?, ?)",
        analysis["image_url"],
        analysis["caption"],
        analysis["caption_confidence"],
        extracted_text
    )
    image_id = cursor.fetchone()[0]

    for tag in analysis["tags"]:
        cursor.execute(
            "INSERT INTO image_tags (image_id, tag_name, confidence, tag_type) VALUES (?, ?, ?, ?)",
            image_id, tag["name"], tag["confidence"], "tag"
        )

    for obj in analysis["objects"]:
        cursor.execute(
            "INSERT INTO image_tags (image_id, tag_name, confidence, tag_type) VALUES (?, ?, ?, ?)",
            image_id, obj["name"], obj["confidence"], "object"
        )

    conn.commit()
    cursor.close()
    conn.close()
    return image_id

What does this code do?

  • extracted_text joins all OCR text lines into a single string separated by | so they fit in one database column.
  • OUTPUT INSERTED.id is a SQL Server feature that returns the auto-generated ID of the row just inserted. This lets you immediately link tags back to the parent image.
  • The two loops insert each tag and each detected object as separate rows in image_tags. The tag_type column stores either "tag" or "object" so you can filter them apart in queries.
  • Save pipeline.py.
  • Now replace your existing if __name__ == "__main__": block at the bottom of the file with this test code to run the full pipeline:
if __name__ == "__main__":
    # Test the full pipeline: analyze + store
    create_tables()
    sample_url = "https://learn.microsoft.com/azure/ai-services/computer-vision/media/quickstarts/presentation.png"
    print(f"Analyzing: {sample_url[:60]}...")
    analysis = analyze_image(sample_url)
    print(f"Caption: {analysis['caption']}")
    image_id = save_results(analysis)
    print(f"Saved to database with ID: {image_id}")
  • Run the script in your terminal:
python pipeline.py

You should see output showing the caption text and a database ID number (e.g., Saved to database with ID: 1).

Seeing an invalid object name error?

  • If you see an error about images or image_tags not being a valid object, the create_tables function may not have run successfully. Try running it again separately:
  • If you see a timeout, your database may have auto-paused. Wait 30 seconds and try again.
python -c "from pipeline import create_tables; create_tables()"

Help me fix my save_results database error.

  • Once you see the success message, delete the if __name__ == "__main__": block from the bottom of the file. You will rebuild it properly in the next step.
  • Save pipeline.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved pipeline.py after removing the test block.

ⓧ I'd like to double check the full code

import os
import pyodbc
from datetime import datetime
from dotenv import load_dotenv
from azure.ai.vision.imageanalysis import ImageAnalysisClient
from azure.ai.vision.imageanalysis.models import VisualFeatures
from azure.core.credentials import AzureKeyCredential

load_dotenv()

VISION_ENDPOINT = os.environ["VISION_ENDPOINT"]
VISION_KEY = os.environ["VISION_KEY"]
SQL_SERVER = os.environ["SQL_SERVER"]
SQL_DATABASE = os.environ["SQL_DATABASE"]
SQL_USERNAME = os.environ["SQL_USERNAME"]
SQL_PASSWORD = os.environ["SQL_PASSWORD"]


def get_vision_client():
    """Create and return an Azure AI Vision client."""
    return ImageAnalysisClient(
        endpoint=VISION_ENDPOINT,
        credential=AzureKeyCredential(VISION_KEY)
    )


def analyze_image(image_url):
    """Analyze an image URL and return structured results."""
    client = get_vision_client()

    result = client.analyze_from_url(
        image_url=image_url,
        visual_features=[
            VisualFeatures.CAPTION,
            VisualFeatures.TAGS,
            VisualFeatures.OBJECTS,
            VisualFeatures.READ
        ]
    )

    analysis = {
        "image_url": image_url,
        "caption": None,
        "caption_confidence": None,
        "tags": [],
        "objects": [],
        "text_lines": []
    }

    if result.caption is not None:
        analysis["caption"] = result.caption.text
        analysis["caption_confidence"] = result.caption.confidence

    if result.tags is not None:
        for tag in result.tags.list:
            analysis["tags"].append({
                "name": tag.name,
                "confidence": tag.confidence
            })

    if result.objects is not None:
        for obj in result.objects.list:
            analysis["objects"].append({
                "name": obj.tags[0].name if obj.tags else "unknown",
                "confidence": obj.tags[0].confidence if obj.tags else 0
            })

    if result.read is not None:
        for block in result.read.blocks:
            for line in block.lines:
                analysis["text_lines"].append(line.text)

    return analysis


def get_db_connection():
    """Create and return a database connection."""
    conn_str = (
        f"Driver={{ODBC Driver 18 for SQL Server}};"
        f"Server=tcp:{SQL_SERVER},1433;"
        f"Database={SQL_DATABASE};"
        f"Uid={SQL_USERNAME};"
        f"Pwd={SQL_PASSWORD};"
        "Encrypt=yes;"
        "TrustServerCertificate=no;"
        "Connection Timeout=30;"
    )
    return pyodbc.connect(conn_str)


def create_tables():
    """Create the database tables if they don't exist."""
    conn = get_db_connection()
    cursor = conn.cursor()

    cursor.execute("""
        IF NOT EXISTS (SELECT * FROM sys.tables WHERE name = 'images')
        CREATE TABLE images (
            id INT IDENTITY(1,1) PRIMARY KEY,
            image_url NVARCHAR(2048) NOT NULL,
            caption NVARCHAR(500),
            caption_confidence FLOAT,
            extracted_text NVARCHAR(MAX),
            analyzed_at DATETIME2 DEFAULT GETDATE()
        )
    """)

    cursor.execute("""
        IF NOT EXISTS (SELECT * FROM sys.tables WHERE name = 'image_tags')
        CREATE TABLE image_tags (
            id INT IDENTITY(1,1) PRIMARY KEY,
            image_id INT NOT NULL,
            tag_name NVARCHAR(100) NOT NULL,
            confidence FLOAT,
            tag_type NVARCHAR(20) NOT NULL,
            FOREIGN KEY (image_id) REFERENCES images(id)
        )
    """)

    conn.commit()
    cursor.close()
    conn.close()
    print("Database tables created successfully.")


def save_results(analysis):
    """Save analysis results to the database."""
    conn = get_db_connection()
    cursor = conn.cursor()

    extracted_text = " | ".join(analysis["text_lines"]) if analysis["text_lines"] else None

    cursor.execute(
        "INSERT INTO images (image_url, caption, caption_confidence, extracted_text) "
        "OUTPUT INSERTED.id "
        "VALUES (?, ?, ?, ?)",
        analysis["image_url"],
        analysis["caption"],
        analysis["caption_confidence"],
        extracted_text
    )
    image_id = cursor.fetchone()[0]

    for tag in analysis["tags"]:
        cursor.execute(
            "INSERT INTO image_tags (image_id, tag_name, confidence, tag_type) VALUES (?, ?, ?, ?)",
            image_id, tag["name"], tag["confidence"], "tag"
        )

    for obj in analysis["objects"]:
        cursor.execute(
            "INSERT INTO image_tags (image_id, tag_name, confidence, tag_type) VALUES (?, ?, ?, ?)",
            image_id, obj["name"], obj["confidence"], "object"
        )

    conn.commit()
    cursor.close()
    conn.close()
    return image_id

Your pipeline now persists every analysis to a cloud database. Next up, you'll write SQL queries to mine patterns from the data you've stored.

Query Your Intelligence Data

You have image analysis results sitting in your database. Now it's time to unlock the real value of storing structured data.

In this step, you'll write SQL queries that mine your stored results for patterns. Then you'll tie the entire pipeline together into a single script that analyzes an image, stores the results, and queries the database in one run.

In this step, get ready to:
  • Write SQL queries that find the most common tags, detected objects, and images with extracted text.
  • Build a main function that orchestrates the full pipeline end-to-end.
  • Run the pipeline and print a formatted intelligence report.
Write the query_database() function

Your database has tags, objects, and text stored across two tables. SQL aggregation queries let you slice this data to answer real questions like "what are the most common tags?" or "which images contain text?"

  • In pipeline.py, scroll to the bottom of the file (below the save_results function) and add the following function:
def query_database():
    """Run sample queries against the stored data."""
    conn = get_db_connection()
    cursor = conn.cursor()

    print("\n--- Most Common Tags (Top 10) ---")
    cursor.execute("""
        SELECT TOP 10 tag_name, COUNT(*) as count, AVG(confidence) as avg_confidence
        FROM image_tags
        WHERE tag_type = 'tag'
        GROUP BY tag_name
        ORDER BY count DESC
    """)
    for row in cursor.fetchall():
        print(f"  {row.tag_name}: {row.count} occurrences (avg confidence: {row.avg_confidence:.2f})")

    print("\n--- Detected Objects ---")
    cursor.execute("""
        SELECT TOP 10 tag_name, COUNT(*) as count
        FROM image_tags
        WHERE tag_type = 'object'
        GROUP BY tag_name
        ORDER BY count DESC
    """)
    for row in cursor.fetchall():
        print(f"  {row.tag_name}: {row.count} detections")

    print("\n--- Images With Extracted Text ---")
    cursor.execute("""
        SELECT image_url, caption, extracted_text
        FROM images
        WHERE extracted_text IS NOT NULL
    """)
    for row in cursor.fetchall():
        print(f"  URL: {row.image_url[:60]}...")
        print(f"  Caption: {row.caption}")
        print(f"  Text: {row.extracted_text[:100]}...")
        print()

    cursor.close()
    conn.close()

What does this code do?

This function runs three SQL queries against your image_tags and images tables to surface patterns in your analysis data.

  • Query 1 finds the top 10 tags by frequency, with their average confidence score. GROUP BY collapses duplicate tag names into counts.
  • Query 2 finds the top 10 detected objects. The tag_type = 'object' filter separates objects from general tags.
  • Query 3 retrieves any images that had text extracted via OCR. It filters for rows where extracted_text IS NOT NULL.
  • Save pipeline.py.

Seeing a syntax error in the SQL?

Make sure each triple-quoted SQL string has matching """ at the start and end. A missing closing set of quotes will cause Python to treat the next line of code as part of the string.

Help me debug my SQL syntax error in query_database.

Write the main() function to orchestrate the pipeline

Right now your functions exist independently. The main() function ties them into a single workflow: create tables, analyze an image, save the results, and query the database.

  • Below the query_database function in pipeline.py, add the main() function and entry point:
def main():
    """Main pipeline: analyze an image, store results, query database."""
    print("=== Image Intelligence Pipeline ===\n")

    # Step 1: Create tables
    create_tables()

    # Step 2: Analyze a sample image
    sample_url = "https://learn.microsoft.com/azure/ai-services/computer-vision/media/quickstarts/presentation.png"
    print(f"\nAnalyzing image: {sample_url[:60]}...")

    analysis = analyze_image(sample_url)

    print(f"  Caption: {analysis['caption']}")
    print(f"  Tags: {', '.join(t['name'] for t in analysis['tags'][:5])}")
    print(f"  Objects: {', '.join(o['name'] for o in analysis['objects'])}")
    print(f"  Text lines: {len(analysis['text_lines'])}")

    # Step 3: Save to database
    image_id = save_results(analysis)
    print(f"\n  Saved to database with ID: {image_id}")

    # Step 4: Query the database
    query_database()

    print("\n=== Pipeline Complete ===")


if __name__ == "__main__":
    main()

What does this code do?

The main() function runs the full pipeline in four steps: ensure tables exist, analyze a sample image URL, save the analysis to the database, then query and print results.

The if __name__ == "__main__" block tells Python to run main() only when you execute the file directly. This prevents the pipeline from running if you import the file from another script later.

The sample image URL is from Microsoft's official Computer Vision documentation. It contains a presentation slide with both visual elements and text, so it exercises all four analysis features (caption, tags, objects, and OCR).

  • Save pipeline.py.

Seeing a NameError for main?

Make sure the if __name__ == "__main__": line is at the very bottom of the file with no indentation. It should be outside any function definition.

Help me fix my NameError in pipeline.py.

Run the full pipeline end-to-end

Everything is wired up. One command now triggers image analysis, database storage, and intelligence queries all in sequence.

  • Run the pipeline by executing this command in your terminal:
python pipeline.py
  • You should see output showing the full pipeline flow: table creation, image analysis results (caption, tags, objects, text line count), a database ID, and then your formatted intelligence report with top tags, detected objects, and extracted text.

What should I see?

Your terminal prints a report with sections separated by dashes. The "Most Common Tags" section shows tags like indoor, text, or furniture with occurrence counts and confidence scores.

The "Detected Objects" section lists objects found in the image. The "Images With Extracted Text" section shows any OCR text that was pulled from the presentation slide.

Seeing a connection error?

If you see a timeout or login failure, your Azure SQL firewall may have dropped your IP. Return to the Azure Portal, navigate to your SQL server's Networking page, and verify your current client IP is still in the firewall rules.

If you see a driver error mentioning ODBC, confirm the driver is installed by running python -c "import pyodbc; print(pyodbc.drivers())" and checking that ODBC Driver 18 for SQL Server appears in the list.

Help me troubleshoot my Azure SQL connection error.

✔️ Awesome, I've got everything!

Great. Double check you've saved pipeline.py and your pipeline ran successfully with output in the terminal.

ⓧ I'd like to double check the full code

import os
import pyodbc
from datetime import datetime
from dotenv import load_dotenv
from azure.ai.vision.imageanalysis import ImageAnalysisClient
from azure.ai.vision.imageanalysis.models import VisualFeatures
from azure.core.credentials import AzureKeyCredential

load_dotenv()

VISION_ENDPOINT = os.environ["VISION_ENDPOINT"]
VISION_KEY = os.environ["VISION_KEY"]
SQL_SERVER = os.environ["SQL_SERVER"]
SQL_DATABASE = os.environ["SQL_DATABASE"]
SQL_USERNAME = os.environ["SQL_USERNAME"]
SQL_PASSWORD = os.environ["SQL_PASSWORD"]


def get_vision_client():
    return ImageAnalysisClient(endpoint=VISION_ENDPOINT, credential=AzureKeyCredential(VISION_KEY))


def analyze_image(image_url):
    client = get_vision_client()
    result = client.analyze_from_url(image_url=image_url, visual_features=[VisualFeatures.CAPTION, VisualFeatures.TAGS, VisualFeatures.OBJECTS, VisualFeatures.READ])
    analysis = {"image_url": image_url, "caption": None, "caption_confidence": None, "tags": [], "objects": [], "text_lines": []}
    if result.caption is not None:
        analysis["caption"] = result.caption.text
        analysis["caption_confidence"] = result.caption.confidence
    if result.tags is not None:
        for tag in result.tags.list:
            analysis["tags"].append({"name": tag.name, "confidence": tag.confidence})
    if result.objects is not None:
        for obj in result.objects.list:
            analysis["objects"].append({"name": obj.tags[0].name if obj.tags else "unknown", "confidence": obj.tags[0].confidence if obj.tags else 0})
    if result.read is not None:
        for block in result.read.blocks:
            for line in block.lines:
                analysis["text_lines"].append(line.text)
    return analysis


def get_db_connection():
    conn_str = (f"Driver={{ODBC Driver 18 for SQL Server}};Server=tcp:{SQL_SERVER},1433;Database={SQL_DATABASE};Uid={SQL_USERNAME};Pwd={SQL_PASSWORD};Encrypt=yes;TrustServerCertificate=no;Connection Timeout=30;")
    return pyodbc.connect(conn_str)


def create_tables():
    conn = get_db_connection()
    cursor = conn.cursor()
    cursor.execute("IF NOT EXISTS (SELECT * FROM sys.tables WHERE name = 'images') CREATE TABLE images (id INT IDENTITY(1,1) PRIMARY KEY, image_url NVARCHAR(2048) NOT NULL, caption NVARCHAR(500), caption_confidence FLOAT, extracted_text NVARCHAR(MAX), analyzed_at DATETIME2 DEFAULT GETDATE())")
    cursor.execute("IF NOT EXISTS (SELECT * FROM sys.tables WHERE name = 'image_tags') CREATE TABLE image_tags (id INT IDENTITY(1,1) PRIMARY KEY, image_id INT NOT NULL, tag_name NVARCHAR(100) NOT NULL, confidence FLOAT, tag_type NVARCHAR(20) NOT NULL, FOREIGN KEY (image_id) REFERENCES images(id))")
    conn.commit()
    cursor.close()
    conn.close()
    print("Database tables created successfully.")


def save_results(analysis):
    conn = get_db_connection()
    cursor = conn.cursor()
    extracted_text = " | ".join(analysis["text_lines"]) if analysis["text_lines"] else None
    cursor.execute("INSERT INTO images (image_url, caption, caption_confidence, extracted_text) OUTPUT INSERTED.id VALUES (?, ?, ?, ?)", analysis["image_url"], analysis["caption"], analysis["caption_confidence"], extracted_text)
    image_id = cursor.fetchone()[0]
    for tag in analysis["tags"]:
        cursor.execute("INSERT INTO image_tags (image_id, tag_name, confidence, tag_type) VALUES (?, ?, ?, ?)", image_id, tag["name"], tag["confidence"], "tag")
    for obj in analysis["objects"]:
        cursor.execute("INSERT INTO image_tags (image_id, tag_name, confidence, tag_type) VALUES (?, ?, ?, ?)", image_id, obj["name"], obj["confidence"], "object")
    conn.commit()
    cursor.close()
    conn.close()
    return image_id


def query_database():
    conn = get_db_connection()
    cursor = conn.cursor()
    print("\n--- Most Common Tags (Top 10) ---")
    cursor.execute("SELECT TOP 10 tag_name, COUNT(*) as count, AVG(confidence) as avg_confidence FROM image_tags WHERE tag_type = 'tag' GROUP BY tag_name ORDER BY count DESC")
    for row in cursor.fetchall():
        print(f"  {row.tag_name}: {row.count} occurrences (avg confidence: {row.avg_confidence:.2f})")
    print("\n--- Detected Objects ---")
    cursor.execute("SELECT TOP 10 tag_name, COUNT(*) as count FROM image_tags WHERE tag_type = 'object' GROUP BY tag_name ORDER BY count DESC")
    for row in cursor.fetchall():
        print(f"  {row.tag_name}: {row.count} detections")
    print("\n--- Images With Extracted Text ---")
    cursor.execute("SELECT image_url, caption, extracted_text FROM images WHERE extracted_text IS NOT NULL")
    for row in cursor.fetchall():
        print(f"  URL: {row.image_url[:60]}...")
        print(f"  Caption: {row.caption}")
        print(f"  Text: {row.extracted_text[:100]}...")
        print()
    cursor.close()
    conn.close()


def main():
    print("=== Image Intelligence Pipeline ===\n")
    create_tables()
    sample_url = "https://learn.microsoft.com/azure/ai-services/computer-vision/media/quickstarts/presentation.png"
    print(f"\nAnalyzing image: {sample_url[:60]}...")
    analysis = analyze_image(sample_url)
    print(f"  Caption: {analysis['caption']}")
    print(f"  Tags: {', '.join(t['name'] for t in analysis['tags'][:5])}")
    print(f"  Objects: {', '.join(o['name'] for o in analysis['objects'])}")
    print(f"  Text lines: {len(analysis['text_lines'])}")
    image_id = save_results(analysis)
    print(f"\n  Saved to database with ID: {image_id}")
    query_database()
    print("\n=== Pipeline Complete ===")


if __name__ == "__main__":
    main()

Your pipeline is complete. You can analyze any image, store the results in a cloud database, and query for patterns. Next up, you'll extend this into a batch processing system in the Secret Mission.

Secret mission

Batch Processing and Summary Report

Your pipeline can analyze one image at a time. But what if you had a folder of 50 images to process? In this secret mission, you'll add batch processing that analyzes multiple images in one run, handles errors gracefully, and generates a summary report of intelligence gathered across all images.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. Both the Azure AI Vision (F0) and Azure SQL Database (free offer) resources are on free tiers, so there are no ongoing costs as long as you stay within the free limits.

Resources you used:

  • Azure AI Vision resource (F0 free tier, 5,000 transactions/month).
  • Azure SQL Database (free offer, 100,000 vCore seconds/month).
  • Local project files (pipeline.py, .env).
  • Python packages (azure-ai-vision-imageanalysis, pyodbc, python-dotenv).
  • ODBC Driver 18 for SQL Server.

Keep everything running

No action needed. Choose this if you're still actively building or want to extend the pipeline later.

  • Both Azure services are on free tiers. The AI Vision resource allows 5,000 transactions/month and the SQL Database provides 100,000 vCore seconds/month at no cost.
  • The SQL Database will auto-pause when idle, so you won't consume vCore seconds while you're not using it.
  • Your local project files and installed packages remain on your machine ready to use.

Pause - I'll come back to this later

Close any active database connections to avoid consuming vCore seconds unnecessarily. Your data and resources stay intact.

  • Make sure your Python script is not running. Any open connections to the Azure SQL Database consume vCore seconds even when idle.
  • Close VS Code if you don't need it open for other work.
  • Both Azure resources remain available when you return. The SQL Database will auto-pause after inactivity and resume automatically on the next connection.

Delete - I don't want to use this again

Remove all project resources and start fresh. Deleting the Resource Group in Azure Portal removes both the AI Vision resource and the SQL Database in one action.

  • Delete the Azure Resource Group:
  • Delete your local project files by running this in your terminal:
rm -rf pipeline.py .env

On Windows, run this instead:

Remove-Item pipeline.py, .env
  • Uninstall the Python packages by running:
pip uninstall azure-ai-vision-imageanalysis pyodbc python-dotenv -y
  • Uninstall ODBC Driver 18 for SQL Server:

That's a Wrap!

That's a Wrap!

Nice work! You just built a cloud-powered image intelligence pipeline that analyzes images with AI and stores structured results in a queryable database.

You've learned how to:

  • Build a Python script that calls Azure AI Vision to analyze images for captions, tags, objects, and text.
  • Provision and configure Azure cloud resources (AI Vision + SQL Database) from scratch.
  • Design a SQL schema and write queries to mine patterns from AI-generated data.
  • Add batch processing with aggregate reporting across multiple images.

Ready to quiz yourself?