fs-schema

Typed schemas for filesystem layouts. Dataclass-like declarations turn directory contracts into validated, navigable Python values.

uv add "fs-schema[mashumaro,orjson]"

from dataclasses import dataclass

import fs_schema as fss

@dataclass
class Contents:
    title: str

class DataDownload(fss.Schema):
    schema = {
        "packs": {
            fss.FILES: ["upload.log", "request.log"],
            fss.Dir(alias="days", fmt="{day:%Y-%m-%d}"): {
                "parts": fss.File(fmt="{stem}.{ext}", max=4),
            },
        },
        "contents": fss.File("contents.json", schema=Contents),
    }

download = fss.raise_mismatch(DataDownload.bind("."))
with open(download.packs.upload_log, encoding="utf-8") as stream:
    print(stream.read())
part = download.packs.days[-1].parts[-1]
print(part.kwargs.stem, part.path.stat().st_size)
contents: Contents = fss.raise_exn(download.contents.load())
print(contents.title)

Binding validates an existing layout. Templates are collections; indexing selects a concrete match whose parsed captures are available through .args and .kwargs. relative_to() plans output paths without claiming they exist. Format-backed collections plan concrete files or recursively navigable directories; only the top-level plan binds the whole schema. Dataclass schemas use Mashumaro for JSON; install the mashumaro or faster orjson extra.

Expanded quickstart

# pyright: reportAttributeAccessIssue=false, reportCallIssue=false
# pyright: reportIndexIssue=false, reportGeneralTypeIssues=false
# pyright: reportArgumentType=false, reportAssignmentType=false
# pyright: reportUnknownMemberType=false, reportUnknownVariableType=false
# pyright: reportUnknownArgumentType=false

from collections.abc import Sequence
from dataclasses import dataclass
from pathlib import Path

import fs_schema as fss


@dataclass
class Manifest:
    delivery_id: str
    expected_parts: int


@dataclass
class CuratedManifest:
    delivery_id: str
    partitions: int


class Batch(fss.Schema):
    schema = {
        # Typed numeric captures make this collection sort numerically
        "parts": fss.File(
            fmt="part-{part:d}.jsonl",
            sort=lambda part: part.kwargs.part,
        ),
    }


class Delivery(fss.Schema):
    schema = {
        "manifest": fss.File("manifest.json", schema=Manifest),
        "batches": {
            # days names the repeated directory collection in Python
            fss.Dir(
                alias="days",
                fmt="{day:%Y-%m-%d}",
                sort=lambda day: day.kwargs.day,
            ): Batch,
        },
    }


class Curated(fss.Schema):
    schema = {
        "manifest": fss.File("manifest.json", schema=CuratedManifest),
        fss.Dir(alias="days", fmt="{day:%Y-%m-%d}"): {
            "parts": fss.File(fmt="part-{part:d}.parquet"),
        },
    }


def curate(source: Path, target: Path) -> Curated | fss.MismatchErr:
    # Bind proves the input layout before any child is used
    if fss.is_mismatch(delivery := Delivery.bind(source)):
        return delivery

    # The declared File schema supplies this dataclass loader
    manifest: Manifest = fss.raise_exn(delivery.manifest.load())
    # Templates are ordered collections; an item is one capture-bearing match
    latest = delivery.batches.days[-1]
    fs = Curated.relative_to(target)
    parquet = convert_parts([part.path for part in latest.parts])
    # One spec: exact files by alias, collection members as (captures, body) pairs.
    fs.create(
        manifest=CuratedManifest(manifest.delivery_id, partitions=1),
        days=[({"day": latest.kwargs.day}, {"parts": [({"part": 0}, parquet)]})],
    )
    return fs.bind()


def convert_parts(parts: Sequence[Path]) -> bytes:
    # Stand-in for an application converter; keeps this example executable.
    return b"\n".join(part.read_bytes() for part in parts)

Development

Dev:
  ./run.sh setup:install-pre-commit
  ./run.sh uv:venv:sync
  ./run.sh pretty / check / tests
  ./run.sh docs:serve

Maintainer:
  ./run.sh pkg:bump VERSION
  ./run.sh release:check
  ./run.sh release:tag
  ./run.sh gh:ci [ref]
  ./run.sh gh:docs [ref]
  ./run.sh gh:publish:testpypi [ref]
  ./run.sh gh:publish:pypi
  ./run.sh gh:release
  ./run.sh release:testpypi
  ./run.sh release:pypi

Lock: edit pyproject.toml, then ./run.sh uv:lock
MRE: ./run.sh docker:check / docker:test

Commands:
  help                         Usual path + command list
  pkg:version                  Read or update the package version and lockfile
  pkg:bump                     Update version, lockfile, & commit
  pkg:build                    Build sdist and wheel into dist/
 Setup
  setup:install-pre-commit     Install git hooks (once per clone; sync first)
  setup:shellcheck             Install pinned shellcheck if not found
  setup:check-image            Assert /etc/os-release and apt pkgs match pinned versions
  setup:check-installs         Checks all host deps present
 Dep management
  uv:venv:sync                 Sync .venv from uv.lock (no resolve)
  uv:lock                      Write uv.lock. Edit pyproject.toml by hand first (no uv add)
  uv:lock:bootstrap            Re-create uv.lock from scratch, ignore current pins
 Dev: checks
  lint                         ruff check + shellcheck
  typecheck                    basedpyright
  check                        pre-commit + docs build
  tests                        pytest against the installed package
 Dev: formatters
  fmt                          ruff format
  pretty                       fmt + ruff check --fix
 Docs
  docs:build                   Build the docs site into site/
  docs:serve                   Serve docs locally
 CI / maintainer
  ci:tox                       Local tox matrix in parallel (one worker per CPU)
  release:check                Build, inspect, and clean-install both local artifacts
  release:tag                  Create annotated vVERSION tag at clean release HEAD
  release:testpypi             Run local TestPyPI release
  release:pypi                 Run local tagged PyPI release
  gh:ci                        Dispatch and watch remote test.yml
  gh:docs                      Dispatch and watch remote docs.yml
  gh:publish:testpypi          Dispatch, test, publish, and verify through gh
  gh:release                   Create gh release for current vVERSION tag
  gh:publish:pypi              Dispatch, test, publish, and verify prod release through gh
 Docker
  docker:build                 Build image
  docker:check                 GHA Checks job (same ./run.sh chain)
  docker:test                  Build prod image + ./run.sh tests
  docker:shell                 Dev shell. Host tree at /app, host uid
  docker:prune                 Dangling image / buildx prune
  docker:matrix                docker:test for each PYTHON_VERSIONS (local MRE; GHA uses baipp matrix)

API reference · Tutorial · Changelog · Source · MIT LICENSE

Changelog

API by git tag. Signatures: reference.

v0.6.0

create writes a plan. Nested directory overrides merge. exists_opt(None) is None.

Write relative_to, create, root, format, parse, put Creating with schemas
relative_to fills paths. create writes that plan and does not check the tree. quickstart
create on a bound schema raises TypeError: use root().
dir.create() makes that directory and required child directories. No files, no optional directories, no collection members.
None leaves an optional child absent. An unknown alias raises KeyError.
A collection takes a filename-to-body map, or a list of (captures, payload) pairs. Keys that are capture names are rejected. delivery
A file collection whose captures were set by format takes the body alone. A directory collection does not. one member
A missing capture raises TypeError and names it. Two values for one capture raise ValueError.
put replaces the path in one step. No body writes an empty file. Body: bytes, text, a file to copy, save, or a dataclass as JSON. Reading and writing
Declare Schema, File, Dir Inheritance and replacement
A nested directory override merges children with the base. A file override replaces that file.
On the class, a file or a collection exposes alias, fmt, match, min, max. Coincident.parts.match.
An exact directory with a schema class is that class. Parent.child.file.match.
A directory collection exposes fmt and match on the collection, not the child schema's fields.
Bind Schema.bind, exists_opt, MismatchErr Applying schemas
bind takes a string, a Path, or any __fspath__ value, including a schema node. A planned root binds itself.
exists_opt(None) returns None. Fixed directories and files
A wrong collection count names dangling symlinks that matched: (dangling: gone.png).
Read follows a live symlink. put replaces a symlink at that path and does not write through it.
Read Integrations
glom is no longer installed with the package. Navigation examples are in the docs.

v0.5.0

Exact names and collections use different constructors. MismatchErr is falsy.

Declare File, Dir, dt, optional, skip_mismatch Defining schemas
Exact File and Dir take a positional name. A collection is keyword-only: fmt, match, or both.
optional=True on an exact name is min=0. optional file
sort, sort_rev, and skip_mismatch on an exact name raise ValueError.
dt("%Y-%m-%d") captures ts. dt("%Y-%m-%d", "day") names it. dt("%Y-%m-%d", "") stays positional.
skip_mismatch drops a formatted name that fails match, and a directory member whose children do not match.
Bind Schema.bind, MismatchErr, is_mismatch Applying schemas
MismatchErr is falsy. A bound schema is truthy. if not result works for Schema | MismatchErr.
A result that might be another schema still needs is_mismatch.
A nested schema binds as that class. An optional exact child is None when absent.

v0.4.6

First public. Alpha. import fs_schema as fss. Python 3.10–3.14.

Declare — mapping; name | fmt | match; counts, alias, sort; one-base inherit/replace. Schema, Layout, File, Dir, FILES, dt Defining schemas
Bind — first failure; extras ignored. Schema.bind, SchemaRoot.bind, MismatchErr, is_mismatch, raise_mismatch Applying schemas
Read — attr/item access; PathLike. Located, Match, exists_opt, .path, __fspath__, .exists, .read_bytes, .read_text, .args, .kwargs, .filter, .find, .where, .get Using schemas
Write — plan then bind; JSON dataclasses via extras mashumaro, orjson. Schema.relative_to, SchemaRoot, .format, .root, .put, .load, put, raise_exn Creating with schemas
Types — generated child attrs; dynamic names stay a union. SchemaRoot[S], __version__ Static typing