From Routing Checks to Trajectory Testing: Evaluating an Agentic Chatbot

Which Agentic Chatbot?

I have been working on a Python based AI test framework for a chatbot interface for my company's product, Postgres AI Hybrid Manager. The manager allows the setup of Postgres clusters across cloud or on-prem and attaching various AI tools such as Langflow.  So a combination of more traditional Postgres backup, migration, telemetry and analytics features along with LLM workflows leveraging the data it holds.

The product already has a control plane UI for managing Postgres estates. It also has full help for the product, all Postgres versions, analytics, AI and add ons. The chatbot brings all these things together: ask a question, get the relevant help, or ask it to do something such as migrate a cluster, or evaluate telemetry that would otherwise require clicking through the UI.

That makes it a pretty handy interface, especially for the less technical. However it is not a simple to test and ensure good quality responses.

A normal deterministic API test is simple. Send a request, check the status code, check the JSON body, perhaps check the database state. An LLM-backed agent does not pass or fail so clearly. It can route to the wrong capability and still return fluent text. It can pick a plausible but wrong tool. It can miss half the task and still sound confident. It can complete the first turn of a conversation and lose the plot on the second. It could get malformed or missing data from tooling that leads it to deliver a misleading conclusion. It might only provide help to something that should be from tool data or was a request for an action such as create a cluster.

So the testing problem was not “does the chatbot return a reasonable response?” It was “how do we test the whole chat path is doing the right thing?”

This is the story of how our agent-eval test framework evolved as we worked to see that our chatbot was not only getting the right answer,  42 , but whether it was asking all the right questions of the right tools to get that answer. Known as trajectory testing ...

 You're Golden 

Before we can tell our story we need to define some terms.

A Golden is an example of a perfect desired output from a test input. They often refer to more complex outputs that may need saving as separate files, but a simple assertable output such as 42, is a golden too!
Whilst complex goldens may be used and marked for semantic similarity against the test output. It is more common for complex outputs to be described by a rubric. A rubric is a checklist of qualitative properties a good answer must exhibit, written in plain English as opposed to a golden example of an answer.

For AI testing the tests are termed evals, ie they evaluate the tool, but not by strict assertions, because one thing you can be sure of with an LLM is that given the same input, you usually get subtly different output, ie they are non-deterministic. Which means for LLM outputs the only way to test them is to use an LLM-as-judge,  ie give that LLM the test output and a rubric or golden and let it mark it against that. Then you set a pass threshold for that mark, to translate your complex output into a pass or fail.

You can also total up all the passes to give you a Task Completion Rate, TCR. So with complex AI agentic LLM interactions a 100% pass of all evals is often not realistic. Hence you set a TCR below 100% for the whole test suite of evals to pass. Start with the smallest useful test. The core principle of evals is not complicated, you want the input to give you the expected output.

But for an Agentic application this may require a sequence of LLM calls and tools: Making the final output dependent on the route that should be chosen, the tool(s) that should be called, the actions to be taken, further LLM calls that may be necessary and finally the core data that the response to the user should contain.

Our first version did not try to solve every part of that. It started with routing, simple and deterministic.

Routing is the starting point

The chatbot originally had an agent per tool. The tool being the code and API calls that performed actions or returned data or help.

Different specialist agents owned different parts of the product surface: Control-plane actions, Postgres database operations, schema design, roles and permissions, cluster reporting, migration, and so on.

Before any specialist can help, something has to choose the right specialist.

So the first eval suite asked a narrow question:

Given this user prompt, did the chatbot route to the expected tool?

That gave us a fast health check. We could keep a corpus of prompts, map each one to an expected destination, run them through either a direct model path or against the real deployment and its tools, and score whether the selected destination tool matched the golden.

 A golden here is just the name of the tool:

- id: "core-iam-001"
    prompt: "List all my projects"
    expected_tool: "control-plane"
    tags: ["core", "control-plane", "project"]

And the check on the other end is deliberately dumb — an equality test, not a semantic one:

self.success = tool_match(predicted_tool, expected_tool)

Agents became skills, but routing remained 

The design moved away from “one agent per tool family” toward a more consolidated orchestrating agent with skills.

That is a better fit for how modern agent systems are evolving. A skill = instructions, constraints, and a subset of tools that are relevant for a task. It is a form of progressive disclosure. Give the model the minium it needs at each step to save tokens.

But this did not make routing irrelevant.

Instead of asking “did we transfer to the right sub-agent?”, the eval asks “was the right skill made visible and selected for this task?” The labels changed but a skill could still use the wrong tool.

Routing evals stayed valuable because they were fast, explainable, and easy to run in CI. But they are limited, routing should always be correct but it doesn't mean that the final agent response is too.

TCR jumps to the endpoint, the response

Task Completion Rate, or TCR, was the next step.

The user asked for a cluster comparison, or a schema recommendation, or help diagnosing a database issue. We need to know whether the full response actually completed these tasks.

Responses are complex goldens so they need the LLM-as-a-judge pattern: run the chatbot, take the actual response, and ask a judge model to score it against expected sections.

The eval has a rubric here for judging the output:

- id: "tcr-core-014"
  prompt: "Compare CPU usage between these two clusters"
  expected_sections:
    - "identifies which cluster has higher CPU usage"
    - "cites at least one supporting metric"
    - "suggests a plausible next step"

The judge gets one simple instruction: score each expected_sections between 0.0–1.0 A metric class then just thresholds it for pass / fail:

self.success = score >= 0.7

The judge must be calibrated and a consistent model used for comparing runs over time. Enabling skill an prompt tuning from metric trends. The rubric must be specific enough to avoid marking waffle as success. But it turns a non-deterministic complex output into a simple pass and fail. It also separated two different levels of QA:

• Can the underlying model answer the task if given the right context?
• Does the deployed chatbot complete the task through the real product path?

That led to two execution modes.

Direct mode calls the model with simulated context. It is faster and useful for prompt and rubric development.

Proxy mode calls the real chatbot. It is slower, but it exercises the production path: routing, skill selection, tool calls, guardrails, streaming responses, conversation state, and the actual service wiring.

Both matter. Direct mode tells you whether the model is capable of the answer. Proxy mode tells you whether your product is capable of really delivering it via agents running your deployment's tools. 

This is the major difference from standard AI LLM testing, the model is only a small pluggable engine for the full agentic skill set that requires the actual deployment domain of data, actions and tools. Direct mode testing of only the model, is occasionally useful but E2E testing of the Chatbot deployment is required for agentic AI Chatbot QA, tuning and validation.   

Multi-step conversations changed the unit of testing

Single-turn TCR is still too small for many real chatbot tasks.

Users do not always provide all required information in one message. They ask to create a cluster, then pick a project, then choose a size, then confirm. They ask for a schema review, then refine the problem, then ask for a migration path. They troubleshoot by adding information over time.

So the framework has to exercise test cases that are conversations, not just single prompts.

That sounds like a minor data-model change. It was not. Once a test has steps, the eval runner has to preserve conversation state. In proxy mode, that means carrying the real conversation_id returned by the chatbot and sending each follow-up as part of the same server-side conversation. In direct mode, it means building a synthetic conversation history so the model sees the prior turns.

In code that split is about as literal as it sounds. Proxy mode threads a real id through each call:

response = client.send_message(prompt=msg, conversation_id=conversation_id)
conversation_id = response.conversation_id  # captured on turn 1, reused after

Direct mode has no server-side conversation to lean on, so it fakes one by re-rendering the transcript into the prompt itself, every turn:

full_prompt = f"## Conversation History\n{render(history)}\n\n{next_prompt}"

Same test case, same expected outcome, but a different code path depending on which half of the system is actually holding the conversation state. That's impacts multi-turn evals because conversation memory is part of the harness code for the actual deployment not just a model issue.

The scoring also becomes more interesting. You want per-step checks, because the assistant should ask the right clarifying question at the right time. You also want an overall score, because a conversation can have reasonable individual turns and still fail to complete the user's goal.

Coding it yourself: deepeval underneath

Everything above sits on top of deepeval, the open-source LLM eval library. We add a Synthesize → Execute → Evaluate pipeline, a plugin system, YAML goldens, CI wiring, and Langfuse push on top of it But the core library underneath is plain deepeval, and you do not need any of the surrounding machinery we used. Here are routing, TCR and multi-step just built directly on deepeval (simplified deepeval 3.6.9)

A test case is just an input/output pair. LLMTestCase is the base unit everything else scores:

from deepeval.test_case import LLMTestCase

test_case = LLMTestCase(
    input="List all my projects",
    actual_output=chatbot_response_text,       # what the system under test said
    expected_output="control-plane",                  # the golden - a skill label here, not prose
    additional_metadata={"predicted_skill": predicted_skill},
)

Routing is a custom metric, not a built-in one. deepeval ships plenty of semantic metrics, but “did it route to the right skill” is an exact-match business rule, so you write your own BaseMetric. This is a simplified version of the same shape our real AgentMatch metric takes:

from deepeval.metrics import BaseMetric
from deepeval.test_case import LLMTestCase

class AgentMatch(BaseMetric):
    def __init__(self, threshold: float = 1.0):
        self.threshold = threshold
        self.async_mode = False  # routing checks are cheap; no need for async here

    def measure(self, test_case: LLMTestCase) -> float:
        predicted = test_case.additional_metadata["predicted_skill"]
        expected = test_case.expected_output
        self.score = 1.0 if tool_match(predicted, expected) else 0.0
        self.success = self.score >= self.threshold
        return self.score

    async def a_measure(self, test_case: LLMTestCase) -> float:
        return self.measure(test_case)

    def is_successful(self) -> bool:
        return bool(self.success)

    @property
    def __name__(self):
        return "Agent Match"

tool_match is the check from earlier. Run it with deepeval's own runner rather than hand-rolled assertions, and you get retries, pretty output, and a result object for free:

from deepeval import evaluate

evaluate(test_cases=[test_case], metrics=[AgentMatch()])

TCR is where deepeval's built-in GEval earns its keep. GEval is deepeval's off-the-shelf LLM-as-judge metric, you give it criteria (or explicit evaluation steps) and it handles the judge prompt, the JSON parsing, and the scoring for you. Our rubric-per-line expected_sections maps onto evaluation_steps almost directly:

from deepeval.metrics import GEval
from deepeval.test_case import LLMTestCase, LLMTestCaseParams

task_completion = GEval(
    name="TaskCompletion",
    evaluation_steps=[
        "Check whether the response identifies which cluster has higher CPU usage",
        "Check whether the response cites at least one supporting metric",
        "Check whether the response suggests a plausible next step",
    ],
    evaluation_params=[LLMTestCaseParams.INPUT, LLMTestCaseParams.ACTUAL_OUTPUT],
    threshold=0.7,
)

test_case = LLMTestCase(
    input="Compare CPU usage between these two clusters",
    actual_output=chatbot_response_text,
)

evaluate(test_cases=[test_case], metrics=[task_completion])

Multi-step conversations get their own test case type. ConversationalTestCase takes a list of Turns instead of a single input/output pair, and pairs with a BaseConversationalMetric instead of BaseMetric:

from deepeval.test_case import ConversationalTestCase, Turn

convo = ConversationalTestCase(
    turns=[
        Turn(role="user", content="Create a new cluster"),
        Turn(role="assistant", content="Sure - which project should it go in?"),
        Turn(role="user", content="acme-prod"),
        Turn(role="assistant", content=final_response_text),
    ],
    expected_outcome="A cluster is created in acme-prod after resolving the missing project name",
)

deepeval has a conversational counterpart to GEval too (ConversationalGEval), scored against the whole turn sequence rather than a single response which is the natural fit for “did the assistant ask the right clarifying question at the right time”, the per-step-plus-overall shape TCR needed once prompts became conversations.

Put together, that is the whole starting kit: LLMTestCase plus a hand-written BaseMetric for hard business rules like routing, GEval for rubric-style task completion, ConversationalTestCase plus ConversationalGEval once a prompt becomes a conversation, and evaluate() to run the lot and get a result object back.

Everything else we built, the YAML goldens, the plugin architecture, the CI wiring, the Langfuse push exists to run more of these at scale and make the failures easy to find. But none of it is required to get started. If you are testing your own agentic chatbot, this is how to begin.

This is where instrumentation started to matter much more.

For a single-turn answer, a markdown report with pass/fail rows is often enough to start debugging. For multi-step conversations, that is thin. You need to know which turn failed, whether the route changed, whether the wrong tool was called, whether the tool call used correct arguments, whether the model forgot earlier context, or whether the final answer simply missed a required section.

That is why we added span-level telemetry and pushed eval traces into Langfuse.

Langfuse made the failures inspectable

The useful thing about Langfuse is not just having another pretty dashboard. Although that is important for spotting quality regressions over time via regular CI/CD automated runs.

The vital thing was being able to treat an eval run as a set of traces. A run becomes a session. Each test case becomes a trace. The trace carries the prompt, response, scores, tags, model, mode, scenario, and the spans emitted by the proxy.

For a chatbot path, those spans are where the debugging starts. You can see routing, tool execution, LLM calls, latency, and token usage where it is available. You can filter by scenario and model. You can compare runs. You can look at a failing conversation and see whether the problem began at route selection, tool selection, tool arguments, or final synthesis.

That changes the tuning loop.

Without traces, an eval failure says “this case failed”. With traces, it can say why it failed.

That distinction matters because the fix lands in different places...
Is it a routing rule?
Is it a skill description?
Is it a tool schema?
Is it the judge rubric?
Is it that the eval has has an expectation that the product has never actually promised?

 Trajectory testing ->  knitted the pieces together

Routing and TCR started as separate signals.

Routing asked whether the right capability was selected. TCR asked whether the final task was completed. Multi-step testing asked whether that held across a conversation. Instrumentation showed what happened between those points.

Trajectory testing is the next natural step: score the path itself.

For an agentic product, the fully correct path is essential to response quality.
So trajectory tests add expectations about intermediate actions:

• which tool or flow should be used
• whether the arguments are valid
• whether the conversation reached the right state
• whether the final answer completed the task

The label-based routing tests are still useful as fast canaries. They tell us whether the classifier shape has drifted and distinguish tiers - see the next section.
But full trajectory tests judge the route by consequence: did the system actually follow the tool path that would satisfy the user?

So retain the fast determisitc routing tests, but move more user-visible behavioural coverage into trajectory and TCR.

Sovereign AI makes the eval problem tiered S/M/L/XL

There is one more constraint that makes this more than a generic chatbot-testing story.

Our chatbot has to work for sovereign and air-gapped deployments. In those environments, prompts, tool results, schema details, and operational data cannot be sent to a hosted frontier model outside the customer's trust boundary. The inference model may run inside the customer's environment.

That usually means a smaller model.

Smaller models are not just cheaper versions of larger ones. They have different context limits, weaker tool-selection behaviour, and less tolerance for an over-wide capability surface. If you show a smaller model every possible tool and skill, you have increased the chance that it chooses a bad one.

So the architecture becomes tiered. Models are effectively T-shirt sized. A small self-hosted model sees a curated subset of reliable skills. A larger model can be allowed to see more. Some experimental or complex skills only make sense for the highest tiers.

That changes the meaning of a routing eval again.

The correct visible skill set is no longer universal. It depends on the model tier. A prompt that should route to an advanced skill for an XL model may need to be dropped, refused, or handled differently for a smaller model that should not see that skill at all.

This is why trajectory testing and routing need to be tier-aware. We are not only asking whether the chatbot can complete a task. We are asking whether it can complete the task through the capability surface that a deployment's LLM size allows.

What I would keep from the journey

The final shape was not obvious at the start.

We began with routing because it was the first integration failure point and the cheapest one to isolate. We added TCR because correct routing did not prove task completion. We added multi-step cases because real users have conversations, not isolated prompts. We added telemetry because multi-step failures are otherwise too hard to debug. We moved toward trajectory testing because the route, tools, arguments, and answer need to be judged as one path.

If I were starting another agentic product eval framework, I would keep that order.

Do not start by trying to build a grand universal benchmark. Start with the smallest failure point that would embarrass the product if it regressed. Then move the signal closer to the user's actual goal.

For a chatbot wired into a real control plane, that means testing more than the output text. It means testing the route, the skill, the tool call, the arguments, the conversation state, the final answer, and the model tier that made those options visible in the first place.

That is the difference between checking that an AI system said something vaguely relevant and checking that it actually did all the things the user asked of it.

Talk about Cloud Prices at PyConLT 2025

Introduction to Cloud Pricing

I am looking forward to speaking at PyConLT 2025. 
My talk is called Cutting the Price of Scraping Cloud Costs (video)

Its been a while (12 years!) since my last Python conference EuroPython Florence 2012, when I spoke as a Django web developer, although I did give a Golang talk at Kubecon USA last year.

I work at EDB, the Postgres company, on our Postgres AI product. The cloud version of which runs across the main cloud providers, AWS, Azure and GCP.

The team I am in handles the identity management and billing components of the product. So whilst I am mainly a Golang micro-service developer, I have dipped my toe into Data Science, having rewritten our Cloud prices ETL using Python & Airflow. The subject of my talk in Lithuania.

Cloud pricing can be surprisingly complex ... and the price lists are not small.

The full price lists for the 3 CSPs together are almost 5 million prices - known as SKUs (Stock Keeping Unit prices)

csp x service x type x tier x region
3    x  200      x 50     x 3     x 50        = 4.5 million

csp = AWS, Azure and GCP

service = vms, k8s, network, load balancer, storage etc.

type = e.g. storage - general purpose E2, N1 ... accelerated A1, A2  multiplied by various property sizes

tier  = T-shirt size tiers of usage, ie more use = cheaper rate - small, medium, large

region = us-east-1, us-west-2, af-south-1, etc.

We need to gather all the latest service SKU that our Postgres AI may use and total them up as a cost estimate for when customers are selecting the various options for creating or adding to their installation.
Applying the additional pricing for our product and any private offer discounts for it, as part of this process.

Therefore we needed to build a data pipeline to gather the SKUs and keep them current.

Previously we used a 3rd party kubecost based provider's data, however our usage was not sufficient to justify for paying for this particular cloud service when its free usage expired.

Hence we needed to rewrite our cloud pricing data pipeline. This pipeline is in Apache Airflow but it could equally be in Dagster or any other data pipeline framework.

My talk deals with the wider points around cloud pricing, refactoring a data pipeline and pipeline framework options. But here I want to provide more detail on the data pipeline's Python code, its use of Embedded Postgres and Click, and the benefits for development and testing.  Some things I didn't have room for in the talk.

Outline of our use of Data Pipelines

Airflow, Dagster, etc. provide many tools for pipeline development.
Notably local development mode for running up the pipeline framework locally and doing test runs.
Including some reloading on edit, it can still be a long process, running up a pipeline and then executing the full set of steps, known as a directed acyclic graph, DAG.

One way to improve the DEVX is if the DAG step's code is encapsulated as much as possible per step.
Removing use of shared state where that is viable and allowing individual steps to be separately tested, rapidly, with fixture data. With fast stand up and tear down, of temporary embedded storage.

To avoid shared state persistence across the whole pipeline we use extract transform load (ETL) within each step, rather than across the whole pipeline. This enables functional running and testing of individual steps outside the pipeline.

The Scraper Class

We need a standard scraper class to fetch the cloud prices from each CSP so use an abstract base class.

from abc import ABC

class BaseScraper(ABC):

   """Abstract base class for Scrapers"""

   batch = 500

   conn = None

   unit_map = {"FAIL": ""}

   root_url = ""

   def map_units(self, entry, key):

       """To standardize naming of units between CSPs"""

       return self.unit_map.get(entry.get(key, "FAIL"), entry[key])

   def scrape_sku(self):

       """Scrapes prices from CSP bulk JSON API - uses CSP specific methods"""

       Pass

   def bulk_insert_rows(self, rows):

       """Bulk insert batches of rows - Note that Psycopg >= 3.1 uses pipeline mode"""

       query = """INSERT INTO api_price.infra_price VALUES

       (%(sku_id)s, %(cloud_provider)s, %(region)s, … %(sku_name)s, %(end_usage_amount)s)"""

       with self.conn.cursor() as cur:

           cur.executemany(query, rows)

This has 3 common methods:

1. mapping units to common ones across all CSP
2. Top level scrape sku methods some CSP differences within sub methods called from it
3. Bulk insert rows - the main concrete method used by all scrapers

To bulk insert 500 rows per query we use Psycopg 3 pipeline mode - so it can send batch updates again and again without waiting for response.

The database update against local embedded Postgres is faster than the time to scrape the remote web site SKUs.

The largest part of the Extract is done at this point. Rather than loading all 5 million SKU as we did with the kubecost data dump, to query out the 120 thousand for our product. Scraping the sources directly we only need to ingest those 120k SKU. Which saves handling 97.6% of the data!

So the resultant speed is sufficient although not as performant as pg_dump loading which uses COPY.

Unfortunately Python Psycopg is significantly slower when using cursor.copy and it mitigated against using zipped up Postgres dumps. Hence all the data artefact creation and loading simply uses the pg_dump utility wrapped as a Python shell command. 

There is no need to use Python here when there is the tried and tested C based pg_dump utility for it that ensures compatibility outside our pipeline. Later version pg_dump can always handle earlier Postgres dumps.

We don't need to retain a long history of artefacts, since it is public data and never needs to be reverted.

This allows us a low retention level, cleaning out most of the old dumps on creation of a new one. So any storage saving on compression is negligible.

Therefore we avoid pg_dump compression, since it can be significantly slower, especially if the data already contains compressed blobs. Plain SQL COPY also allows for data inspection if required - eg grep for a SKU, when debugging why a price may be missing.

Postgres Embedded wrapped with Go

Unlike MySQL, Postgres doesn't do in memory databases. The equivalent for temporary or test run database lifetime, is the embedded version of Postgres. Run from an auto-created temp folder of files. 
Python doesn’t have maintained wrapper for Embedded Postgres, sadly project https://github.com/Simulmedia/pyembedpg is abandoned 😢

Hence use the most up to date wrapper from Go. Running the Go binary via a Python shell command.

It still lags behind by a version of Postgres, so its on Postgres 16 rather than latest 17.

But for the purposes of embedded use that is irrelevant.

By using separate temporary Postgres per step we can save a dumped SQL artefact at the end of a step and need no data dependency between steps, meaning individual step retry in parallel, just works.
The performance of localhost dump to socket is also superior.

By processing everything in the same (if embedded) version of our final target database as the Cloud Price, Go micro-service, we remove any SQL compatibility issues and ensure full Postgresql functionality is available.

The final data artefacts will be loaded to a Postgres cluster price schema micro-service running on CloudNativePG

Use a Click wrapper with Tests

The click package provides all the functionality for our pipeline..

> pscraper -h

Usage: pscraper [OPTIONS] COMMAND [ARGS]...

   price-scraper: python web scraping of CSP prices for api-price

Options:

  -h, --help  Show this message and exit.

Commands:

  awsscrape     Scrape prices from AWS

  azurescrape  Scrape prices from Azure

  delold            Delete old blob storage files, default all over 12 weeks old are deleted

  gcpscrape     Scrape prices from GCP - set env GCP_BILLING_KEY

  pgdump        Dump postgres file and upload to cloud storage - set env STORAGE_KEY
                      > pscraper pgdump --port 5377 --file price.sql 

  pgembed      Run up local embeddedPG on a random port for tests

> pscraper pgembed

  pgload           Load schema to local embedded postgres for testing

> pscraper pgload --port 5377 --file price.sql

This caters for developing the step code entirely outside the pipeline for development and debug.
We can run pgembed to create a local db, pgload to add the price schema. Then run individual scrapes from a pipenv pip install -e version of the the price scraper package.

For unit testing we can create a mock response object for the data scrapers that returns different fixture payloads based on the query and monkeypatch it in. This allows us to functionally test the whole scrape and data artefact creation ETL cycle as unit functional tests.

Any issues with source data changes can be replicated via a fixture for regression tests.

class MockResponse:

"""Fake to return fixture value of requests.get() for testing scrape parsing"""

name = "Mock User"

payload = {}

content = ""

status_code = 200

url = "http://mock_url"

def __init__(self, payload={}, url="http://mock_url"):

self.url = url

self.payload = payload

self.content = str(payload)

def json(self):

return self.payload

def mock_aws_get(url, **kwargs):

    """Return the fixture JSON that matches the URL used"""

for key, fix in fixtures.items():

if key in url:

return MockResponse(payload=fix, url=url)

return MockResponse()

class TestAWSScrape(TestCase):

"""Tests for the 'pscraper awsscrape' command"""

def setUpClass():

"""Simple monkeypatch in mock handlers for all tests in the class"""

psycopg.connect = MockConn

requests.get = mock_aws_get

# confirm that requests is patched hence returns short fixture of JSON from the AWS URLs

result = requests.get("{}/AmazonS3/current/index.json".format(ROOT))

assert len(result.json().keys()) > 5 and len(result.content) < 2000

A simple DAG with Soda Data validation

The click commands for each DAG are imported at the top, one for the scrape and one for postgres embedded, the DAG just becomes a wrapper to run them, adding Soda data validation of the scraped data ...

def scrape_azure():
   """Scrape Azure via API public json web pages"""
   from price_scraper.commands import azurescrape, pgembed
   folder, port = setup_pg_db(PORT)
   error = azurescrape.run_azure_scrape(port, HOST)
   if not error:
       error = csp_dump(port, "azure")
   if error:
       pgembed.teardown_pg_embed(folder) 
       notify_slack("azure", error)
       raise AirflowFailException(error)
  
   data_test = SodaScanOperator(
       dag=dag,
       task_id="data_test",
       data_sources=[
           {
               "data_source_name": "embedpg",
               "soda_config_path": "price-scraper/soda/configuration_azure.yml",
           }
       ],
       soda_cl_path="price-scraper/soda/price_azure_checks.yml",
   )
   data_test.execute(dict())
   pgembed.teardown_pg_embed(folder) 

We setup a new Embedded Postgres (takes a few seconds) and then scrape directly to it.

We then use the SodaScanOperator to check the data we have scraped, if there is no error we dump to blob storage otherwise notify Slack with the error and raise it ending the DAG

Our Soda tests check that the number of and prices are in the ranges that they should be for each service. We also check we have the amount of tiered rates that we expect. We expect over 10 starting usage rates and over 3000 specific tiered prices.

If the Soda tests pass, we dump to cloud storage and teardown temporary Postgres. A final step aggregates together each steps data. We save the money and maintenance of running a persistent database cluster in the cloud for our pipeline.