fastflowtransform.packages

PackageDependency dataclass

Dependency entry from a package's own manifest (project.yml).

Example in project.yml inside the package:

dependencies:
  - name: shared.core
    version: ">=0.8,<1.0"
    optional: false

Source code in src/fastflowtransform/packages.py

@dataclass
class PackageDependency:
    """
    Dependency entry from a package's own manifest (project.yml).

    Example in project.yml inside the package:

        dependencies:
          - name: shared.core
            version: ">=0.8,<1.0"
            optional: false
    """

    name: str
    version_constraint: str | None = None
    optional: bool = False
    raw: Mapping[str, Any] = field(default_factory=dict)

PackageManifest dataclass

Package-level metadata loaded from project.yml inside the package.

Source code in src/fastflowtransform/packages.py

@dataclass
class PackageManifest:
    """
    Package-level metadata loaded from project.yml inside the package.
    """

    name: str
    version: str
    fft_version: str | None
    dependencies: list[PackageDependency] = field(default_factory=list)
    models_dir: str | None = None
    raw: Mapping[str, Any] = field(default_factory=dict)
    root: Path | None = None

LockedSource dataclass

Concrete, pinned source info that ends up in packages.lock.yml.

Source code in src/fastflowtransform/packages.py

@dataclass
class LockedSource:
    """
    Concrete, pinned source info that ends up in packages.lock.yml.
    """

    kind: str  # "path" | "git"
    path: str | None = None
    git: str | None = None
    rev: str | None = None
    subdir: str | None = None

    def to_mapping(self) -> dict[str, Any]:
        out: dict[str, Any] = {"kind": self.kind}
        if self.path is not None:
            out["path"] = self.path
        if self.git is not None:
            out["git"] = self.git
        if self.rev is not None:
            out["rev"] = self.rev
        if self.subdir is not None:
            out["subdir"] = self.subdir
        return out

LockFile dataclass

packages.lock.yml structure.

Right now we only write it, we do not use it to drive resolution.

Source code in src/fastflowtransform/packages.py

@dataclass
class LockFile:
    """
    packages.lock.yml structure.

    Right now we only *write* it, we do not use it to drive resolution.
    """

    fft_version: str | None
    entries: list[LockEntry] = field(default_factory=list)

    @classmethod
    def from_mapping(cls, data: Mapping[str, Any]) -> LockFile:
        entries: list[LockEntry] = []
        for row in data.get("packages", []) or []:
            src = row.get("source") or {}
            source = LockedSource(
                kind=src.get("kind", "path"),
                path=src.get("path"),
                git=src.get("git"),
                rev=src.get("rev"),
                subdir=src.get("subdir"),
            )
            entries.append(
                LockEntry(
                    name=row["name"],
                    version=str(row["version"]),
                    source=source,
                )
            )
        return cls(
            fft_version=data.get("fft_version"),
            entries=entries,
        )

    def to_mapping(self) -> dict[str, Any]:
        return {
            "fft_version": self.fft_version,
            "packages": [e.to_mapping() for e in self.entries],
        }

ResolvedPackage dataclass

Concrete package that has been:

• located (path or git checkout)
• manifest-loaded (project.yml)
• dependency-validated

Source code in src/fastflowtransform/packages.py

@dataclass
class ResolvedPackage:
    """
    Concrete package that has been:

      - located (path or git checkout)
      - manifest-loaded (project.yml)
      - dependency-validated
    """

    name: str
    version: str
    root: Path  # directory containing project.yml + models/
    models_dir: str  # path inside root where models live
    source: LockedSource
    manifest: PackageManifest

resolve_packages

resolve_packages(project_dir, cfg=None)

Resolve all packages declared in packages.yml for the given project:

• locate local path packages
• clone/fetch git packages into .fastflowtransform/packages
• load per-package project.yml as a manifest (name/version/fft_version/dependencies)

• validate:

  • manifest.name matches spec.name
  • manifest.fft_version is compatible with FFT_VERSION (if declared)
  • spec.version (constraint) matches manifest.version (if declared)
  • inter-package dependencies are satisfied

• write packages.lock.yml with pinned sources

Returns a list of ResolvedPackage objects. If packages.yml is missing or empty, returns [].

Source code in src/fastflowtransform/packages.py

def resolve_packages(
    project_dir: Path,
    cfg: PackagesConfig | None = None,
) -> list[ResolvedPackage]:
    """
    Resolve all packages declared in packages.yml for the given project:

      - locate local path packages
      - clone/fetch git packages into .fastflowtransform/packages
      - load per-package project.yml as a manifest (name/version/fft_version/dependencies)
      - validate:

          * manifest.name matches spec.name
          * manifest.fft_version is compatible with FFT_VERSION (if declared)
          * spec.version (constraint) matches manifest.version (if declared)
          * inter-package dependencies are satisfied

      - write packages.lock.yml with pinned sources

    Returns a list of ResolvedPackage objects. If packages.yml is missing or
    empty, returns [].
    """
    project_dir = Path(project_dir).expanduser().resolve()

    if cfg is None:
        cfg = load_packages_config(project_dir)

    specs: list[PackageSpec] = list(cfg.packages or [])
    if not specs:
        return []

    cache_dir = project_dir / ".fastflowtransform" / "packages"
    cache_dir.mkdir(parents=True, exist_ok=True)

    manifests_by_name: dict[str, PackageManifest] = {}
    resolved_by_name: dict[str, ResolvedPackage] = {}

    for spec in specs:
        root = _materialize_package_source(project_dir, cache_dir, spec)
        manifest = _load_package_manifest(root)

        # Name check: spec.name must match manifest.name
        if spec.name and manifest.name != spec.name:
            raise RuntimeError(
                f"Package name mismatch for spec '{spec.name}': "
                f"manifest reports '{manifest.name}' in {root / 'project.yml'}."
            )

        # Check FFT core compatibility if declared in manifest
        if manifest.fft_version and not version_satisfies(FFT_VERSION, manifest.fft_version):
            raise RuntimeError(
                f"Package '{manifest.name}' ({manifest.version}) "
                f"requires FFT version '{manifest.fft_version}', "
                f"but running '{FFT_VERSION}'."
            )

        # Check that the spec's own version constraint (if any) is satisfied
        if spec.version and not version_satisfies(manifest.version, spec.version):
            raise RuntimeError(
                f"Package '{manifest.name}' has version {manifest.version} "
                f"but spec requires '{spec.version}'."
            )

        if manifest.name in manifests_by_name:
            other = manifests_by_name[manifest.name]
            raise RuntimeError(
                f"Duplicate package name '{manifest.name}' loaded from:\n"
                f"  - {other.root}\n"
                f"  - {root}\n"
                "Package names must be globally unique."
            )

        manifests_by_name[manifest.name] = manifest

        # Resolve models_dir: packages.yml overrides manifest.models_dir; default "models"
        models_dir = spec.models_dir or manifest.models_dir or "models"

        locked_src = _lock_source_for_spec(spec, root)
        resolved_by_name[manifest.name] = ResolvedPackage(
            name=manifest.name,
            version=manifest.version,
            root=root,
            models_dir=models_dir,
            source=locked_src,
            manifest=manifest,
        )

    # Validate dependencies (only across the set of resolved packages)
    _validate_package_dependencies(resolved_by_name)

    # Write lock file (best-effort; failure shouldn't be fatal)
    _write_lock_file(project_dir, list(resolved_by_name.values()))

    # Return in deterministic order
    return sorted(resolved_by_name.values(), key=lambda p: p.name)

parse_version

parse_version(v)

Parse a very simple semver string 'MAJOR.MINOR.PATCH'.

We ignore pre-release / build metadata; they are treated as equal.

Raises ValueError if we cannot parse.

Source code in src/fastflowtransform/packages.py

def parse_version(v: str) -> tuple[int, int, int]:
    """
    Parse a very simple semver string 'MAJOR.MINOR.PATCH'.

    We ignore pre-release / build metadata; they are treated as equal.

    Raises ValueError if we cannot parse.
    """
    m = _SEMVER_RE.match(v.strip())
    if not m:
        raise ValueError(f"Invalid semantic version (expected 'x.y.z'): {v!r}")
    return int(m.group(1)), int(m.group(2)), int(m.group(3))

compare_versions

compare_versions(a, b)

Compare two version strings.

<0: a < b 0: a == b

0: a > b

Source code in src/fastflowtransform/packages.py

def compare_versions(a: str, b: str) -> int:
    """
    Compare two version strings.

      <0: a < b
       0: a == b
      >0: a > b
    """
    a_t = parse_version(a)
    b_t = parse_version(b)
    if a_t < b_t:
        return -1
    if a_t > b_t:
        return 1
    return 0

version_satisfies

version_satisfies(actual, constraint)

Return True iff a version string 'actual' satisfies a constraint expression.

Empty / None constraint always returns True.

Source code in src/fastflowtransform/packages.py

def version_satisfies(actual: str, constraint: str | None) -> bool:
    """
    Return True iff a version string 'actual' satisfies a constraint expression.

    Empty / None constraint always returns True.
    """
    if not constraint:
        return True
    checks = _parse_constraints(constraint)
    for op, target in checks:
        cmp = compare_versions(actual, target)
        if op == "==":
            if cmp != 0:
                return False
        elif op == "!=":
            if cmp == 0:
                return False
        elif op == ">":
            if cmp <= 0:
                return False
        elif op == "<":
            if cmp >= 0:
                return False
        elif op == ">=":
            if cmp < 0:
                return False
        elif op == "<=":
            if cmp > 0:
                return False
        else:  # pragma: no cover
            raise ValueError(f"Unknown operator in version constraint: {op!r}")
    return True