DEV Community: Andre Carbajal

Applying SAST to Any Application with CodeQL

Andre Carbajal — Tue, 30 Jun 2026 04:12:29 +0000

Applying SAST to Any Application with CodeQL

Static Application Security Testing (SAST) helps you inspect source code before an attacker ever reaches the running application. OWASP maintains a community page for Source Code Analysis Tools, which is a useful starting point when you need to compare static analysis options.

For this article, I intentionally avoid Sonar, Snyk, Semgrep, and Veracode. Instead, the hands-on example uses CodeQL, GitHub's semantic code analysis engine.

Repository for the demo: https://github.com/andre-carbajal/codeql-sast-demo

What we are building

The repo contains a small Node.js/Express app with educational vulnerabilities and a GitHub Actions workflow that runs CodeQL automatically.

The goal is not to build a production app. The goal is to show the repeatable pattern you can apply to almost any application:

Identify the language and build process.
Add a SAST workflow to CI.
Configure the query set.
Push code and review code scanning results.
Fix findings and keep the workflow running on every pull request.

Demo application

The Express app has a safe landing page and a few intentionally unsafe routes:

app.get('/profile', (request, response) => {
  const name = request.query.name || 'developer';
  response.send(`<h1>Hello ${name}</h1>`);
});

That route reflects user input into HTML. A safer variant escapes the output first:

app.get('/safe/profile', (request, response) => {
  const name = escapeHtml(request.query.name || 'developer');
  response.send(`<h1>Hello ${name}</h1>`);
});

The app also includes examples for command execution and file path handling so CodeQL has realistic security patterns to analyze.

Add CodeQL to GitHub Actions

Create .github/workflows/codeql.yml:

name: CodeQL SAST

on:
  push:
    branches: ["main"]
  pull_request:
    branches: ["main"]
  schedule:
    - cron: "31 7 * * 1"

permissions:
  contents: read

jobs:
  analyze:
    name: Analyze (${{ matrix.language }})
    runs-on: ubuntu-latest

    permissions:
      security-events: write
      packages: read
      actions: read
      contents: read

    strategy:
      fail-fast: false
      matrix:
        language: ["javascript-typescript"]

    steps:
      - name: Checkout repository
        uses: actions/checkout@v7

      - name: Initialize CodeQL
        uses: github/codeql-action/init@v4
        with:
          languages: ${{ matrix.language }}
          config-file: ./.github/codeql/codeql-config.yml

      - name: Autobuild
        uses: github/codeql-action/autobuild@v4

      - name: Perform CodeQL Analysis
        uses: github/codeql-action/analyze@v4
        with:
          category: "/language:${{ matrix.language }}"

For JavaScript and TypeScript, the important language identifier is javascript-typescript. If you apply this to a different stack, update the matrix language and make sure the repository can be built in CI.

Configure the query suite

Create .github/codeql/codeql-config.yml:

name: CodeQL SAST demo config

queries:
  - uses: security-extended
  - uses: security-and-quality

paths-ignore:
  - node_modules/**
  - test/**

The security-extended and security-and-quality suites make the scan broader than the default configuration. In a real production repository, start broad, review the signal-to-noise ratio, and then tune the config.

Run the demo locally

Clone the repository and run:

npm install
npm test
npm start

Then open:

http://localhost:3000

The tests focus on the safe path. The vulnerable routes are there so the SAST workflow has something meaningful to inspect.

Apply the same pattern to any application

Use this checklist when applying SAST to another project:

Pick the language: use the CodeQL language identifier for your app.
Understand the build: compiled languages may need a real build step before analysis.
Start with CI: run SAST on pull requests, pushes to main, and a scheduled scan.
Review findings in context: not every alert has the same business impact.
Fix and test: patch the vulnerable code and add regression tests.
Keep tuning: adjust ignored paths and query suites as the codebase grows.

What to look for in GitHub

After pushing the workflow, go to the repository on GitHub and check:

Actions: confirm the CodeQL SAST workflow finished successfully.
Security > Code scanning: review CodeQL alerts.
Pull requests: confirm future changes are scanned before merge.

Result from the demo repo

The first CodeQL run completed successfully and produced five open code scanning alerts in the intentionally vulnerable demo app:

js/path-injection: uncontrolled data used in a path expression.
js/command-line-injection: uncontrolled command line.
js/missing-rate-limiting: missing rate limiting on exposed routes.
js/reflected-xss: reflected cross-site scripting.

That is exactly what we want from a teaching repository: the workflow runs, uploads SARIF results to GitHub code scanning, and gives developers concrete findings to review.

Final thoughts

SAST is most valuable when it becomes part of the development workflow instead of a one-time audit. CodeQL works well for this because the configuration can live next to the code, run in GitHub Actions, and produce code scanning alerts where developers already review changes.

Use the demo repo as a starting point, then adapt the workflow and query configuration to your own language, build process, and risk profile.

Testing Management Tools Compared: Real-World Developer Examples

Andre Carbajal — Tue, 30 Jun 2026 03:21:25 +0000

Choosing a test management tool is rarely just a QA decision. Developers feel the consequences every day: how hard it is to publish automated results, how much context a failed test carries, whether CI artifacts are traceable, and whether test case IDs become useful metadata or bureaucratic friction.

This article compares five widely used testing management tools from a developer's point of view:

TestRail
Xray
Zephyr Scale
Azure Test Plans
qTest

The companion repository is public and runnable:

https://github.com/andre-carbajal/testing-management-tools-comparison

It includes a TypeScript + Playwright project that runs real tests, emits JUnit/JSON/HTML reports, converts Playwright output into a neutral TestRun schema, and generates local dry-run payloads for each tool. No vendor credentials are required.

Why test management tools still matter in CI/CD

Modern teams already have automated tests, pull requests, CI dashboards, and observability. So why add a test management layer?

Because CI answers what happened in this build, while test management answers broader questions:

Which requirements or Jira issues are covered by automated tests?
Which manual and automated checks belong to a release gate?
Which failures are new, repeated, waived, or blocked?
Which test cases are business-critical enough to audit?
Which teams own gaps in coverage?

The developer pain starts when the tool requires fragile scripts, manual exports, or hard-coded IDs scattered through test code. A good integration keeps automation-first workflows intact: tests run in CI, reports are archived, and the management tool receives only the metadata it needs.

Comparison table

Tool	Best fit	Developer integration model	Strengths	Tradeoffs
TestRail	Teams that want a standalone QA test repository	REST API result publishing, usually from CI	Clear test case/run model, mature reporting, easy to understand	Requires mapping automation IDs to TestRail case IDs; separate from issue trackers unless integrated
Xray	Jira-first teams	JUnit import or Xray JSON/import APIs	Strong Jira-native traceability across tests, executions, plans, and defects	Jira data model can become heavy; implementation differs between Cloud and Data Center
Zephyr Scale	Jira teams with growing test libraries	JUnit automation import and Zephyr API	Good Jira-native test organization and scalable test cycles	Teams must decide whether to auto-create cases or enforce curated mappings
Azure Test Plans	Teams already on Azure DevOps	Azure DevOps Test REST API, pipeline attachments, work-item links	Strong fit with Boards, Pipelines, and enterprise Microsoft environments	Less attractive outside Azure DevOps; API payloads are tied to project/run IDs
qTest	Enterprise QA governance across many stacks	qTest Manager APIs, Launch/Pulse flows, JUnit handoff artifacts	Strong portfolio-level process, auditability, and multi-tool governance	More setup overhead; exact workflow depends on qTest modules adopted by the organization

Real-world workflow

The repository implements this workflow:

flowchart TD
  A[Developer writes Playwright tests] --> B[CI runs npm test]
  B --> C[Playwright HTML report]
  B --> D[JUnit XML report]
  B --> E[Playwright JSON report]
  E --> F[Neutral TestRun schema]
  F --> G[Tool-specific dry-run payloads]
  D --> H[Tools that support JUnit import]

The important design choice is that test code does not know about every vendor. The tests include stable automation IDs like [CART-001], and a separate mapping layer translates those IDs into tool-specific identifiers.

Example test from tests/checkout.spec.ts:

test('[CART-001] adds a product to the cart and updates inventory', async ({ page }) => {
  await page.getByTestId('add-backpack').click();

  await expect(page.getByTestId('cart-count')).toHaveText('1');
  await expect(page.getByTestId('stock-count')).toHaveText('2');
  await expect(page.getByTestId('order-status')).toHaveText('Item added');
});

The Playwright config emits several reports in one run:

reporter: [
  ['list'],
  ['html', { outputFolder: 'playwright-report', open: 'never' }],
  ['junit', { outputFile: 'reports/junit/playwright-results.xml' }],
  ['json', { outputFile: 'reports/json/playwright-results.json' }],
  ...(process.env.CI ? [['github'] as const] : [])
]

That follows the Playwright reporter model documented by the Playwright project: local HTML for humans, JUnit for CI/tool imports, JSON for custom automation, and GitHub annotations in CI.

Neutral schema used by the repository

The repository defines the portable shape in src/schema.ts:

export interface TestCaseResult {
  id: string;
  title: string;
  status: TestStatus;
  durationMs: number;
  failureMessage?: string;
  source: {
    framework: 'playwright';
    suite: string;
    project?: string;
    file?: string;
  };
  links: {
    testRailCaseId?: number;
    jiraIssueKey?: string;
    azureTestCaseId?: number;
    qTestCaseId?: number;
  };
}

export interface ManagementToolAdapter {
  readonly tool: string;
  buildDryRun(run: TestRun): AdapterOutput;
}

This isolates vendor differences in src/adapters/*.ts. If a team later swaps tools, the tests and the Playwright report pipeline do not need to be rewritten.

Tool-by-tool implementation notes

TestRail

TestRail is the easiest mental model for many teams: create cases, organize them into runs, then publish results. The example adapter builds a dry-run request for the TestRail add_results_for_cases style of workflow:

{
  "method": "POST",
  "path": "/index.php?/api/v2/add_results_for_cases/:run_id",
  "body": {
    "results": [
      {
        "case_id": 1001,
        "status_id": 1,
        "elapsed": "1s",
        "comment": "CART-001 passed in Local Playwright demo run",
        "custom_automation_id": "CART-001"
      }
    ]
  }
}

Use TestRail when your team values a standalone QA repository with explicit test runs and readable reporting. The main engineering decision is the ID mapping strategy: do not bury numeric case IDs randomly across tests; centralize them in a mapping file or source them from metadata.

Xray

Xray fits teams that already use Jira as the shared system of record. Developers can keep producing standard JUnit reports while QA and product teams see executions, defects, requirements, and coverage in Jira.

The repository demonstrates two integration styles:

Upload reports/junit/playwright-results.xml through an Xray import endpoint.
Build an Xray-style execution payload from the neutral schema.

{
  "method": "POST",
  "path": "/api/v2/import/execution/junit",
  "query": {
    "projectKey": "SHOP",
    "testPlanKey": "SHOP-PLAN-1"
  },
  "fileUpload": "reports/junit/playwright-results.xml"
}

Choose Xray when Jira traceability is more important than tool independence. Be deliberate about test keys: a good key strategy makes failures searchable; a bad one turns Jira into a noisy dumping ground.

Zephyr Scale

Zephyr Scale also targets Jira-centric teams, but the workflow often emphasizes scalable test libraries, folders, cycles, and automation imports. The example adapter treats JUnit XML as the handoff artifact and shows an automation import request:

{
  "method": "POST",
  "path": "/v2/automations/executions/junit",
  "query": {
    "projectKey": "SHOP",
    "autoCreateTestCases": false
  },
  "fileUpload": "reports/junit/playwright-results.xml"
}

For demos, autoCreateTestCases can be convenient. In production, most teams should prefer explicit mappings so accidental test names do not create permanent test inventory.

Azure Test Plans

Azure Test Plans is the natural option when the organization already uses Azure DevOps Boards and Pipelines. Instead of making JUnit the only integration point, many teams use the Azure DevOps Test REST API to create or update runs and results.

The adapter shows two steps:

Create a test run.
Add results to that run.

{
  "method": "POST",
  "path": "/{organization}/{project}/_apis/test/Runs/:run_id/results",
  "query": {
    "api-version": "7.2-preview.6"
  },
  "body": [
    {
      "testCase": { "id": 5001 },
      "automatedTestName": "CART-001",
      "outcome": "Passed",
      "durationInMs": 42
    }
  ]
}

Choose Azure Test Plans if Azure DevOps is already your delivery backbone. If your team uses GitHub, Jira, or GitLab for most work, the integration may feel less natural.

qTest

qTest is strongest in enterprise QA environments where test governance spans many products, release trains, automation frameworks, and compliance requirements. The adapter demonstrates a qTest Manager-style log payload and also points to JUnit as a possible handoff artifact for qTest automation workflows.

{
  "method": "POST",
  "path": "/api/v3/projects/:project_id/test-runs/:test_run_id/test-logs",
  "body": [
    {
      "test_case_id": 9001,
      "automation_content": "CART-001",
      "status": "PASSED",
      "note": "CART-001 passed from Local Playwright demo run"
    }
  ]
}

Choose qTest when process governance, auditability, and multi-team consistency matter more than minimal setup. Developers should expect more environment-specific configuration than with a simple JUnit-only integration.

Recommendation matrix

Team situation	Recommended default	Why
Small team with no Jira dependency	TestRail	Clear model, easy CI publishing, fewer Jira assumptions
Jira is the center of delivery	Xray or Zephyr Scale	Test cases, executions, requirements, and defects stay close together
Azure DevOps is the delivery platform	Azure Test Plans	Native Boards/Pipelines/Test integration
Enterprise QA office governs multiple products	qTest	Stronger portfolio governance and process standardization
Developer team wants minimal lock-in	Keep JUnit + neutral JSON first	Standard artifacts preserve migration options

Developer checklist before adopting any tool

Define automation IDs. Use stable IDs like CART-001; do not rely on mutable test titles alone.
Centralize mappings. Keep CART-001 -> vendor case ID mappings in one place.
Retain standard artifacts. Always save JUnit, JSON, and HTML reports in CI even after vendor upload succeeds.
Separate dry-run from publish. Build payloads locally first; only publish when credentials and workspace IDs are configured.
Document status mapping. Decide how skipped, timed-out, flaky, and interrupted tests should appear in the management tool.
Avoid auto-creating test inventory by accident. Auto-create is fine for demos, risky for production governance.

How to run the example

git clone https://github.com/andre-carbajal/testing-management-tools-comparison.git
cd testing-management-tools-comparison
npm install
npm test
npm run adapters

After the run, inspect:

ls reports/json reports/junit example-payloads
cat example-payloads/neutral-test-run.json
cat example-payloads/testrail.json

Final takeaway

The best test management tool is not the one with the longest feature list. It is the one that fits your operating model while preserving a clean developer workflow.

For most teams, the safest architecture is:

Playwright or your test framework produces standard reports.
CI stores those reports as artifacts; this repo includes docs/github-actions/ci.yml as the workflow template.
A small adapter maps automation IDs to tool-specific IDs.
Publishing to the management tool is a separate, observable step.

That architecture keeps developers productive today and gives the organization room to change tools tomorrow.

Sources and further reading

Playwright reporters: https://github.com/microsoft/playwright/blob/main/docs/src/test-reporters-js.md
Playwright CI: https://github.com/microsoft/playwright/blob/main/docs/src/ci-intro.md
TestRail API results: https://support.testrail.com/hc/en-us/articles/7077819312404-Results
Xray Cloud REST API: https://docs.getxray.app/display/XRAYCLOUD/REST+API
Zephyr Scale Cloud API: https://support.smartbear.com/zephyr-scale-cloud/api-docs/
Azure DevOps Test REST API: https://learn.microsoft.com/en-us/rest/api/azure/devops/test/?view=azure-devops-rest-7.2
Tricentis qTest documentation: https://documentation.tricentis.com/qtest/

Applying API Testing Frameworks: Real-World Microservices Examples

Andre Carbajal — Tue, 30 Jun 2026 02:43:38 +0000

Most API testing tutorials test a single endpoint in isolation. Real systems are messier: services call each other over HTTP, share JWT tokens, and fail in ways that only show up when the whole thing runs together.

This article takes a different approach. We will build a small but realistic microservices system — user-service, product-service, and order-service — and then write a complete test suite for it using two different frameworks: Jest + Supertest on the Node.js side and Pytest + HTTPX on the Python side. Every code snippet in this article lives in a working repository at github.com/andre-carbajal/api-testing-microservices.

The system we are testing

Before writing a single test, it helps to understand what we are dealing with.

┌──────────────┐     JWT auth      ┌────────────────┐
│  user-service│ ◄───────────────► │  order-service │
│   :3001      │                   │   :3003        │
└──────────────┘                   └──────┬─────────┘
                                          │ stock check
                                          ▼
                                   ┌────────────────┐
                                   │product-service │
                                   │   :3002        │
                                   └────────────────┘

Service	Responsibility
user-service	Registration, JWT login, user profiles
product-service	Product catalog, stock management
order-service	Place orders; calls product-service to validate stock and decrement it

The order-service authenticates callers using the same JWT secret as the user-service — a common real-world pattern where auth is decentralized via a shared secret (or public key in production).

Two frameworks, two strategies

We are going to apply two distinct testing strategies, each matching what its framework is best suited for:

	Jest + Supertest	Pytest + HTTPX
Layer	Unit / in-process	Integration / E2E
Services running?	No — app imported directly	Yes — subprocesses
HTTP mocking?	Yes (order-service tests)	No — real HTTP
Speed	Fast (~1 s)	Slower (~5–10 s)
What it catches	Logic bugs, contract violations	Cross-service wiring issues

Both layers are necessary. Unit tests are fast and precise; integration tests prove the system actually works end-to-end.

Part 1 — Jest + Supertest (Node.js)

Why Supertest?

Supertest wraps your Express app and lets you make real HTTP requests against it without starting a server. It binds to an ephemeral port under the hood, so your tests are fast and completely isolated from the network.

cd tests/jest && npm install

Testing the user-service

The user-service handles registration, login, and profile retrieval. Let's look at the test for registration:

// tests/jest/user-service.test.js
const request = require("supertest");
const { app, users } = require("../../services/user-service/src/index");

// Helper to register a user
const registerUser = (overrides = {}) =>
  request(app)
    .post("/auth/register")
    .send({ name: "Alice", email: "alice@example.com", password: "Pass1234!", ...overrides });

beforeEach(() => {
  users.clear(); // reset in-memory state between tests
});

describe("POST /auth/register", () => {
  it("creates a new user and returns 201 with user data (no password)", async () => {
    const res = await registerUser();

    expect(res.status).toBe(201);
    expect(res.body).toMatchObject({ name: "Alice", email: "alice@example.com", role: "user" });
    expect(res.body).not.toHaveProperty("password"); // never expose the hash
  });

  it("returns 409 when email is already registered", async () => {
    await registerUser();
    const res = await registerUser(); // duplicate

    expect(res.status).toBe(409);
    expect(res.body.error).toMatch(/already registered/i);
  });
});

A few things worth noting here:

beforeEach(() => users.clear()) — the service exports its in-memory Map so tests can reset state without restarting the process. This is a deliberate design choice for testability.
expect(res.body).not.toHaveProperty("password") — this assertion is easy to forget but critical. A regression here leaks password hashes to clients.
toMatch(/already registered/i) — matching on a regex instead of an exact string makes the test resilient to minor copy changes.

Testing the login flow

describe("POST /auth/login", () => {
  beforeEach(async () => {
    await registerUser(); // ensure user exists
  });

  it("returns a JWT token on valid credentials", async () => {
    const res = await request(app)
      .post("/auth/login")
      .send({ email: "alice@example.com", password: "Pass1234!" });

    expect(res.status).toBe(200);
    expect(res.body).toHaveProperty("token");
    expect(typeof res.body.token).toBe("string");
  });

  it("returns 401 on wrong password", async () => {
    const res = await request(app)
      .post("/auth/login")
      .send({ email: "alice@example.com", password: "wrongpassword" });

    expect(res.status).toBe(401);
  });
});

Mocking inter-service HTTP in the order-service

This is where things get interesting. The order-service calls the product-service over HTTP to validate stock and decrement it after a successful order. In unit tests, we do not want a real product-service running. We want to control exactly what it returns.

The service is designed with a seam for exactly this purpose:

// services/order-service/src/index.js (simplified)
let httpClient = null;

const getHttpClient = () => {
  if (httpClient) return httpClient; // injected mock
  return require("node-fetch");      // real in production
};

module.exports = { app, orders, setHttpClient };

In tests, we inject a jest.fn() that mimics the product-service responses:

// tests/jest/order-service.test.js
const { app, orders, setHttpClient } = require("../../services/order-service/src/index");

const mockFetch = (productOverrides = {}) => {
  const defaultProduct = {
    id: 1, name: "Mechanical Keyboard", price: 129.99, stock: 50, ...productOverrides,
  };

  return jest.fn().mockImplementation((url, options = {}) => {
    const method = options.method || "GET";

    if (method === "GET" && url.includes("/products/")) {
      return Promise.resolve({
        ok: true,
        json: () => Promise.resolve(defaultProduct),
      });
    }

    if (method === "PATCH" && url.includes("/stock")) {
      return Promise.resolve({
        ok: true,
        json: () => Promise.resolve({ ...defaultProduct, stock: defaultProduct.stock - 2 }),
      });
    }
  });
};

afterEach(() => {
  setHttpClient(null); // restore real fetch
});

Now we can test the "insufficient stock" path without touching any real service:

it("returns 422 when stock is insufficient", async () => {
  setHttpClient(mockFetch({ stock: 1 })); // only 1 unit in stock
  const token = makeToken();

  const res = await request(app)
    .post("/orders")
    .set("Authorization", `Bearer ${token}`)
    .send({ items: [{ productId: 1, quantity: 5 }] }); // requesting 5

  expect(res.status).toBe(422);
  expect(res.body.error).toMatch(/insufficient stock/i);
});

And the happy path, verifying the total is calculated correctly:

it("creates an order and returns 201 with confirmed status", async () => {
  setHttpClient(mockFetch()); // price: 129.99
  const token = makeToken();

  const res = await request(app)
    .post("/orders")
    .set("Authorization", `Bearer ${token}`)
    .send({ items: [{ productId: 1, quantity: 2 }] });

  expect(res.status).toBe(201);
  expect(res.body).toMatchObject({
    status: "confirmed",
    userId: 1,
    total: 259.98, // 129.99 * 2
  });
});

Run the Jest suite

cd tests/jest
npm test

You should see something like:

PASS  user-service.test.js
PASS  product-service.test.js
PASS  order-service.test.js

Test Suites: 3 passed, 3 total
Tests:       22 passed, 22 total
Time:        1.4 s

Part 2 — Pytest + HTTPX (Python)

Why HTTPX?

HTTPX is a modern HTTP client for Python with a clean async API. Paired with pytest-asyncio, it lets you write async test functions that feel natural:

async def test_something(client):
    res = await client.get("/health")
    assert res.status_code == 200

Unlike the Jest tests that import the app directly, Pytest tests run against real running services. This means we need to start them first.

pip install -r tests/pytest/requirements.txt

conftest.py — shared fixtures

conftest.py is where Pytest fixtures live. Ours does three things: start the Node services as subprocesses, wait for them to be healthy, and expose async helpers for auth.

# tests/pytest/conftest.py
import subprocess, time, os
import httpx, pytest

USER_URL = os.getenv("USER_SERVICE_URL", "http://localhost:3001")
PRODUCT_URL = os.getenv("PRODUCT_SERVICE_URL", "http://localhost:3002")
ORDER_URL = os.getenv("ORDER_SERVICE_URL", "http://localhost:3003")


def _wait_for_service(url: str, retries: int = 20, delay: float = 0.5) -> None:
    for _ in range(retries):
        try:
            r = httpx.get(f"{url}/health", timeout=2)
            if r.status_code == 200:
                return
        except httpx.TransportError:
            pass
        time.sleep(delay)
    raise RuntimeError(f"Service at {url} did not become healthy in time")


@pytest.fixture(scope="session")
def services():
    """Start all three Node services once per test session."""
    procs = []
    root = os.path.abspath(os.path.join(os.path.dirname(__file__), "../.."))

    for rel_dir, port in [
        ("services/user-service", 3001),
        ("services/product-service", 3002),
        ("services/order-service", 3003),
    ]:
        p = subprocess.Popen(
            ["node", "src/index.js"],
            cwd=os.path.join(root, rel_dir),
            env={**os.environ, "PORT": str(port)},
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
        procs.append(p)

    for url in [USER_URL, PRODUCT_URL, ORDER_URL]:
        _wait_for_service(url)

    yield {"user": USER_URL, "product": PRODUCT_URL, "order": ORDER_URL}

    for p in procs:
        p.terminate(); p.wait()


@pytest.fixture
async def client():
    async with httpx.AsyncClient(timeout=10) as c:
        yield c


@pytest.fixture
async def auth_token(client, services, registered_user):
    res = await client.post(
        f"{services['user']}/auth/login",
        json={"email": registered_user["email"], "password": registered_user["password"]},
    )
    assert res.status_code == 200
    return res.json()["token"]

The scope="session" on the services fixture means the services start once and stay up for the entire test run — much faster than restarting them for every test.

Integration tests for the user-service

# tests/pytest/test_user_service.py
import pytest

pytestmark = pytest.mark.asyncio


class TestRegister:
    async def test_register_returns_201_with_user_data(self, client, services):
        res = await client.post(
            f"{services['user']}/auth/register",
            json={"name": "Bob", "email": "bob@pytest.com", "password": "PyTest1234!"},
        )
        assert res.status_code == 201
        body = res.json()
        assert body["email"] == "bob@pytest.com"
        assert body["role"] == "user"
        assert "password" not in body  # never expose the hash

    async def test_register_returns_409_on_duplicate_email(self, client, services, registered_user):
        res = await client.post(
            f"{services['user']}/auth/register",
            json=registered_user,  # same email as the fixture
        )
        assert res.status_code == 409
        assert "already registered" in res.json()["error"].lower()


class TestLogin:
    async def test_login_returns_jwt_with_three_segments(self, client, services, registered_user):
        res = await client.post(
            f"{services['user']}/auth/login",
            json={"email": registered_user["email"], "password": registered_user["password"]},
        )
        assert res.status_code == 200
        token = res.json().get("token")
        assert token is not None
        assert len(token.split(".")) == 3  # JWTs always have three base64 segments

The crown jewel: end-to-end order flow

This test exercises the entire system in a single scenario — register, login, browse, order, verify stock, verify history:

# tests/pytest/test_order_flow.py
import pytest

pytestmark = pytest.mark.asyncio


class TestFullOrderFlow:
    async def test_end_to_end_order_flow(self, client, services):
        user_url, product_url, order_url = (
            services["user"], services["product"], services["order"]
        )

        # Step 1: Register
        reg_res = await client.post(
            f"{user_url}/auth/register",
            json={"name": "Charlie", "email": "charlie@e2e.com", "password": "E2ETest99!"},
        )
        assert reg_res.status_code in (201, 409)

        # Step 2: Login
        login_res = await client.post(
            f"{user_url}/auth/login",
            json={"email": "charlie@e2e.com", "password": "E2ETest99!"},
        )
        assert login_res.status_code == 200
        token = login_res.json()["token"]
        headers = {"Authorization": f"Bearer {token}"}

        # Step 3: Browse products
        products_res = await client.get(f"{product_url}/products")
        assert products_res.status_code == 200
        products = products_res.json()
        item_a, item_b = products[0], products[1]
        stock_before = item_a["stock"]

        # Step 4: Place an order
        order_res = await client.post(
            f"{order_url}/orders",
            headers=headers,
            json={"items": [
                {"productId": item_a["id"], "quantity": 2},
                {"productId": item_b["id"], "quantity": 1},
            ]},
        )
        assert order_res.status_code == 201
        order = order_res.json()
        assert order["status"] == "confirmed"
        assert len(order["items"]) == 2
        assert order["total"] > 0

        # Step 5: Verify stock was decremented
        stock_res = await client.get(f"{product_url}/products/{item_a['id']}")
        assert stock_res.json()["stock"] == stock_before - 2

        # Step 6: Verify order isolation — another user cannot access this order
        reg2 = await client.post(
            f"{user_url}/auth/register",
            json={"name": "Dave", "email": "dave@e2e.com", "password": "E2ETest99!"},
        )
        login2 = await client.post(
            f"{user_url}/auth/login",
            json={"email": "dave@e2e.com", "password": "E2ETest99!"},
        )
        headers2 = {"Authorization": f"Bearer {login2.json()['token']}"}
        forbidden = await client.get(f"{order_url}/orders/{order['id']}", headers=headers2)
        assert forbidden.status_code == 403

This single test would have caught the order-service stock-decrement bug, the JWT forwarding wiring, and the order isolation rule — all at once. That is the power of an end-to-end test.

Run the Pytest suite

cd tests/pytest
pytest -v

Output:

tests/pytest/test_user_service.py::TestRegister::test_register_returns_201_with_user_data PASSED
tests/pytest/test_user_service.py::TestLogin::test_login_returns_jwt_with_three_segments PASSED
tests/pytest/test_order_flow.py::TestFullOrderFlow::test_end_to_end_order_flow PASSED
tests/pytest/test_order_flow.py::TestFullOrderFlow::test_order_isolation_between_users PASSED

5 passed in 6.4s

Key patterns to take away

1. Design services for testability

Export internal state (like the in-memory users Map) so tests can reset it. Add a dependency injection seam for HTTP clients so you can mock outbound calls. These are not hacks — they are good design.

// Export state for test teardown
module.exports = { app, users };

// Export a setter so tests can inject a mock HTTP client
module.exports = { app, orders, setHttpClient };

2. Use `beforeEach` to reset state, not test ordering

Never rely on tests running in a specific order. Each test should set up everything it needs and clean up after itself. The beforeEach(() => users.clear()) pattern enforces this.

3. Match the right tool to the right layer

Supertest: fast, in-process, ideal for unit-testing Express route logic and middleware.
HTTPX async: clean async API for integration and E2E tests against real running services.
Do not try to use one tool for everything.

4. Test the contract, not the implementation

A test like expect(res.body).not.toHaveProperty("password") documents a security contract. A test that checks the exact SQL query being run is brittle and tests the wrong thing.

5. Layer your test pyramid

          ┌─────────┐
          │   E2E   │ ← Pytest + HTTPX (few, slow, high confidence)
         ┌┴─────────┴─┐
         │Integration │ ← Pytest + HTTPX (services + real HTTP)
        ┌┴────────────┴┐
        │    Unit      │ ← Jest + Supertest (many, fast, isolated)
        └──────────────┘

Most of your tests should be at the bottom. A handful of E2E tests at the top catch wiring issues that unit tests cannot.

Repository

All the code from this article is available at:

👉 github.com/andre-carbajal/api-testing-microservices

The repository includes:

Three fully working Express microservices
22 Jest + Supertest unit tests
Pytest integration and E2E tests with async fixtures
A conftest.py that manages service lifecycle automatically

Clone it, run the tests, break something, and see which test catches it first.

Wrapping up

Testing APIs in a microservices architecture is not just about making requests and checking status codes. It is about choosing the right layer for each concern:

Use Jest + Supertest to test your route logic fast, in isolation, with full control over dependencies.
Use Pytest + HTTPX to prove your services wire together correctly in a real running environment.
Design your services with testability in mind — exportable state, injectable HTTP clients, and clean health endpoints.

The patterns here scale well: whether you have three services or thirty, the same layered approach — many fast unit tests, fewer slower integration tests, a handful of E2E smoke tests — gives you confidence without making your test suite painful to run.

Found this useful? The repo is open source — PRs and issues welcome.

1. Design services for testability

// Export state for test teardown
module.exports = { app, users };

// Export a setter so tests can inject a mock HTTP client
module.exports = { app, orders, setHttpClient };

2. Use `beforeEach` to reset state, not test ordering

Never rely on tests running in a specific order. Each test should set up everything it needs and clean up after itself. The beforeEach(() => users.clear()) pattern enforces this.

3. Match the right tool to the right layer

Supertest: fast, in-process, ideal for unit-testing Express route logic and middleware.
HTTPX async: clean async API for integration and E2E tests against real running services.
Do not try to use one tool for everything.

4. Test the contract, not the implementation

A test like expect(res.body).not.toHaveProperty("password") documents a security contract. A test that checks the exact SQL query being run is brittle and tests the wrong thing.

5. Layer your test pyramid

          ┌─────┐
          │ E2E │  ← Pytest + HTTPX (few, slow, high confidence)
         ┌┴─────┴┐
         │Integration│ ← Pytest + HTTPX (services + real HTTP)
        ┌┴──────────┴┐
        │    Unit    │ ← Jest + Supertest (many, fast, isolated)
        └────────────┘

Most of your tests should be at the bottom. A handful of E2E tests at the top catch wiring issues that unit tests cannot.

Repository

All the code from this article is available at:

👉 github.com/andre-carbajal/api-testing-microservices

The repository includes:

Three fully working Express microservices
22 Jest + Supertest unit tests
Pytest integration and E2E tests with async fixtures
A conftest.py that manages service lifecycle automatically

Clone it, run the tests, break something, and see which test catches it first.

Wrapping up

Testing APIs in a microservices architecture is not just about making requests and checking status codes. It is about choosing the right layer for each concern:

Use Jest + Supertest to test your route logic fast, in isolation, with full control over dependencies.
Use Pytest + HTTPX to prove your services wire together correctly in a real running environment.
Design your services with testability in mind — exportable state, injectable HTTP clients, and clean health endpoints.

Found this useful? The repo is open source — PRs and issues welcome.

SQL + AI: Real-World Database Solutions You Can Use Today

Andre Carbajal — Sat, 27 Jun 2026 20:49:39 +0000

📦 All code examples in this article are available on GitHub:
github.com/andre-carbajal/sql-ai-database-solutions

Databases and AI used to live in separate worlds. Your PostgreSQL instance handled structured queries, and your ML pipeline ran somewhere else entirely. That separation is rapidly disappearing.

In 2025, your SQL database can store embeddings, answer natural language questions, power semantic search, and serve as the memory layer for autonomous AI agents — all without adding a separate vector database to your stack.

This article walks through four real-world patterns with working code:

pgvector — storing and querying embeddings in PostgreSQL
RAG pipelines — augmenting LLMs with your own database content
NL2SQL / Text-to-SQL — letting users query your database in plain English
AI Agents over SQL — autonomous agents that reason and act on your data

Prerequisites

pip install psycopg2-binary pgvector langchain langchain-openai openai sqlalchemy python-dotenv

# Docker: run PostgreSQL with pgvector support
docker run --name pgvector-demo \
  -e POSTGRES_PASSWORD=secret \
  -p 5432:5432 \
  -d pgvector/pgvector:pg16

Set your environment variables:

# .env
OPENAI_API_KEY=sk-...
DATABASE_URL=postgresql://postgres:secret@localhost:5432/aidb

1. pgvector — Semantic Search Inside PostgreSQL

pgvector is a PostgreSQL extension that adds a native vector data type, allowing you to store high-dimensional embeddings and run similarity searches with standard SQL.

Why this matters

Instead of maintaining a separate vector database (Pinecone, Weaviate, Chroma), you keep your vectors beside your relational data. That means you can combine semantic search with SQL filters in a single query.

Setup

-- Enable the extension
CREATE EXTENSION IF NOT EXISTS vector;

-- Create a products table with an embedding column
CREATE TABLE products (
    id         SERIAL PRIMARY KEY,
    name       TEXT NOT NULL,
    description TEXT,
    category   TEXT,
    price      DECIMAL(10, 2),
    embedding  VECTOR(1536)   -- OpenAI text-embedding-3-small dimensions
);

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

Inserting embeddings with Python

import os
import psycopg2
import numpy as np
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
conn = psycopg2.connect(os.environ["DATABASE_URL"])
cur = conn.cursor()

def get_embedding(text: str) -> list[float]:
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding

products = [
    ("Wireless Noise-Cancelling Headphones", "Premium audio with 30hr battery", "electronics", 299.99),
    ("Ergonomic Office Chair",              "Lumbar support, adjustable arms",  "furniture",   449.00),
    ("Python Programming Book",             "Comprehensive guide for beginners", "books",        39.99),
    ("Mechanical Keyboard",                 "Cherry MX switches, RGB backlit",   "electronics", 129.00),
    ("Standing Desk",                       "Electric height-adjustable desk",   "furniture",   599.00),
]

for name, description, category, price in products:
    embedding = get_embedding(f"{name}: {description}")
    cur.execute(
        """
        INSERT INTO products (name, description, category, price, embedding)
        VALUES (%s, %s, %s, %s, %s)
        """,
        (name, description, category, price, embedding)
    )

conn.commit()
print("Products inserted with embeddings.")

Querying: hybrid semantic + structured SQL

def semantic_search(query: str, max_price: float = None, category: str = None, limit: int = 5):
    query_embedding = get_embedding(query)

    sql = """
        SELECT
            name,
            description,
            category,
            price,
            1 - (embedding <=> %s::vector) AS similarity
        FROM products
        WHERE 1=1
    """
    params = [query_embedding]

    if max_price:
        sql += " AND price <= %s"
        params.append(max_price)

    if category:
        sql += " AND category = %s"
        params.append(category)

    sql += " ORDER BY embedding <=> %s::vector LIMIT %s"
    params += [query_embedding, limit]

    cur.execute(sql, params)
    return cur.fetchall()

# "I want something for my home office under $500"
results = semantic_search(
    query="comfortable workspace setup for long work sessions",
    max_price=500.0
)

for name, description, category, price, similarity in results:
    print(f"[{similarity:.3f}] {name} (${price}) — {description}")

Output:

[0.891] Ergonomic Office Chair ($449.00) — Lumbar support, adjustable arms
[0.832] Mechanical Keyboard ($129.00) — Cherry MX switches, RGB backlit
[0.801] Standing Desk ($599.00) — Electric height-adjustable desk

The <=> operator computes cosine distance. pgvector also supports <-> (L2/Euclidean) and <#> (inner product).

2. RAG Pipeline — Your Database as an LLM Knowledge Base

Retrieval-Augmented Generation (RAG) lets you inject your own data into an LLM's context at query time. Instead of fine-tuning, you retrieve the most relevant chunks from your database and pass them as context.

The architecture

User Question
     │
     ▼
Embed the Question
     │
     ▼
Vector Search in PostgreSQL  ──▶  Top-K relevant chunks
     │
     ▼
Build LLM Prompt with context
     │
     ▼
LLM generates answer grounded in your data

Full RAG pipeline with LangChain + pgvector

from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import PGVector
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.chains import RetrievalQA
from langchain_community.document_loaders import TextLoader

# 1. Load and chunk your documents
loader = TextLoader("company_docs.txt")
documents = loader.load()

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50
)
chunks = splitter.split_documents(documents)

# 2. Store chunks + embeddings in PostgreSQL
CONNECTION_STRING = os.environ["DATABASE_URL"]
COLLECTION_NAME = "company_knowledge_base"

vector_store = PGVector.from_documents(
    documents=chunks,
    embedding=OpenAIEmbeddings(model="text-embedding-3-small"),
    collection_name=COLLECTION_NAME,
    connection_string=CONNECTION_STRING,
)

# 3. Build a retrieval-augmented QA chain
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
retriever = vector_store.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 4}
)

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    return_source_documents=True
)

# 4. Ask questions
result = qa_chain.invoke({"query": "What is our refund policy?"})
print(result["result"])
print("\nSources:")
for doc in result["source_documents"]:
    print(f"  - {doc.metadata.get('source', 'unknown')}")

Adding metadata filters for multi-tenant RAG

# Store documents with tenant metadata
vector_store = PGVector.from_documents(
    documents=chunks,
    embedding=OpenAIEmbeddings(),
    collection_name="docs",
    connection_string=CONNECTION_STRING,
    # Each chunk carries metadata you can filter on
)

# Retrieve only documents for a specific tenant
retriever = vector_store.as_retriever(
    search_kwargs={
        "k": 5,
        "filter": {"tenant_id": "acme-corp"}
    }
)

3. NL2SQL — Natural Language to SQL Queries

NL2SQL (also called Text-to-SQL) lets users query your database in plain English. An LLM reads your schema and translates a natural language question into a valid SQL query.

Naive approach: direct schema + prompt

from openai import OpenAI

client = OpenAI()

SCHEMA = """
Tables:
  orders(id, customer_id, product_id, quantity, total_price, created_at, status)
  customers(id, name, email, country, created_at)
  products(id, name, category, price, stock)
"""

def nl_to_sql(question: str) -> str:
    prompt = f"""You are a SQL expert. Given the following database schema, write a PostgreSQL query to answer the user's question.
Return ONLY the SQL query, no explanation.

Schema:
{SCHEMA}

Question: {question}

SQL:"""

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0
    )
    return response.choices[0].message.content.strip()


# Real-world examples
questions = [
    "What are the top 5 countries by total revenue this year?",
    "Which products have never been ordered?",
    "Show me customers who placed more than 3 orders in the last 30 days",
]

for q in questions:
    print(f"Q: {q}")
    print(f"SQL: {nl_to_sql(q)}")
    print()

Output:

-- "What are the top 5 countries by total revenue this year?"
SELECT c.country, SUM(o.total_price) AS total_revenue
FROM orders o
JOIN customers c ON o.customer_id = c.id
WHERE EXTRACT(YEAR FROM o.created_at) = EXTRACT(YEAR FROM CURRENT_DATE)
GROUP BY c.country
ORDER BY total_revenue DESC
LIMIT 5;

-- "Which products have never been ordered?"
SELECT p.id, p.name, p.category
FROM products p
LEFT JOIN orders o ON p.id = o.product_id
WHERE o.id IS NULL;

Production-grade NL2SQL with validation

Never run LLM-generated SQL without validation. Here's a safer pattern:

import sqlparse
import re

ALLOWED_STATEMENTS = {"SELECT"}

def is_safe_sql(sql: str) -> bool:
    """Reject anything that isn't a SELECT statement."""
    parsed = sqlparse.parse(sql)
    if not parsed:
        return False
    statement_type = parsed[0].get_type()
    return statement_type in ALLOWED_STATEMENTS

def execute_nl_query(question: str, conn) -> list[dict]:
    sql = nl_to_sql(question)

    # Strip markdown fences if LLM added them
    sql = re.sub(r"```

sql|

```", "", sql).strip()

    if not is_safe_sql(sql):
        raise ValueError(f"Unsafe SQL rejected: {sql}")

    cur = conn.cursor()
    cur.execute(sql)
    columns = [desc[0] for desc in cur.description]
    rows = cur.fetchall()
    return [dict(zip(columns, row)) for row in rows]

Few-shot prompting for better accuracy

Adding examples dramatically improves query accuracy on complex schemas:

FEW_SHOT_EXAMPLES = """
Example 1:
Question: How many orders were placed last month?
SQL: SELECT COUNT(*) FROM orders WHERE created_at >= DATE_TRUNC('month', NOW() - INTERVAL '1 month') AND created_at < DATE_TRUNC('month', NOW());

Example 2:
Question: What is the average order value by product category?
SQL: SELECT p.category, AVG(o.total_price) AS avg_order_value FROM orders o JOIN products p ON o.product_id = p.id GROUP BY p.category ORDER BY avg_order_value DESC;
"""

def nl_to_sql_fewshot(question: str) -> str:
    prompt = f"""You are a SQL expert for a PostgreSQL e-commerce database.

Schema:
{SCHEMA}

{FEW_SHOT_EXAMPLES}

Question: {question}
SQL:"""
    # ... same API call as before

4. AI Agents over SQL — Autonomous Database Reasoning

The most powerful pattern: an AI agent that can autonomously decide which queries to run, inspect the results, and iterate to answer complex multi-step questions.

Building a SQL agent with LangChain + custom tools

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain import hub
import psycopg2

conn = psycopg2.connect(os.environ["DATABASE_URL"])

@tool
def query_database(sql: str) -> str:
    """
    Execute a read-only SQL SELECT query against the e-commerce database.
    Use this to retrieve data needed to answer user questions.
    Never use INSERT, UPDATE, DELETE, or DROP.
    """
    if not is_safe_sql(sql):
        return "Error: Only SELECT queries are allowed."
    try:
        cur = conn.cursor()
        cur.execute(sql)
        columns = [desc[0] for desc in cur.description]
        rows = cur.fetchmany(50)  # cap results
        result = [dict(zip(columns, row)) for row in rows]
        return str(result)
    except Exception as e:
        return f"Query error: {e}"

@tool
def get_schema() -> str:
    """
    Returns the database schema — table names, columns, and types.
    Call this first to understand what data is available before writing queries.
    """
    cur = conn.cursor()
    cur.execute("""
        SELECT table_name, column_name, data_type
        FROM information_schema.columns
        WHERE table_schema = 'public'
        ORDER BY table_name, ordinal_position;
    """)
    rows = cur.fetchall()
    schema = {}
    for table, column, dtype in rows:
        schema.setdefault(table, []).append(f"{column} ({dtype})")
    return "\n".join(
        f"{table}: {', '.join(cols)}"
        for table, cols in schema.items()
    )

@tool
def semantic_product_search(query: str) -> str:
    """
    Search for products using semantic/natural language similarity.
    Use this when the user describes what they want rather than specifying exact product names.
    """
    results = semantic_search(query, limit=5)
    return str([{"name": r[0], "description": r[1], "price": r[3]} for r in results])

# Assemble the agent
llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [get_schema, query_database, semantic_product_search]

prompt = hub.pull("hwchase17/openai-tools-agent")

agent = create_openai_tools_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=10
)

# Complex multi-step question
response = agent_executor.invoke({
    "input": """
    I'm looking for ergonomic office products. 
    Find the top 3 most relevant products, then check if we have customers 
    from Germany who bought office furniture in the past 6 months. 
    Give me a summary of both findings.
    """
})

print(response["output"])

The agent will autonomously:

Call semantic_product_search("ergonomic office products")
Call get_schema() to understand the tables
Write and run a SQL query for German office furniture buyers
Synthesize both results into a coherent answer

Putting It All Together: Architecture Overview

┌─────────────────────────────────────────────────────┐
│                   PostgreSQL Database               │
│                                                     │
│  ┌─────────────┐  ┌──────────────┐  ┌───────────┐  │
│  │  products   │  │  embeddings  │  │  orders   │  │
│  │  customers  │  │  (pgvector)  │  │  ...      │  │
│  └─────────────┘  └──────────────┘  └───────────┘  │
└───────────────────────┬─────────────────────────────┘
                        │
           ┌────────────┼────────────┐
           ▼            ▼            ▼
     ┌──────────┐ ┌──────────┐ ┌──────────┐
     │ Semantic │ │  NL2SQL  │ │  RAG     │
     │  Search  │ │  Agent   │ │ Pipeline │
     └──────────┘ └──────────┘ └──────────┘
           │            │            │
           └────────────┼────────────┘
                        ▼
                  ┌──────────┐
                  │  LLM     │
                  │ (GPT-4o) │
                  └──────────┘
                        │
                        ▼
                  User Response

Performance Tips

Index strategy for pgvector:

-- HNSW: faster queries, slower inserts — best for production read-heavy workloads
CREATE INDEX ON products USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

-- IVFFlat: faster inserts, slightly less accurate — good for bulk loading
CREATE INDEX ON products USING ivfflat (embedding vector_l2_ops)
WITH (lists = 100);
-- Rule of thumb: lists ≈ sqrt(number_of_rows)

Batching embeddings:

# Don't call the embeddings API one row at a time
texts = [f"{p['name']}: {p['description']}" for p in products]

response = client.embeddings.create(
    model="text-embedding-3-small",
    input=texts  # batch up to 2048 inputs at once
)
embeddings = [item.embedding for item in response.data]

Caching repeated NL2SQL queries:

import hashlib
import json

query_cache: dict[str, str] = {}

def cached_nl_to_sql(question: str) -> str:
    key = hashlib.md5(question.lower().strip().encode()).hexdigest()
    if key not in query_cache:
        query_cache[key] = nl_to_sql(question)
    return query_cache[key]

Security Checklist

Before shipping any of these patterns to production:

✅ Allowlist only SELECT — reject DDL and DML from LLM-generated SQL
✅ Use parameterized queries when executing with user-supplied values
✅ Create a read-only DB user for your AI layer
✅ Set query timeouts to prevent expensive accidental full-table scans
✅ Cap result set sizes from agent tool calls
✅ Log all LLM-generated queries for auditing
✅ Never expose raw error messages — they leak schema information

-- Create a restricted role for your AI application
CREATE ROLE ai_readonly;
GRANT CONNECT ON DATABASE aidb TO ai_readonly;
GRANT USAGE ON SCHEMA public TO ai_readonly;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO ai_readonly;
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO ai_readonly;

Public Repository

All the code from this article — including Docker Compose setup, schema migrations, and ready-to-run examples — is in the companion repository:

🔗 github.com/andre-carbajal/sql-ai-database-solutions

The repo structure:

sql-ai-database-solutions/
├── docker-compose.yml          # PostgreSQL + pgvector, ready to go
├── schema/
│   └── 001_init.sql            # Tables, pgvector extension, indexes
├── examples/
│   ├── 01_pgvector_search.py   # Semantic search
│   ├── 02_rag_pipeline.py      # RAG with LangChain
│   ├── 03_nl2sql.py            # Natural language to SQL
│   └── 04_sql_agent.py         # Autonomous SQL agent
├── .env.example
└── README.md

What's Next?

The patterns here are just the start. Some directions worth exploring:

DB-GPT — an open-source, full-stack AI database framework
Vanna.AI — RAG-based NL2SQL with self-learning capabilities
pgai — Timescale's extension for running ML directly inside PostgreSQL
Multi-agent SQL workflows — agents that collaborate across multiple databases

The line between "database" and "AI system" is getting thinner every month. If you're already running PostgreSQL, you're closer to a production AI stack than you might think.

Have questions or want to share what you've built? Drop a comment below or open an issue on the repo!

DEV Community: Andre Carbajal

Applying SAST to Any Application with CodeQL

Applying SAST to Any Application with CodeQL

What we are building

Demo application

Add CodeQL to GitHub Actions

Configure the query suite

Run the demo locally

Apply the same pattern to any application

What to look for in GitHub

Result from the demo repo

Final thoughts

Testing Management Tools Compared: Real-World Developer Examples

Why test management tools still matter in CI/CD

Comparison table

Real-world workflow

Neutral schema used by the repository

Tool-by-tool implementation notes

TestRail

Xray

Zephyr Scale

Azure Test Plans

qTest

Recommendation matrix

Developer checklist before adopting any tool

How to run the example

Final takeaway

Sources and further reading

Applying API Testing Frameworks: Real-World Microservices Examples

The system we are testing

Two frameworks, two strategies

Part 1 — Jest + Supertest (Node.js)

Why Supertest?

Testing the user-service

Testing the login flow

Mocking inter-service HTTP in the order-service

Run the Jest suite

Part 2 — Pytest + HTTPX (Python)

Why HTTPX?

conftest.py — shared fixtures

Integration tests for the user-service

The crown jewel: end-to-end order flow

Run the Pytest suite

Key patterns to take away

1. Design services for testability

2. Use beforeEach to reset state, not test ordering

3. Match the right tool to the right layer

4. Test the contract, not the implementation

5. Layer your test pyramid

Repository

Wrapping up

1. Design services for testability

2. Use beforeEach to reset state, not test ordering

3. Match the right tool to the right layer

4. Test the contract, not the implementation

5. Layer your test pyramid

Repository

Wrapping up

SQL + AI: Real-World Database Solutions You Can Use Today

Prerequisites

1. pgvector — Semantic Search Inside PostgreSQL

Why this matters

Setup

Inserting embeddings with Python

Querying: hybrid semantic + structured SQL

2. RAG Pipeline — Your Database as an LLM Knowledge Base

The architecture

Full RAG pipeline with LangChain + pgvector

Adding metadata filters for multi-tenant RAG

3. NL2SQL — Natural Language to SQL Queries

Naive approach: direct schema + prompt

Production-grade NL2SQL with validation

Few-shot prompting for better accuracy

4. AI Agents over SQL — Autonomous Database Reasoning

Building a SQL agent with LangChain + custom tools

Putting It All Together: Architecture Overview

Performance Tips

Security Checklist

Public Repository

What's Next?

2. Use `beforeEach` to reset state, not test ordering

2. Use `beforeEach` to reset state, not test ordering