Develop AI Solutions with Azure Database for PostgreSQL

1. Compute Tiers

2. Managed Capabilities

1. Automatic backups — 7–35 day retention, point-in-time restore to any second within the window. AES-256 encrypted at rest.
2. High Availability — zone-redundant HA with automatic failover. Standby replica in a different AZ.
3. Extensions — enable with CREATE EXTENSION:
   1. pgvector — vector similarity search (cosine, L2, inner product)
   2. pg_trgm — fuzzy text matching for AI document search
   3. uuid-ossp — UUID generation
   4. hstore — key-value pairs in a column

1. Connection Parameters

1. Endpoint: servername.postgres.database.azure.com
2. Direct port: 5432
3. PgBouncer port: 6432
4. Username format (Entra): username@servername

2. Entra ID Authentication (Recommended)

Short-lived OAuth2 tokens instead of passwords. Works with managed identities. No stored credentials.

from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()
token = credential.get_token("https://ossrdbms-aad.database.windows.net/.default")
# Use token.token as the password in your psycopg2 connection string

3. TLS / SSL Modes

1. disable — no encryption. Azure rejects this for managed PostgreSQL.
2. require — encrypts connection but does NOT validate server certificate.
3. verify-ca — validates the certificate authority.
4. verify-full — validates CA AND hostname. Recommended for production.

4. PgBouncer — Connection Pooling

AI applications make many short-lived DB calls (one per inference request). Each new TCP connection is expensive. PgBouncer pools connections — the app creates a connection to PgBouncer, which reuses existing DB connections.

az postgres flexible-server parameter set \\
  --resource-group rg --server-name myserver \\
  --name pgbouncer.enabled --value true

# Connect via PgBouncer on port 6432
postgresql://[email protected]:6432/mydb?sslmode=require

1. Hierarchy: Server → Database → Schema → Table

Default schema is public. Use separate databases for full isolation; separate schemas for logical grouping with cross-schema JOINs possible.

2. AI-Relevant Data Types

1. JSONB — binary JSON with GIN indexing. Use for flexible metadata, model parameters, nested structures.
2. TIMESTAMPTZ — always use over TIMESTAMP for global apps. Stores UTC, displays per session timezone.
3. BIGSERIAL — auto-incrementing 64-bit integer PK. Use over SERIAL (32-bit overflows on high-volume tables).
4. TEXT — unbounded string. Same performance as VARCHAR in PostgreSQL. No reason to use VARCHAR unless enforcing max length.
5. UUID — universally unique ID. Use DEFAULT gen_random_uuid(). Good for distributed, globally-unique identifiers.

3. Conversation History Table Example

CREATE TABLE conversations (
    id          BIGSERIAL PRIMARY KEY,
    session_id  UUID NOT NULL DEFAULT gen_random_uuid(),
    user_id     VARCHAR(255) NOT NULL,
    started_at  TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP,
    metadata    JSONB DEFAULT '{}'::jsonb
);

CREATE TABLE messages (
    id              BIGSERIAL PRIMARY KEY,
    conversation_id BIGINT NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
    role            VARCHAR(50) NOT NULL CHECK (role IN ('user', 'assistant', 'system')),
    content         TEXT NOT NULL,
    created_at      TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
);

4. Indexes

CREATE INDEX idx_messages_conversation_id ON messages(conversation_id);
CREATE INDEX idx_messages_conv_created ON messages(conversation_id, created_at);

Indexes speed reads but slow writes. Add indexes when query profiling reveals slow queries — not preemptively on every column.

5. Transactional DDL (PostgreSQL Superpower)

BEGIN;
ALTER TABLE conversations ADD COLUMN category VARCHAR(100);
CREATE INDEX idx_conversations_category ON conversations(category);
COMMIT;   -- or ROLLBACK on failure

1. JSONB Queries

-- Access JSONB field
SELECT metadata->>'model' AS model_name FROM conversations;

-- Filter on JSONB (containment operator @>)
SELECT * FROM conversations WHERE metadata @> '{"tier": "premium"}'::jsonb;

-- Check JSONB key existence
SELECT * FROM conversations WHERE metadata ? 'user_preferences';

2. Upserts with ON CONFLICT

INSERT INTO conversations (session_id, user_id) VALUES ($1, $2)
ON CONFLICT (session_id)
DO UPDATE SET user_id = EXCLUDED.user_id, started_at = CURRENT_TIMESTAMP;

3. Keyset Pagination (Avoid OFFSET)

OFFSET gets slower as the offset grows. Keyset pagination is O(log n) with the right index:

-- Get next 50 messages after last seen id
SELECT * FROM messages
WHERE conversation_id = $1 AND id > $last_id
ORDER BY id ASC LIMIT 50;

4. INSERT RETURNING (Avoid Extra SELECT)

INSERT INTO messages (conversation_id, role, content)
VALUES ($1, 'user', $2)
RETURNING id, created_at;   -- Get generated values without extra round-trip

Enable pgvector and Create Table

-- Enable extension (once per database)
CREATE EXTENSION IF NOT EXISTS vector;

-- Create table with embedding column
CREATE TABLE documents (
    id          SERIAL PRIMARY KEY,
    title       TEXT NOT NULL,
    content     TEXT NOT NULL,
    embedding   vector(1536),  -- matches text-embedding-3-small
    category    TEXT,
    created_at  TIMESTAMPTZ DEFAULT NOW()
);

-- HNSW index for fast approximate search
CREATE INDEX ON documents
    USING hnsw (embedding vector_cosine_ops)
    WITH (m=16, ef_construction=64);

-- Insert with Python
import psycopg2
cur.execute(
    "INSERT INTO documents (title, content, embedding) VALUES (%s, %s, %s)",
    (title, content, embedding)  # embedding = list of floats
)

Vector Search + RAG Pipeline

import psycopg2
import openai

def rag_answer(user_question):
    # 1. Embed the question
    q_emb = oai.embeddings.create(
        input=user_question,
        model="text-embedding-3-small"
    ).data[0].embedding

    # 2. Find top-5 similar docs (cosine distance)
    cur.execute("""
        SELECT content, title,
               1 - (embedding <=> %s::vector) AS score
        FROM documents
        ORDER BY embedding <=> %s::vector
        LIMIT 5
    """, [str(q_emb), str(q_emb)])

    chunks = cur.fetchall()
    context = "\n\n".join(
        f"[{title}]: {content}"
        for content, title, score in chunks
    )

    # 3. Generate answer with context
    response = oai.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content":
                f"Answer using context:\n{context}"},
            {"role": "user", "content": user_question}
        ]
    )
    return response.choices[0].message.content