Reverse ETL to Attio

Reverse ETL is generally about putting data into a business application. The data can come from apps or products a company uses. More often this data is stored in SQL databases, serving as a middle layer and then reverse ETL to the application being used. Many applications in the market allow you to do this using no code solutions. And for non-tech teams that’s an easier way to move forward. But it comes with limitations - and one of the major ones is - all logic lives inside the black box.

dlt skips that. It gives you more control over the source, and at the SQL database middle layer that you land data to and how it's loaded into the destination. In our earlier post we showed how to build a custom pipeline with Google Sheets as destination using the @dlt.destination decorator. This post takes it further: dltHub's AI Harness agent builds all three stages — ingestion via the REST API toolkit, the enrichment layer, and the reverse-ETL stage via @dlt.destination.

An example case study for GTM workflow

To pitch a product we often need to find people who can invest in the product, for example a company posts jobs for a “Customer Success Manager”, signals they are expanding their CSM team and capabilities. Building on that I created a simple workflow for a GTM team where

• Load job postings for a particular role from a job portal to the database
• Enrich those job postings with company and people data using People Data Labs API
• Load the data to the CRM of choice - Attio in this case

Here’s the workflow step-by-step:

Step 1 - Created a REST API pipeline from job board (Arbeitnow) to DuckDB.

I used dltHub AI Harness’s REST API toolkit - to load the filtered “customer success manager” postings from Arbeitnow’s job board API to DuckDB.

@dlt.source(name="arbeitnow")
def arbeitnow_source() -> Any:
    config: RESTAPIConfig = {
        "client": {
            "base_url": "<https://arbeitnow.com/api/>",
            "paginator": {
                "type": "json_link",
                "next_url_path": "links.next",
            },
        },
        "resources": [
            {
                "name": "jobs_raw",
                "endpoint": {
                    "path": "job-board-api",
                    "data_selector": "data",
                },
                "primary_key": "slug",
                "write_disposition": "merge",
                "processing_steps": [
                    {"filter": lambda item: "customer success manager" in item["title"].lower()},
                ],
            },
        ],
    }

    (jobs_raw,) = rest_api_resources(config)

    @dlt.resource(name="jobs", primary_key="slug", write_disposition="merge")
    def jobs(created_at: dlt.sources.incremental[int] = dlt.sources.incremental(
        "created_at", initial_value=0
    )):
        yield from jobs_raw

    yield jobs

Step 2 - Enriched the job entries using People Data Labs

This turn the “company is hiring” signal to “here’s the company, and here are the people to talk to” who can make the decision to buy the product. I had the agent add two People Data Labs enrichments:

• Company Enrichment (data about the company from job postings)
• People enrichment (people who can own a buying decision e.g. VP Sales, Head of Customer Success, RevOps…)

Why this stayed cost-aware:

When I flagged the enrichment-credit burn, the agent added a check that reads the pipeline's own previous output from DuckDB before calling PDL.

def _already_enriched_domains() -> set[str]:
    rows = dlt.attach(pipeline_name="pdl_enrichment").dataset()["companies"].select("domain").fetchall()
    return {row[0] for row in rows}

known_domains = _already_enriched_domains()
if company["domain"] in known_domains:
    continue  # already have this one — skip the PDL call

Six lines. Not a feature to configure, not a setting buried in a UI, just the next line of code in the same pipeline that made the call in the first place.

Step 3 - Wrote a simple reverse ETL pipeline to Attio

This step writes everything so far (the job signal, the company enrichment, the decision-makers found in above steps) from DuckDB into Attio, the CRM the sales team works out of, so a rep can open it up and see "here's a company that's expanding, and here are the people there worth reaching out to.”

The agent found Attio's docs, and it worked out upsert support, so that a weekly re-sync doesn't create duplicate companies or people. Companies turned out simple: domains is a unique field on Attio's side, so a single PUT with matching_attribute=domains in the query string, the param that tells Attio's endpoint which field to match an existing record on, it updates if it finds one, creates if it doesn't. The people endpoint has no unique field, so it recommended a PATCH-or-POST fallback: upsert_person queries by LinkedIn URL first, PATCHes if a record comes back, POSTs a new one if not. It also sets the person's company field, a reference to the company record, matched by domain.

@dlt.destination(batch_size=1, name="attio")
def attio_destination(items, table, api_key: str = dlt.secrets.value):
    headers = {"Authorization": f"Bearer {api_key}"}
    for item in items:
        if table["name"] == "companies":
            upsert_company(headers, item)
        elif table["name"] == "people":
            upsert_person(headers, item)

And then loading the data to the Attio destination.

This is how it looks deployed in the dltHub workspace:

Company’s record:

Person’s record:

Note: the pipeline runs on real records. The company and person shown here are fictional, for privacy. Only the values are swapped; the records are otherwise exactly as the pipeline writes them.

Why this matters

Reverse ETL tools generally offer a catalogue of supported destinations you pick from. If your destination isn’t in that catalogue or the semantics differ, your workflow is stuck. This is where @dlt.destination lets you write your own custom destinations. As you saw in the above example, companies upsert cleanly on domains and people don’t, so this needed a PATCH-or-POST fallback, and this is what customization allows you.

The second advantage: source, warehouse, and reverse ETL live in one codebase. That gives you versioned history, and you can add sources or swap destinations later without switching tools. That just makes it easier to grow your reverse ETL pipeline.

Extending this pipeline

You can add a step here: create a canonical data model for the job, company, and people data before writing anything to the reverse-ETL destination. This can be done using dltHub's AI Harness transformation toolkit.

Since this pipeline has one source, one destination, and nobody else reading the same data, I skipped this step to keep it simple. That changes once a second source starts feeding Attio too, or once something besides Attio needs the same company and people data - that's when you want one shared definition of "company" or "person" instead of writing that logic into every destination separately. And that's where a CDM helps.

Try it yourself

To build a workflow like the one above, install dltHub AI Harness and start with this prompt.

Build a GTM reverse-ETL pipeline with three stages:

1. Ingest: use the REST API toolkit to pull job postings from Arbeitnow's public job board API, filtered client-side to postings whose title contains "customer success manager". It's a public API, no key needed. Load into DuckDB, and figure out which field can drive incremental loading.

2. Enrich: for each unique hiring company, use People Data Labs to look up the company, then find a few people there in sales or customer-success leadership roles — the people who'd actually own a buying decision. Skip companies you've already enriched in a previous run, so you're not spending enrichment credits twice on the same signal.

3. Reverse ETL: write the companies and people into Attio as CRM records. Companies should upsert cleanly by domain. People don't have a unique field to match on, so look them up by LinkedIn URL and update or create accordingly, and link each person to their company so they show up nested under it in Attio. Attio doesn't support bulk writes, so process one record at a time.

A few yeses and some back-and-forth with the agent will get you there in no time.

More on how to create your own custom destination for reverse ETL: https://dlthub.com/docs/dlt-ecosystem/destinations/destination

10+ pipelines to reverse ETL into one or more destinations? Or simply need an extra set of hands? Let us help!