API reference

Examples on this page build on these imports and definitions:

from dataclasses import dataclass
from datetime import datetime
from pathlib import Path

import fs_schema as fss

Public API at a glance

All names in this table are available from fs_schema.

Defining schemas

Subclass Schema. Put the layout in schema. Nested mapping = nested directory. One Schema base. No methods, mixins, or extra metaclasses.

@dataclass
class Manifest:
    delivery_id: str
    rows: int


class Batch(fss.Schema):
    schema = {
        "parts": fss.File(fmt="part-{part:d}.parquet"),
    }


class Delivery(fss.Schema):
    schema = {
        "transfer": {
            fss.FILES: ["download.log", "request.json"],
        },
        "manifest": fss.File("manifest.json", schema=Manifest),
        "batches": {
            fss.Dir(
                alias="days",
                fmt="{day:%Y-%m-%d}",
                sort=lambda match: match.kwargs["day"],
            ): Batch,
        },
    }

Mapping forms

name is the only positional arg. Need name, fmt, or match. Dir(...): Child → child type is Child. Inline {...} → private Schema with stable identity. Dir(schema=Required) → class-time subset check. RHS still owns children.

Templates, matching, and allowed match counts

name + match validates that basename. fmt + match filters; captures stay from fmt. match alone: unnamed groups → args, named → kwargs. Optional groups → None.

Untyped format fields → str. :d → int. Datetime fields → datetime.

part_decl = fss.File(
    fmt="part-{part:d}.{ext}",
    match=r"(?i).+\.(?:json|yaml|toml)",
)
day_decl = fss.Dir(alias="days", fmt=fss.dt("%Y-%m-%d"), min=0)

fss.dt("%Y-%m-%d") → "{ts:%Y-%m-%d}". Capture is kwargs["ts"]. fss.dt("%Y-%m-%d", "day") names that capture. fss.dt("%Y-%m-%d", "") stays positional.

Inheritance and replacement

Subclass layout starts from the base. Add or replace nodes.

class AuditedDelivery(Delivery):
    schema = {
        "audit": "audit.json",
    }

Alias, name, or fmt collision fails at class creation. Matching keys replace the inherited node. Anonymous match-only decls append.

Declaration API

Exact basename:

File(
    name: str,
    *,
    alias: str | None = None,
    match: str | None = None,
    optional: bool = False,
    schema: type[T] | Callable[[Path], T] | None = None,
) -> File[T]

Dir(
    name: str,
    *,
    alias: str | None = None,
    match: str | None = None,
    optional: bool = False,
    schema: type[S] | None = None,
) -> Dir[S]

Collection (fmt and/or match, no name):

File(
    *,
    fmt: FmtLike,
    match: str | None = None,
    min: int = 1,
    max: int | None = None,
    alias: str | None = None,
    sort: Callable[[Match], str | int | float | datetime | Located] | None = None,
    sort_rev: bool = False,
    skip_mismatch: bool = False,
    schema: type[T] | Callable[[Path], T] | None = None,
) -> File[T]

Dir(
    *,
    fmt: FmtLike,
    match: str | None = None,
    min: int = 1,
    max: int | None = None,
    alias: str | None = None,
    sort: Callable[[Match], str | int | float | datetime | Located] | None = None,
    sort_rev: bool = False,
    skip_mismatch: bool = False,
    schema: type[S] | None = None,
) -> Dir[S]

Regex-only collection omits fmt and supplies match instead.

dt(pattern: str) -> FmtLike

The mapping table above defines the valid Layout key-value pairings.

Applying schemas

bind checks the tree. Success → that Schema class. Failure → MismatchErr. Fixed Dir(...): Child → Child. Repeated Dir(...): Batch → each item is a Batch with captures. Exact optional=True → None if absent. Write after bind via bound.root().

result = Delivery.bind("/srv/incoming/delivery-42")
if not result:
    print(result)
else:
    delivery: Delivery = result

MismatchErr is falsy. Bound schemas are truthy. if not result works when the type stays S | MismatchErr. Cross-schema returns still need is_mismatch.

delivery = fss.raise_mismatch(
    Delivery.bind("/srv/incoming/delivery-42")
)

Schema.bind takes a path. A string, a Path, and any value with __fspath__ all work, including a schema node. A planned root binds itself.

validated_again = Delivery.bind(delivery)
planned_delivery = Delivery.relative_to("/srv/incoming/delivery-42")
validated_from_plan = planned_delivery.bind()

Bind checks structure and counts now. Ignores extras. First mismatch wins.

Schema.bind(root: str | os.PathLike[str]) -> Self | MismatchErr
SchemaRoot[S].bind() -> S | MismatchErr
is_mismatch(x: object) -> TypeIs[MismatchErr]
raise_mismatch(x: T | MismatchErr) -> T

Using schemas

Fixed directories and files

A bound Schema, planned SchemaRoot, fixed directory, or fixed file represents one path and satisfies Located. Each works with os.fspath and APIs that accept os.PathLike[str].

root_path = delivery.path
manifest_path = Path(delivery.manifest)
request_text = delivery.transfer.request_json.read_text()
request_exists = delivery.transfer.request_json.exists()

Located promises only .path and __fspath__. Concrete fixed values also have exists(). Fixed files add read_bytes(), read_text(), and create().

Children support attribute access and exact item lookup. An explicit alias is used unchanged. Otherwise, each run outside [A-Za-z0-9_] in the disk name becomes _.

log_file = delivery.transfer.download_log
same_request = delivery.transfer["request.json"]

Use exists_opt(path) when an absent optional node should become None rather than a missing path. exists_opt(None) is None.

exists_opt(path: str | os.PathLike[str] | None) -> Path | None

Template collections and matches

A template declaration provides a sequence of its current matches. The sequence has the parent directory in .path, but it is not Located. Indexing or iteration returns a concrete Match, which is Located and contains parsed captures.

days = delivery.batches.days
parent_directory = days.path
latest_day = days[-1]
day_value = latest_day.kwargs["day"]

for day in days:
    for part in day.parts:
        part_number = part.kwargs.part
        part_path = part.path

args contains unnamed captures. kwargs supports mapping and attribute access for named captures. Optional regex captures can be None. Use mapping lookup for names such as "items" that collide with mapping methods.

A predicate receives the pair (match.args, match.kwargs). Filtering returns a materialized result that keeps collection capabilities such as slicing and, for format-backed collections, format(). Empty results are collections too.

where combines all supplied constraints. Positional values match a capture prefix, with None acting as a wildcard for that position; named values match a subset, where None matches an optional capture value. Unknown named capture fields raise KeyError, even on an empty planned collection. More positional values than a match has simply reject that match. Calling where() without constraints returns an equivalent collection.

get accepts positive and negative indexes. With no explicit default, an out-of-range index returns None; an explicit None or other default is returned unchanged. Other errors are not hidden. This makes one-result queries concise without changing ordinary sequence indexing.

def is_selected_day(args, kwargs):
    return kwargs["day"] == datetime(2026, 9, 10)

selected_days = days.filter(is_selected_day)
selected_day = days.find(is_selected_day)
selected_by_capture = days.where(day=datetime(2026, 9, 10))
selected_by_capture_or_none = selected_by_capture.get()

filter(predicate, /) -> Self
find(predicate, /) -> Match | None
where(*args: str | int | datetime | None,
      **kwargs: str | int | datetime | None) -> Self
get(index: int = 0) -> Match | None
get(index: int, default: T) -> Match | T
get(*, default: T) -> Match | T

Capture names are dynamic, so static checking does not try to narrow keyword names passed to where.

Formatting a format-backed collection plans one concrete path without I/O or captures. A formatted directory remains recursively navigable, and a formatted file retains its read, load, and put methods. Regex-only collections and individual matches are not formattable.

planned_day = days.format(day=datetime(2026, 9, 11))

Public path and match protocols

@runtime_checkable
class Located(Protocol):
    @property
    def path(self) -> Path: ...
    def __fspath__(self) -> str: ...

@runtime_checkable
class Match(Located, Protocol):
    @property
    def args(self) -> tuple[str | int | datetime | None, ...]: ...
    @property
    def kwargs(self) -> Mapping[str, str | int | datetime | None]: ...

The concrete kwargs mapping also supports capture access by attribute, as shown above.

Static typing of dynamic children

Dynamically resolved class children use the bare type boundary. This is broad enough for direct recursive navigation such as Root.inline.leaf, but it cannot preserve whether each dynamically named child is a file class, collection class, or exact generated Schema subtype. Instance child access likewise returns a union of possible child kinds. Delivery.RootT remains type[SchemaRoot[Schema]].

Use an explicit fss.SchemaRoot[Delivery] annotation when the schema parameter must remain exact. Fixed Schema-backed values retain their exact classes; only dynamic static lookup loses that precision.

Creating with schemas

relative_to picks a root and fills in the paths. It does not write anything. create writes that plan and hands it back. bind reads the tree and checks that it matches the schema.

A file argument is whatever put accepts. A directory argument names child aliases. Leave an optional child out, or pass None, and it stays absent. An unknown alias raises KeyError. create() with no arguments creates that directory and every required child directory. It does not write files, and it does not create optional directories or collection members.

You do not create a collection as a whole. format(**captures) names one member. When that member is a directory, the captures are remembered by every template under it. parse(basename) names one member from a filename you already have, and does not remember captures. create on the member writes it. A basename that does not fit raises ValueError there, rather than later at bind.

A list of members can go in the spec. Each item is a (captures, payload) pair: the captures are the format arguments, and the payload is what you would pass to create on that member. A mapping of filename to body writes those names as given. The keys are filenames. If a key is a capture name instead of a filename, the error says so.

If a file collection's captures were all remembered by an enclosing format, pass the file body by itself. A directory collection does not do that, because a mapping there already means filenames. A bare body that is still missing a capture raises TypeError and names the missing one. Two different values for the same capture raise ValueError.

fs: fss.SchemaRoot[Delivery] = Delivery.relative_to("/srv/curated/delivery-42")
event_date = datetime(2026, 9, 10)
manifest = Manifest("delivery-42", 1_000)
parquet_bytes = b"parquet payload"
fs.create(
    manifest=manifest,
    batches={"days": [({"day": event_date}, {"parts": [({"part": 0}, parquet_bytes)]})]},
)
another = fs.batches.days.format(day=datetime(2026, 9, 11))
another.parts.format(part=0).create(b"next")
created = fs.bind()

create on a bound schema raises TypeError. Call root() and write on the plan. A bound file can still create; that is put.

bind() returns the schema when the tree matches, and MismatchErr when it does not. root() starts a fresh plan at the same path. That plan has not been checked. bind() lives on the plan from relative_to or root(). Keep that plan if you will bind after writing a child.

fs: fss.SchemaRoot[Delivery] = delivery.root()

Reading and writing

manifest: Manifest = fss.raise_exn(delivery.manifest.load())


def load_manifest(path: Path) -> Manifest:
    return Manifest(path.stem, 0)


custom_manifest = fss.File(
    "manifest.custom",
    schema=load_manifest,
)

put writes one file and creates any missing parent directories. The finished contents replace the previous file in one step, so a reader sees either the old file or the new one. Pass None, or omit the body, and the file is empty. The body can be bytes, text, another file to copy, an object with save, or a dataclass stored as JSON. save is called on the file that then becomes the destination. Install fs-schema[mashumaro] for dataclass JSON, or fs-schema[orjson] to use its faster encoder. load() returns a decoding failure as a value. raise_exn raises that failure and keeps the success type.

put(path, data=None) -> None
raise_exn(value: T | Exception) -> T