Week 1 · Ingest anything › Day 3

Day 3. Landing files from folders and shared drives

1. Why this matters

The most common data source in medium-sized companies is still a folder: a network share, a OneDrive/SharePoint library synced to a server, a vendor's export directory. Folder ingestion looks trivial and is where most silent errors live: the file read twice, the file read while still being written, the file replaced with a corrected version under the same name.

2. The landing pattern

1. Discover: list candidate files (glob), skip anything that isn't finished.
2. Fingerprint: compute a SHA-256 of the bytes.
3. Check the manifest: have we already landed this exact content? If yes, skip and log duplicate.
4. Copy into the raw zone at a path that encodes source and business date: raw/qsr/business_date=2026-09-14/pos_Q204__<sha8>.csv.
5. Record a manifest row: source, original path, size, mtime, sha256, received_at, raw_path, status.

You copy, never move. The source folder belongs to someone else; moving files breaks their process and destroys your ability to re-check.

import hashlib, shutil, sqlite3, time
from pathlib import Path

def sha256(path: Path, chunk=1 << 20) -> str:
    h = hashlib.sha256()
    with path.open("rb") as f:
        for block in iter(lambda: f.read(chunk), b""):
            h.update(block)
    return h.hexdigest()

def is_stable(path: Path, wait=2.0) -> bool:
    """A file still being written changes size. Skip partials and temp names."""
    if path.suffix in {".partial", ".tmp", ".crdownload"} or path.name.startswith("~$"):
        return False
    s1 = path.stat().st_size; time.sleep(wait); s2 = path.stat().st_size
    return s1 == s2 and s1 > 0

def land(src_dir: Path, raw_dir: Path, db: sqlite3.Connection, source: str):
    db.execute("""create table if not exists manifest(
        sha256 text primary key, source text, src_path text, size int, mtime real,
        received_at text, raw_path text)""")
    for p in sorted(src_dir.rglob("*.csv")):
        if not is_stable(p):
            continue
        digest = sha256(p)
        if db.execute("select 1 from manifest where sha256=?", (digest,)).fetchone():
            continue                                  # exact same bytes already landed
        business_date = p.parent.name                 # folder is the date here
        dest = raw_dir / source / f"business_date={business_date}" / f"{p.stem}__{digest[:8]}{p.suffix}"
        dest.parent.mkdir(parents=True, exist_ok=True)
        tmp = dest.with_suffix(dest.suffix + ".tmp")
        shutil.copy2(p, tmp); tmp.replace(dest)       # atomic rename: readers never see half a file
        db.execute("insert into manifest values (?,?,?,?,?,datetime('now'),?)",
                   (digest, source, str(p), p.stat().st_size, p.stat().st_mtime, str(dest)))
        db.commit()

3. Same name, new content

Vendors often fix a file and re-drop it under the same name. Your manifest will see a new hash for an existing (source, src_path). Decide the rule explicitly and write it down:

4. Watching vs polling

You can poll (list the folder every N minutes) or watch (OS events via the watchdog library). Polling is simpler, works on network shares where events are unreliable, and is what Dagster sensors do in Week 3. Use watching only for local disks with latency needs under a minute.