Skip to content

file-reference.md

Latest commit

 

History

History
1014 lines (795 loc) · 44.5 KB

file-reference.md

File metadata and controls

1014 lines (795 loc) · 44.5 KB
Raw

FileReference & App.upload() — Complete Developer Reference

This document is the single source of truth for large-payload data transfer in application-sdk v3. Read it before writing any code that passes files between tasks, writes run artifacts, or pushes data upstream.

Why This Exists

Temporal serialises every Input and Output through its payload codec. The hard limit is 2 MB per payload. A single SQLite database, Parquet file, or JSON manifest from a real connector run is almost always larger than that. Embedding raw bytes in your model will fail at runtime with a cryptic Temporal serialisation error.

FileReference solves this by storing the actual bytes in object storage (S3, GCS, Azure Blob) and putting only a tiny pointer in the Temporal payload. The SDK uploads and downloads automatically — task authors never call storage APIs directly.

App.upload() solves a different problem: pushing a file to a stable, addressable path that survives beyond the current run and is visible to external systems (the Atlan platform, downstream pipelines, SDR).

These are two distinct APIs with distinct purposes. Using the wrong one causes subtle failures.

Decision Matrix — Which API to Use

Scenario API Why
Pass a SQLite / Parquet / JSON file from task A to task B in the same run FileReference field on Input/Output Auto persisted on output, auto materialised on input. Zero storage code needed.
Task B always needs what task A produced, and the file is always present FileReference field (eager, default) SDK downloads it before task B runs.
Task B only sometimes needs the file (e.g. only when mode == "full") Annotated[FileReference | None, Lazy()] SDK skips the download; task B fetches on demand with fetch().
Multiple tasks all share the same large manifest file Single FileReference passed through the chain SDK downloads it exactly once per task via dedup.
Push a completed artifact so it appears in the Atlan UI after the run await self.upload(UploadInput(local_path=...)) RETAINED is the default tier — writes to a stable run-scoped prefix that the platform indexes.
Write a file that must persist across multiple runs (e.g. incremental bookmark) await self.upload(UploadInput(..., tier=StorageTier.PERSISTENT)) Writes to a fixed path not cleaned up between runs.
Directory of output files (e.g. partitioned Parquet) from task A to task B FileReference(local_path=str(dir_path)) — same API, directory-aware SDK uploads all files in the directory and re-creates the structure on download.
Assert that everything a step wrote is really in the store before handing a prefix downstream await self.verify_refs(VerifyRefsInput(refs=..., prefix=...)) The refs are the producer's declaration; a prefix walk cannot tell "absent" from "lost".
Land a fanned-out step's output as one outbound tree for publish / QI / lineage await self.upload_refs(UploadRefsInput(files=..., prefix=...)) Uploads each declared ref by reference, so it works when the producing activities ran on other pods.
Pass intermediate data between tasks using App.upload() Don't — use FileReference App.upload() routes to Atlan's atlan-objectstore in SDR, polluting it with internal artifacts; bypasses SHA-256 dedup; and doesn't wire into cross-worker auto-materialization.

Key rule: FileReference is for within-run transfer. App.upload() is for durable outbound delivery.

Anti-pattern: App.upload() for task-to-task data. Using App.upload() to move data between tasks — instead of FileReference — causes three distinct harms:

  1. Wrong bucket. In SDR, App.upload() routes to the Atlan-owned atlan-objectstore. Intermediate pipeline data (raw SQL rows, partial transforms) pollutes Atlan's bucket with artifacts that the publish app treats as connector output.
  2. Re-uploads by default. App.upload() runs on the same upload engine as the interceptor's persist, so it writes the same sidecars, local markers and directory manifest. But it is a separate @task call that uploads every file again unless you pass skip_if_exists=True, and its destination is a durable outbound prefix, not a run-scoped hand-off.
  3. No cross-worker auto-materialization. FileReference input fields are automatically re-downloaded before a task runs when the local file is absent (cross-worker fault tolerance). App.upload() produces a bare FileReference without wiring it into the SDK's materialization machinery.

For task-to-task data: declare FileReference on your Output and Input contracts — the interceptor handles persistence and materialization automatically.

Which store does each API write to?

These two APIs route to different object stores:

API Store Who can read it
FileReference (via interceptor) objectstore — infra.storage, customer-owned Other tasks in the same deployment
App.upload() atlan-objectstore — infra.upstream_storage, Atlan-owned (in SDR); falls back to infra.storage in local dev Atlan system apps (publish, QI, lineage) and the Atlan UI

The activity interceptor's persist step always writes FileReference objects to infra.storage (objectstore). This is intentional: intermediate task outputs stay inside the customer's deployment perimeter and never cross into Atlan's infrastructure unless the connector explicitly decides to hand them off.

App.upload() uses upstream_storage or storage. In SDR deployments upstream_storage is the atlan-objectstore Dapr binding (pointing to {tenant}/api/blobstorage), so App.upload() routes to Atlan's S3.

App.download() reads from the same place. A ref that App.upload() or App.download() returns from a distinct upstream store carries store=StoreTarget.UPSTREAM. When you pass it into another @task, the interceptor materialises it from the upstream store, not the deployment store. Without that field the next task would list an empty prefix in the wrong bucket and fail with AAF-STR-009. Every other ref, including those from a single-store deployment where both names alias one store, has store=DEPLOYMENT. Do not set the field by hand. It records where the bytes are, and the SDK sets it. App.download() and the interceptor share one transfer engine. Passing dl.ref to a task on the same worker therefore costs one listing and a local re-hash, not a second download, because download() leaves the same local .sha256 markers that materialise does. App.upload() and the interceptor's persist also share one upload engine. A directory up.ref carries a manifest_path and leaves a local .sha256 marker beside each file, so a task on the same worker reuses the files instead of downloading them back, and never reads a stray object under the prefix as part of the hand-off. Both refuse the same sensitive local paths: a protected system directory itself or anything under it (/etc, /proc, /sys, /dev, /root), and any path through a credential directory (.ssh, .aws, .gnupg, .kube, .vault), an .env* name, or a match for ATLAN_UPLOAD_FILE_BLOCKED_PATHS. For a directory that applies to every file in it: one .env inside an output directory refuses the whole upload, naming the file, rather than shipping it or dropping it silently. The tag is honoured only for refs produced inside the run. A workflow's start input (including continue_with) and every @update argument have any store=UPSTREAM reset to DEPLOYMENT, so to consume another app's upstream output, call App.download() on it.

Consequence for connector authors: if your connector produces artifacts that Atlan's publish app must consume, you must call App.upload() explicitly from run() — the interceptor alone is not sufficient. Omitting the explicit call produces a silent failure: the DAG runs to completion but the publish app finds nothing in Atlan's S3 and publishes zero assets. See ADR-0014 for the full rationale.

Part 1 — FileReference

What Is a FileReference?

A frozen Pydantic model that holds a pointer to a file (or directory) in object storage plus an optional local path on the current worker:

class FileReference(BaseModel, frozen=True):
    local_path: str | None = None      # absolute path on the worker's disk
    storage_path: str | None = None    # object-store key (set after persist)
    is_durable: bool = False           # True once uploaded to the store
    tier: StorageTier = StorageTier.TRANSIENT
    file_count: int = 1                # number of files; >1 for directory uploads
    auto_materialize: bool = True      # False = SDK never touches this ref
    manifest_path: str | None = None   # directory refs: key of the exact-file-set manifest

You never set is_durable, file_count, or manifest_path yourself — the SDK fills them in during persist and materialize. storage_path is optional: leave it unset and the SDK generates a unique file_refs/{uuid} key, or set it to pin a canonical key (single file) or prefix (directory).

On an entry point's contract, a FileReference must declare its shape. An entry point's input/return contracts are a public boundary — another app or the platform DAG reads what you write there — so every FileReference field on one needs an artifactSchemas entry in your pkl contract. Omitting it warns at worker build today and is an error in v4.0. A FileReference on an internal @task contract is exempt. See Apps — Declaring artifact schemas.

Declared or not, every FileReference the interceptor sees emits one artifact-validation outcome on each side of the task — at ingest after materialize and at hand-off before persist. A declared artifact is scanned against its declaration; an undeclared one reports not_declared rather than nothing at all. Report-only unless the app opts into artifact_validation_mode = "hard".

Constructing a FileReference

Constructing a FileReference is just a Pydantic model instantiation — it works anywhere (a @task, run(), a helper function). What requires a @task-decorated method is persisting the ref so the SDK can auto-upload it on activity return:

# Single file
ref = FileReference(local_path="/tmp/output/tables.parquet")

# Single file, non-default tier
ref = FileReference(
    local_path="/tmp/output/tables.parquet",
    tier=StorageTier.RETAINED,
)

# Directory (SDK walks all files recursively)
ref = FileReference(local_path="/tmp/output/")

# Opt-out of automatic lifecycle (advanced — you own download/upload)
ref = FileReference(local_path="/tmp/special.bin", auto_materialize=False)

The SDK only auto-persists a FileReference when it is returned from a @task method as part of an Output model. A ref constructed in run() and never returned from a task will not be uploaded — run() is a Temporal workflow function with sandbox restrictions, and the upload happens activity-side. As a rule of thumb: build refs inside @task methods and return them; treat refs received in run() as already-durable handles to pass between tasks.

Tiers

The tier controls where the file is stored and when it is cleaned up.

StorageTier.TRANSIENT (default)

Prefix:  file_refs/{uuid}
Cleanup: deleted by App.cleanup_storage() at the end of every run
Use for: intermediate files between tasks that are no longer needed after
         the run completes

Example — extraction output that the transform task needs but nothing else:

class ExtractOutput(Output):
    raw_db: FileReference | None = None

# In the extract task:
return ExtractOutput(
    raw_db=FileReference(local_path=str(sqlite_path))
    # tier=StorageTier.TRANSIENT is the default
)

StorageTier.RETAINED

Prefix:  artifacts/apps/{app_name}/workflows/{workflow_id}/{run_id}/file_refs/
Cleanup: NOT deleted by standard cleanup — must be removed manually or via
         a retention policy
Use for: run artifacts you want available after the run completes
         (downloadable from the Atlan UI, inspectable for debugging)

Example — a processed Parquet file that needs to be accessible after the run:

class TransformOutput(Output):
    result_parquet: FileReference | None = None

return TransformOutput(
    result_parquet=FileReference(
        local_path=str(output_parquet),
        tier=StorageTier.RETAINED,
    )
)

StorageTier.PERSISTENT

Prefix:  persistent-artifacts/apps/{app_name}/file_refs/
Cleanup: never deleted by any standard SDK operation
Use for: files that must survive across multiple runs — incremental
         watermarks, shared lookup tables, cross-run caches

Example — an incremental bookmark that the next run will read:

class BookmarkOutput(Output):
    watermark_file: FileReference | None = None

return BookmarkOutput(
    watermark_file=FileReference(
        local_path=str(watermark_path),
        tier=StorageTier.PERSISTENT,
    )
)

Lifecycle in Detail

Persist (output side)

After your @task returns an Output containing an ephemeral FileReference (one with local_path set but is_durable=False), the SDK automatically uploads it to infra.storage (objectstore — the deployment store):

  1. Computes the storage path from the tier prefix.
  2. Uploads the local file (or all files in a local directory) to the store.
  3. Writes a {storage_path}.sha256 sidecar to the store — used for integrity verification on download.
  4. Writes a {local_path}.sha256 marker locally — the digest plus the file's (size, mtime_ns, inode). A later task on the same worker reuses the file without downloading it, and without reading it while that stat still matches.
  5. Returns a durable ref (is_durable=True, storage_path set, and size_bytes for a single file) in the serialised Output that Temporal stores.

The original local file is not deleted unless you pass retain_local_copy=False to App.upload() (that is a different code path).

Materialize (input side)

Before your @task runs, the SDK inspects every FileReference field on the Input. For each durable ref that needs downloading (file absent, marker missing, size different from size_bytes, or file hash different from the marker), the SDK:

  1. Downloads the file from the store. A ref with no local_path (the usual hand-off shape: only a storage_path) gets a fresh temp name under the run's scratch directory, {TEMPORARY_PATH}/artifacts/apps/{app}/workflows/{workflow_id}/{run_id}/, which cleanup_files() removes when the run completes. Directory refs get a fresh subdirectory there. Outside an activity there is no run to scope to, so the destination is {TEMPORARY_PATH}/file-refs/. Either way it is never the system temp directory.
  2. Verifies the downloaded content SHA-256 against the stored sidecar.
  3. Writes a fresh local sidecar.
  4. Sets local_path on the ref so your task sees a ready-to-use local file.

Retry safety: if the task retries on a different worker pod, local_path will be a stale path from another machine. The sidecar check catches this: the local sidecar is absent (different pod, different disk), so the SDK re-downloads.

Cache hit: if the same worker hands the ref on, or retries, and the local file is intact, the SDK skips the download entirely. A file whose size, mtime and inode still match its marker is trusted without being read, so even a multi-GB hand-off costs only a stat. If the stat has moved, or the marker was written by an older SDK that recorded only the digest, the file is re-hashed in a worker thread (never on the event loop): a match re-stamps the marker, a mismatch re-downloads. A replacement that restores the original size and mtime on the same inode (cp -p over the file in place) is not caught.

The local marker is SDK bookkeeping. Do not parse it in app code or tests: its format is not a contract.

Directory FileReferences

Passing a directory works exactly like a single file from the task author's perspective:

# Produce: task returns a directory ref
output_dir = Path("/tmp/output/partitioned/")
return ExtractOutput(result_dir=FileReference(local_path=str(output_dir)))

# Consume: task receives a materialized directory ref
async def transform(self, input: TransformInput) -> TransformOutput:
    dir_path = Path(input.result_dir.local_path)
    for parquet_file in dir_path.glob("*.parquet"):
        process(parquet_file)

The SDK uploads every file in the directory recursively (bounded concurrency), preserves the directory structure under the storage prefix, and recreates it on download. file_count on the durable ref tells you how many files were uploaded.

The manifest. After the last file lands, persist writes a manifest — every file's relative path, SHA-256 and size — and puts its key on the durable ref as manifest_path. The manifest lives under the tier's file_refs/ namespace, never inside the data prefix. Materialize downloads exactly the manifest's files and checks each against the manifest digest; it does not list the prefix. So:

  • objects under the prefix that the manifest does not name — a failed earlier attempt's leftovers, a timed-out attempt still writing — are ignored;
  • a listed file overwritten after the manifest was written fails with StorageIntegrityError instead of being processed;
  • a missing manifest fails retryably rather than falling back to a listing.

Pinning a directory prefix. Because consumers read the manifest, a directory ref may pin storage_path to a canonical prefix exactly as a single-file ref pins its key. For TRANSIENT and RETAINED refs the pin must sit under the run prefix (output_path), or persist raises StorageConfigError: two runs pinning one shared prefix would write the same keys, and either run's cleanup would delete the other's data. PERSISTENT pins may live anywhere under persistent-artifacts/:

return ExtractOutput(
    tables=FileReference(
        local_path=str(output_dir),
        storage_path=f"{output_path}/raw/table/",  # honoured; trailing / optional
    )
)

A TRANSIENT pin is still deleted at end of run, manifest included.

Refs without a manifest — produced by an SDK that predates manifests, or built by hand — are materialized by listing the prefix. The listing is checked against file_count as a lower bound: fewer objects than declared fails retryably (StorageError, code STR-009), including an empty listing for a ref that declares any files — the default file_count=1 does. More objects than declared only logs file_ref.materialize.count_mismatch at WARNING. A hand-built ref over a prefix that can legitimately be empty must declare file_count=0.

The per-file sidecar fast-path applies inside directories too: on retry, files whose local copy still matches are skipped (against the manifest digest when there is one, else the local sidecar); only changed or missing files are re-downloaded.

Lazy Materialization

By default every FileReference field is downloaded before the task runs. Mark a field Lazy to skip the automatic download:

from typing import Annotated
from application_sdk.contracts.types import FileReference, Lazy

class MyInput(Input):
    # This is always downloaded before the task runs:
    manifest: FileReference | None = None

    # This is LEFT AS A DURABLE REF — not downloaded automatically:
    full_export: Annotated[FileReference | None, Lazy()] = None

Inside the task, call fetch() to download on demand:

from application_sdk.storage.reference import fetch

async def my_task(self, input: MyInput) -> MyOutput:
    if input.full_export and self.needs_full_export():
        ref = await fetch(input.full_export)  # downloads now, cached for this task
        with open(ref.local_path, "rb") as f:
            data = f.read()

When to use Lazy:

  • Your task only sometimes needs the file (conditional logic, mode flags).
  • Lightweight tasks (send-metrics, notify, query-metastore) that happen to receive an Input model with large refs they don't use — forcing a 500 MB download on a 50 ms task will cause timeouts.
  • Fan-out tasks that receive a shared manifest but only need their slice.

When NOT to use Lazy:

  • The task always reads the file — eager download is simpler and you get the download logged and retried before your code runs.

Dedup — Same File, Multiple Fields

If multiple Input fields point at the same storage_path (e.g. a shared spec file passed through a pipeline), the SDK downloads it exactly once:

class PipelineInput(Input):
    spec: FileReference | None = None          # storage_path = "file_refs/abc"
    spec_copy: FileReference | None = None     # storage_path = "file_refs/abc"  (same)

Both fields end up with the same local_path after materialization. A file_ref.materialize.dedup_hit DEBUG event is emitted for the second ref. You do not need to deduplicate manually.

The same holds across activities. local_path is a deterministic function of (run, stage, entity), so concurrent activities dispatched together can hand the SDK the same destination. The materialise-and-verify step holds a per-path lock and every download stages in .sdk-partial/ before an atomic os.replace publish, so the second activity waits and reuses the file, and a reader never observes a partial download at local_path (CONNECT-1126).

Part 2 — App.upload()

What It Does

App.upload() uploads a local file to a stable, run-scoped path in the deployment object store and returns a FileReference pointing at it. Use it when:

  • You want the artifact to outlive the current run.
  • You need to push a file to Atlan's artifact layer (visible in the UI, downloadable by support, indexed by the platform).
  • You need a FileReference with a known prefix (not a random file_refs/{uuid}).

Signature

await self.upload(input: UploadInput) -> UploadOutput

UploadInput Fields

class UploadInput(BaseModel):
    local_path: str = ""                       # path to the file or directory on disk
    ref: FileReference | None = None           # upload from a store key instead of local disk
    tier: StorageTier = StorageTier.RETAINED   # where to store it
    storage_path: str | None = None            # override the full destination key
    storage_subdir: str | None = None          # append a subdir under the run prefix
    skip_if_exists: bool = False               # skip upload when remote SHA-256 matches
    raise_on_empty: bool = False               # raise if the upload produces 0 files
Field Required Description
local_path One of local_path / ref Path to the file or directory to upload. Defaults to "".
ref One of local_path / ref Existing FileReference to upload from: its storage_path is the deployment-store source key used when local_path is absent on this pod. See Uploading without a local copy. Symmetric with DownloadInput.ref.
tier No (default RETAINED) Tier that controls the destination prefix and cleanup policy.
storage_path No Fully-qualified destination key. Overrides the auto-generated path. Use this when you need an exact fixed path (e.g. argo-artifacts/spec.json).
storage_subdir No Subdirectory appended under the run prefix. Useful for grouping related uploads without spelling out the full path.
skip_if_exists No (default False) When True, skip uploading files whose SHA-256 already matches the stored {key}.sha256 sidecar. Useful for retried tasks and idempotent re-uploads.
raise_on_empty No (default False) When True, raise StorageEmptyUploadError if the upload completes with zero files (e.g. local_path pointed at an empty directory). Leave False for incremental extractors where a quiet run legitimately produces no output; set True when zero files indicates a bug.

Path Computation

When storage_path is not set, the destination key is computed as:

{app_prefix}/{filename}

Where app_prefix depends on the tier:

Tier app_prefix (= destination key without filename)
RETAINED (default) artifacts/apps/{app_name}/workflows/{run_id}
TRANSIENT file_refs
PERSISTENT persistent-artifacts/apps/{app_name}

Note: the {uuid} segment (e.g. file_refs/{uuid}/tables.parquet) only appears when the FileReference activity interceptor auto-persists a ref returned from a @task — that path is computed by StorageTier.auto_persist_key(), not by App.upload(). App.upload() uses StorageTier.upload_prefix(), which puts the file directly under {app_prefix}/{filename}.

When storage_subdir is set, it is appended between the run prefix and the filename:

artifacts/apps/{app_name}/workflows/{run_id}/{storage_subdir}/{filename}

TRANSIENT and PERSISTENT tiers ignore storage_subdir.

Uploading without a local copy

App.upload() does not require the artifacts to be on the calling pod. When local_path is absent, it streams from the deployment store at ref.storage_path instead of the local filesystem — so it works when the tasks that produced the tree ran on other workers (KEDA-scaled SDR), when the writer deleted its local copy (use_consolidation=True), or when the caller only ever had a FileReference:

# The transformed tree is in the deployment store; this pod never held it.
result = await self.upload(
    UploadInput(
        ref=staged_ref,                    # source: deployment store at staged_ref.storage_path
        storage_path=staged_ref.storage_path,  # pin the destination → key-preserving
        tier=StorageTier.RETAINED,
    )
)

Two behaviours worth knowing:

  • Pin storage_path to the ref's own prefix and the copy is key-preserving — each key keeps its name in the destination store. Leave it off and the artifacts land under the canonical prefix for the tier (see the table above).
  • A copy whose source and destination are the same object is already satisfied: it is recognised and skipped (reason="skipped:same_object") rather than moving bytes. That is what makes the two-store dual write free in this shape. A ref pointing at a key that was never written is not silently treated as satisfied — it fails with StorageNotFoundError.

A partially-present local directory gets both halves when the destination is a different store — i.e. an SDR hand-off to the upstream store: the local files are uploaded and anything present only in the deployment store is streamed, so the destination copy is complete regardless of which pod produced what. Within a single store the local directory stays authoritative and store-only files are not merged in; there is no second store to reconcile against, so a partial local tree publishes as partial.

Deriving the ref is automatic — passing local_path alone gives you the same fallback, since the SDK derives ref from it. Pass ref explicitly when the source key is not the mirror of a local path. Either way, the artifact must have reached the deployment store in the first place: that happens when a @task returns a FileReference (the interceptor persists it), and a ref created with is_durable=True opts out of that copy — leaving nothing to fall back to.

UploadOutput

class UploadOutput(BaseModel):
    ref: FileReference    # durable ref pointing at the uploaded file
    synced: bool = False  # True if at least one file was actually transferred
    reason: str = ""      # human-readable outcome, e.g. "uploaded" or "skipped:hash_match"
Field Description
ref Durable FileReference with is_durable=True and storage_path set. file_count reflects the number of files uploaded (1 for a single file, N for a directory).
synced True if at least one file was transferred to the store. False when all files were skipped (e.g. skip_if_exists=True and every file already matched its SHA-256 sidecar).
reason Short string describing the transfer outcome. Typical values: "uploaded", "skipped:hash_match", "skipped:same_object" (source and destination are the same object — see Uploading without a local copy), "empty". Useful for logging.

The returned ref is already durable — you can pass it as a FileReference field on a subsequent task's Input.

Usage Patterns

Pattern 1 — Upload a run artifact (most common)

from application_sdk.contracts.storage import UploadInput
from application_sdk.contracts.types import StorageTier

# Inside run() or a @task method:
upload_result = await self.upload(
    UploadInput(local_path="/tmp/output/tables.parquet")  # RETAINED is the default tier
)
# upload_result.ref is now a durable FileReference
# storage_path will be something like:
# artifacts/apps/myapp/workflows/run-abc123/tables.parquet

Pattern 2 — Upload to a known subdir (e.g. argo-artifacts pattern)

upload_result = await self.upload(
    UploadInput(
        local_path="/tmp/spec.json",
        storage_subdir="argo-artifacts",
    )
)
# storage_path: artifacts/apps/myapp/workflows/.../run-.../argo-artifacts/spec.json

Pattern 3 — Upload to a fully-specified path

upload_result = await self.upload(
    UploadInput(
        local_path="/tmp/spec.json",
        storage_path="argo-artifacts/my-connector/latest-spec.json",
    )
)
# storage_path: argo-artifacts/my-connector/latest-spec.json (exactly)

Pattern 4 — Upload then pass the ref to another task

# In the extract task:
upload_result = await self.upload(UploadInput(local_path=extract_output.db_path))

# Pass the durable ref to the transform task:
transform_input = TransformInput(source_db=upload_result.ref)
transform_output = await self.transform(transform_input)

Pattern 5 — PERSISTENT for cross-run state

# Write and upload a watermark after each successful run:
watermark_path = Path(TEMPORARY_PATH) / "watermark.json"
watermark_path.write_text(json.dumps({"last_run": arrow.utcnow().isoformat()}))

await self.upload(
    UploadInput(
        local_path=str(watermark_path),
        tier=StorageTier.PERSISTENT,
        storage_path=f"persistent-artifacts/apps/{self._app_name}/watermark.json",
    )
)

upload() vs FileReference — Side-by-Side

# ── FileReference (inter-task, automatic) ─────────────────────────────────

class ExtractOutput(Output):
    raw_db: FileReference | None = None     # tier=TRANSIENT by default

# In the extract task — just set local_path:
return ExtractOutput(raw_db=FileReference(local_path=str(db_path)))

# In the transform task — local_path is already set by the SDK:
async def transform(self, input: TransformInput) -> TransformOutput:
    db = sqlite3.connect(input.raw_db.local_path)   # ready to use


# ── App.upload() (outbound, stable path) ──────────────────────────────────

# In the run() method — push result for the platform to index:
result = await self.upload(
    UploadInput(local_path=transform_output.result_parquet.local_path)  # RETAINED is the default
)
# result.ref.storage_path is stable and survives workflow completion

Handing a declaration on — verify_refs() and upload_refs()

A FileReference names one object. A prefix names whatever a listing happens to return — and a listing cannot tell "absent" from "lost". A transformed/ tree short by one entity looks exactly like a run that only produced three, so a consumer that walks the prefix (the publish app does) diffs the tenant against the subset and archives everything the walk missed.

APP-CORRECTNESS-001 is the rule: verify cross-activity handoffs against a producer-written declaration, never by listing a storage prefix. The refs the producing tasks returned are that declaration — App.verify_refs() is what turns them from a value nobody read into an assertion.

Call it from run() at the point a step stops passing typed refs and starts passing a prefix:

from application_sdk.contracts.storage import VerifyRefsInput

results = await asyncio.gather(*transform_tasks)
refs = [r.transformed_file for r in results if r.transformed_file is not None]

await self.verify_refs(
    VerifyRefsInput(
        refs=[r.model_copy(update={"auto_materialize": False}) for r in refs],
        prefix=transformed_data_prefix,
    )
)

Each ref is checked by its storage_path, used verbatim — matching the key persist_file_reference wrote, so the check asks about the key that actually exists rather than a re-normalised guess at it.

A ref counts as a directory ref when its key ends in / (what the interceptor's directory branch writes) or its file_count is anything other than 1 (which catches a hand-pinned directory key with no slash). Those are listed — with sidecars excluded, since every object carries a {key}.sha256 and a raw listing would return roughly 2N keys for N files — and must yield at least file_count data objects. Everything else is checked with a HEAD.

Both signals are needed: a directory holding exactly one file has file_count == 1, so discriminating on the count alone would HEAD its prefix as an object key and 404 on a tree that is perfectly intact.

With prefix set, a ref that resolves outside it fails too: that is data the downstream walk will never reach, which is the same hole by a different route. Anything missing raises StorageHandoffIncompleteError (DATA_INTEGRITY, non-retryable) naming the keys, so the run fails at the producer instead of surfacing as a short publish two stages later.

Set auto_materialize=False on the copies you pass in. Verification is a metadata lookup; without it the interceptor would download every declared file onto whichever pod runs the check.

store selects which store to assert against and defaults to StoreTarget.DEPLOYMENT — the store the interceptor persists refs to. Pass StoreTarget.UPSTREAM after an App.upload() fan-out to assert what the publish app will actually read.

SqlApp.run() does all of this for you and surfaces the refs it verified on ExtractionOutput.transformed_files, so a connector's own Atlan bridge can hand each one to App.upload(ref=...) instead of scanning a directory that, on a fanned-out run, was written by pods it never shared a filesystem with.

A run() override that adds an entity the default run() does not drive must finish with SqlApp.finalize_extraction(base, [extra_transform_output]), not by concatenating onto base.transformed_files by hand. super().run() verified the four refs it drove and then returned, so a ref appended afterwards is never asserted — the override reads as if it verified everything while shipping an unverified entity, which is this same shortfall one layer up.

Delivering the declaration — App.upload_refs()

verify_refs() asserts a declaration. upload_refs() delivers one: it takes the same list and lands every ref under a single destination prefix, then verifies what it delivered. This is the fan-in half of the same handoff, and it is the loop connectors that cross the SDR store boundary have been hand-rolling.

The hand-rolled version usually scans os.path.join(output_path, "transformed") and uploads whatever it finds. That is wrong on any run where the transforms fanned out: the directory is written by the activities, so the calling pod sees only the subset that happened to run locally — and on a fully distributed run, nothing at all. upload_refs() passes each ref instead, so the SDK streams from the deployment store for the files this pod never held.

from application_sdk.contracts.storage import DeclaredFile, UploadRefsInput

delivered = await self.upload_refs(
    UploadRefsInput(
        files=[DeclaredFile(ref=r) for r in extraction.transformed_files],
        source_prefix=extraction.transformed_data_prefix,
        prefix=f"artifacts/apps/{app}/workflows/{wf_id}/{run_id}/transformed",
    )
)
return MyOutput(transformed_data_prefix=delivered.prefix)

Three things it does not leave optional:

  • Every upload opts into raise_on_empty=True. A declared file that contributes zero objects is a hole in the tree the consumer will walk.
  • An empty declaration returns an empty prefix. delivered.prefix is "" when there was nothing to deliver, so you hand "" downstream rather than a prefix naming an empty tree. An empty tree is not a quiet no-op to a consumer that diffs against it — PublishNode reads it as "delete everything".
  • The delivery is verified against the declaration in the store it landed in, before the task returns.

Destination keys come from one of two explicit knobs, never a guess: DeclaredFile(label=...) names the leaf directly (use this when the refs' own keys carry no structure, e.g. file_refs/<uuid>.json), or source_prefix is stripped from each ref's key to give the leaf (use this when the keys already have the shape the destination should keep). A declared file that yields neither raises UnplaceableDeclaredFileError — a rule that recovers the entity segment from four refs and drops it from one would reshape the tree on exactly the small runs nobody inspects.

Part 3 — End-to-End Examples

Example A — Extract → Transform → Upload (typical connector)

# contracts.py
class ExtractOutput(Output):
    raw_db: FileReference | None = None          # TRANSIENT: temp inter-task file

class TransformInput(Input):
    raw_db: FileReference | None = None

class TransformOutput(Output):
    result_parquet: FileReference | None = None  # RETAINED: kept after run


# connector.py
class MyConnector(App):
    @task
    async def extract(self, input: ExtractInput) -> ExtractOutput:
        db_path = Path(TEMPORARY_PATH) / "raw.db"
        run_extraction(db_path)
        return ExtractOutput(raw_db=FileReference(local_path=str(db_path)))
        # SDK uploads raw.db to file_refs/{uuid} automatically on return

    @task
    async def transform(self, input: TransformInput) -> TransformOutput:
        # SDK downloaded raw.db before this method was called
        df = read_sqlite(input.raw_db.local_path)
        out = Path(TEMPORARY_PATH) / "result.parquet"
        df.to_parquet(out)
        return TransformOutput(
            result_parquet=FileReference(local_path=str(out), tier=StorageTier.RETAINED)
        )
        # SDK uploads result.parquet to artifacts/.../run-.../file_refs/result.parquet

    async def run(self, input: RunInput) -> RunOutput:
        extract_out = await self.extract(ExtractInput(...))
        transform_out = await self.transform(
            TransformInput(raw_db=extract_out.raw_db)
        )
        # Push artifact so the platform can index it:
        upload = await self.upload(
            UploadInput(
                local_path=transform_out.result_parquet.local_path,
                storage_subdir="argo-artifacts",
            )
        )
        return RunOutput(artifact_ref=upload.ref)

Example B — Fan-out with shared manifest and lazy heavy ref

class ParseInput(Input):
    manifest: FileReference | None = None                             # eager
    full_snapshot: Annotated[FileReference | None, Lazy()] = None    # lazy

class MyConnector(App):
    @task
    async def parse(self, input: ParseInput) -> ParseOutput:
        # manifest is already downloaded — use it directly
        items = json.loads(Path(input.manifest.local_path).read_text())

        if self.requires_full_snapshot(items):
            # Only download the large snapshot when actually needed
            snap_ref = await fetch(input.full_snapshot)
            snapshot_data = load_snapshot(snap_ref.local_path)

        return ParseOutput(results=process(items))

Example C — Persistent watermark across runs

async def run(self, input: RunInput) -> RunOutput:
    # Try to read the watermark from the previous run:
    watermark_key = f"persistent-artifacts/apps/{self._app_name}/watermark.json"
    try:
        local_wm = await download_file(watermark_key, "/tmp/watermark.json")
        last_run = json.loads(Path("/tmp/watermark.json").read_text())["last_run"]
    except StorageNotFoundError:
        last_run = None   # first run

    # ... do the work ...

    # Write updated watermark:
    Path("/tmp/watermark.json").write_text(json.dumps({"last_run": now_iso()}))
    await self.upload(UploadInput(
        local_path="/tmp/watermark.json",
        storage_path=watermark_key,
        tier=StorageTier.PERSISTENT,
    ))

Part 4 — Common Pitfalls

Do not construct FileReference outside a @task

# WRONG — module level or __init__:
MY_REF = FileReference(local_path="/tmp/file.bin")   # breaks Temporal sandbox

# CORRECT — inside a @task function body:
@task
async def my_task(self, input) -> Output:
    ref = FileReference(local_path="/tmp/file.bin")
    return Output(ref=ref)

Do not embed large bytes in Input/Output directly

# WRONG:
class BadOutput(Output):
    raw_bytes: bytes = b""         # will hit 2 MB Temporal limit

# CORRECT:
class GoodOutput(Output):
    data_file: FileReference | None = None

Do not put FileReference in a dict key

# WRONG — the SDK walks dict values, not keys:
data = {ref: "some_value"}         # ref will not be persisted

# CORRECT:
data = {"file": ref}

Do not read local_path before the task runs

# WRONG — in run(), local_path is a stale hint, not guaranteed to exist:
path = input.ref.local_path
data = open(path).read()           # may not exist on this machine

# CORRECT — read local_path only inside @task, after the SDK has materialised it:
@task
async def my_task(self, input: MyInput) -> MyOutput:
    data = open(input.ref.local_path).read()   # safe: SDK set this before call

Do not use FileReference for outbound delivery

# WRONG — FileReference is internal to the run:
return Output(ref=FileReference(local_path=str(result), tier=StorageTier.RETAINED))
# The Atlan platform cannot index this on its own

# CORRECT — use App.upload() to push to a stable, platform-visible path:
await self.upload(UploadInput(local_path=str(result)))  # RETAINED is the default tier

Part 5 — Observability

All events below are structured log records routed through OTLP. They appear in Grafana under the application_sdk.storage logger with the fields listed.

Events for files ≥ 10 MiB are logged at INFO; smaller files at DEBUG.

Persist events

Event Fields
file_ref.persist.start storage_path, local_path, file_size_bytes, tier
file_ref.persist.complete storage_path, bytes_uploaded, duration_ms, sha256
file_ref.persist.failed storage_path, local_path, error_type, bytes_uploaded

Materialize events

Event Fields
file_ref.materialize.start storage_path, file_size_bytes, is_cache_hit=False
file_ref.materialize.complete storage_path, bytes_downloaded, duration_ms, sha256
file_ref.materialize.skipped storage_path, local_path, is_cache_hit=True
file_ref.materialize.failed storage_path, bytes_transferred_before_failure, error_type
file_ref.materialize.dedup_hit storage_path, dedup_key, reused_local_path, local_path
file_ref.materialize.lazy_skipped storage_path

Low-level transfer events

Event Fields
storage.upload success store_path, size_bytes, elapsed_ms, throughput_mibps
storage.download success store_path, size_bytes, elapsed_ms, throughput_mibps
storage.upload failure store_path, error_class, elapsed_ms
storage.download failure store_path, error_class, elapsed_ms

Part 6 — Migration from upload_to_atlan

BaseMetadataExtractor.upload_to_atlan is deprecated. It still works (it now just forwards to App.upload internally), but emits a DeprecationWarning and will be removed in the next major SDK release.

Deprecated:

await self.upload_to_atlan(UploadInput(output_path=input.output_path))

Replacement:

up = await self.upload(UploadInput(local_path=input.output_path))
records_uploaded = up.ref.file_count or 0

App.upload() is concurrent, SHA-256 verified, and works identically in hosted and SDR deployments.