The SDK owns every primitive needed to load schema files from disk, but the orchestration exists only inside the infrahubctl command layer, coupled to a rich Console and to typer.Exit. Consumers that are not a CLI therefore cannot reuse it, and the Infrahub server has hand-rolled a divergent copy.
infrahubctl schema load (infrahub_sdk/ctl/schema.py:180) is the reference flow:
schemas_data = load_yamlfile_from_disk_and_exit(paths=schemas, file_type=SchemaFile, console=console)
validate_schema_content_and_exit(client=client, schemas=schemas_data)
response = await client.schema.load(schemas=[item.payload for item in schemas_data], branch=branch)
InfrahubRepositoryIntegrator.import_schema_files in the Infrahub server re-implements the same three steps, including its own recursive directory walk, because neither helper is callable from a Prefect flow:
load_yamlfile_from_disk_and_exit (ctl/utils.py:187) takes a Console and raises typer.Exit
validate_schema_content_and_exit (ctl/schema.py:38) prints to a module-level console and exits
The duplication has already cost a defect. Infrahub's copy omitted the invalid-file handling that load_yamlfile_from_disk_and_exit provides, so an empty or malformed schema file in a user's repository reached schema.validate(None) and surfaced as Schema not valid, found '1' error(s) ... Input should be a valid dictionary instead of the Empty YAML/JSON file reason the loader had already recorded. Note also that the CLI can safely use item.payload (the self.content or {} accessor) precisely because _and_exit filtered the invalid files first; a consumer reusing payload without that guard silently turns unloadable files into empty dicts.
Proposed change
Extract a library-level API with no console or typer coupling, for example:
@dataclass
class SchemaFileFailure:
identifier: str | None
location: Path
reason: str | None
@dataclass
class SchemaLoadResult:
payloads: list[dict[str, Any]]
failures: list[SchemaFileFailure]
def load_schema_files(paths: list[Path]) -> SchemaLoadResult: ...
built on the existing SchemaFile.load_from_disk (yaml.py:121), which already handles recursion into directories, extension filtering, and missing paths. The two _and_exit helpers then become thin wrappers that render failures and exit, so CLI behaviour is unchanged.
Consequences for the Infrahub server
import_schema_files collapses to calling load_schema_files and mapping failures onto its own ValidationError, deleting roughly 20 lines including the hand-rolled is_file()/is_dir() walk and its find_files() call.
Why this matters for testing
The loading and failure semantics currently have no cheap home. Exercising them in Infrahub means the full integration fixture: Neo4j, a running API, and a cloned git repository, because import_schema_files resolves paths through a git worktree, even though the failure path touches neither the client nor the database. Once the logic lives here, the same behaviour is a unit test over tmp_path files that runs in milliseconds and covers both consumers.
Caveats to settle during implementation
- Infrahub's walk uses a git-aware
find_files(..., branch_name, commit, directory) while load_from_disk does a plain recursive glob. The worktree directory is already resolved before the walk, so they are probably equivalent, but that needs verifying rather than assuming.
load_from_disk raises FileNotValidError for a path that does not exist, whereas Infrahub currently logs a warning and silently skips, so a repository whose .infrahub.yml lists a non-existent schema path imports nothing and reports no failure. Adopting the SDK behaviour changes that, arguably for the better.
- The SDK is consumed as a submodule, so this lands here first and the Infrahub side follows with a pointer bump.
The SDK owns every primitive needed to load schema files from disk, but the orchestration exists only inside the
infrahubctlcommand layer, coupled to a richConsoleand totyper.Exit. Consumers that are not a CLI therefore cannot reuse it, and the Infrahub server has hand-rolled a divergent copy.infrahubctl schema load(infrahub_sdk/ctl/schema.py:180) is the reference flow:InfrahubRepositoryIntegrator.import_schema_filesin the Infrahub server re-implements the same three steps, including its own recursive directory walk, because neither helper is callable from a Prefect flow:load_yamlfile_from_disk_and_exit(ctl/utils.py:187) takes aConsoleand raisestyper.Exitvalidate_schema_content_and_exit(ctl/schema.py:38) prints to a module-levelconsoleand exitsThe duplication has already cost a defect. Infrahub's copy omitted the invalid-file handling that
load_yamlfile_from_disk_and_exitprovides, so an empty or malformed schema file in a user's repository reachedschema.validate(None)and surfaced asSchema not valid, found '1' error(s) ... Input should be a valid dictionaryinstead of theEmpty YAML/JSON filereason the loader had already recorded. Note also that the CLI can safely useitem.payload(theself.content or {}accessor) precisely because_and_exitfiltered the invalid files first; a consumer reusingpayloadwithout that guard silently turns unloadable files into empty dicts.Proposed change
Extract a library-level API with no console or typer coupling, for example:
built on the existing
SchemaFile.load_from_disk(yaml.py:121), which already handles recursion into directories, extension filtering, and missing paths. The two_and_exithelpers then become thin wrappers that renderfailuresand exit, so CLI behaviour is unchanged.Consequences for the Infrahub server
import_schema_filescollapses to callingload_schema_filesand mappingfailuresonto its ownValidationError, deleting roughly 20 lines including the hand-rolledis_file()/is_dir()walk and itsfind_files()call.Why this matters for testing
The loading and failure semantics currently have no cheap home. Exercising them in Infrahub means the full integration fixture: Neo4j, a running API, and a cloned git repository, because
import_schema_filesresolves paths through a git worktree, even though the failure path touches neither the client nor the database. Once the logic lives here, the same behaviour is a unit test overtmp_pathfiles that runs in milliseconds and covers both consumers.Caveats to settle during implementation
find_files(..., branch_name, commit, directory)whileload_from_diskdoes a plain recursive glob. The worktree directory is already resolved before the walk, so they are probably equivalent, but that needs verifying rather than assuming.load_from_diskraisesFileNotValidErrorfor a path that does not exist, whereas Infrahub currently logs a warning and silently skips, so a repository whose.infrahub.ymllists a non-existent schema path imports nothing and reports no failure. Adopting the SDK behaviour changes that, arguably for the better.