vendor_fabric.secrets_sync.tools

Provider capability functions for native vendor-fabric secret synchronization.

Module Contents

Classes

ValidateConfigSchema

Schema for validating a secrets sync configuration.

RunPipelineSchema

Schema for running the secrets sync pipeline.

GetConfigInfoSchema

Schema for getting configuration information.

Functions

validate_config

Validate a secrets sync pipeline configuration file.

run_pipeline

Run the secrets synchronization pipeline.

dry_run

Perform a dry run to see what changes would be made.

get_config_info

Get detailed information about a pipeline configuration.

get_targets

Get the list of sync targets from a configuration.

get_sources

Get the list of secret sources from a configuration.

Data

API

class vendor_fabric.secrets_sync.tools.ValidateConfigSchema(/, **data: Any)

Bases: pydantic.BaseModel

Schema for validating a secrets sync configuration.

Initialization

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

config_path: str = 'Field(...)'
model_config: ClassVar[pydantic.config.ConfigDict] = 'ConfigDict(...)'
classmethod model_fields() dict[str, pydantic.fields.FieldInfo]
classmethod model_computed_fields() dict[str, pydantic.fields.ComputedFieldInfo]
property model_extra: dict[str, Any] | None
property model_fields_set: set[str]
classmethod model_construct(_fields_set: set[str] | None = None, **values: Any) typing_extensions.Self
model_copy(*, update: collections.abc.Mapping[str, Any] | None = None, deep: bool = False) typing_extensions.Self
model_dump(*, mode: Literal[json, python] | str = 'python', include: pydantic.main.IncEx | None = None, exclude: pydantic.main.IncEx | None = None, context: Any | None = None, by_alias: bool | None = None, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False, exclude_computed_fields: bool = False, round_trip: bool = False, warnings: bool | Literal[none, warn, error] = True, fallback: Callable[[Any], Any] | None = None, serialize_as_any: bool = False, polymorphic_serialization: bool | None = None) dict[str, Any]
model_dump_json(*, indent: int | None = None, ensure_ascii: bool = False, include: pydantic.main.IncEx | None = None, exclude: pydantic.main.IncEx | None = None, context: Any | None = None, by_alias: bool | None = None, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False, exclude_computed_fields: bool = False, round_trip: bool = False, warnings: bool | Literal[none, warn, error] = True, fallback: Callable[[Any], Any] | None = None, serialize_as_any: bool = False, polymorphic_serialization: bool | None = None) str
classmethod model_json_schema(by_alias: bool = True, ref_template: str = DEFAULT_REF_TEMPLATE, schema_generator: type[pydantic.json_schema.GenerateJsonSchema] = GenerateJsonSchema, mode: pydantic.json_schema.JsonSchemaMode = 'validation', *, union_format: Literal[any_of, primitive_type_array] = 'any_of') dict[str, Any]
classmethod model_parametrized_name(params: tuple[type[Any], ...]) str
model_post_init(context: Any, /) None
classmethod model_rebuild(*, force: bool = False, raise_errors: bool = True, _parent_namespace_depth: int = 2, _types_namespace: pydantic._internal._namespace_utils.MappingNamespace | None = None) bool | None
classmethod model_validate(obj: Any, *, strict: bool | None = None, extra: pydantic.config.ExtraValues | None = None, from_attributes: bool | None = None, context: Any | None = None, by_alias: bool | None = None, by_name: bool | None = None) typing_extensions.Self
classmethod model_validate_json(json_data: str | bytes | bytearray, *, strict: bool | None = None, extra: pydantic.config.ExtraValues | None = None, context: Any | None = None, by_alias: bool | None = None, by_name: bool | None = None) typing_extensions.Self
classmethod model_validate_strings(obj: Any, *, strict: bool | None = None, extra: pydantic.config.ExtraValues | None = None, context: Any | None = None, by_alias: bool | None = None, by_name: bool | None = None) typing_extensions.Self
dict(*, include: pydantic.main.IncEx | None = None, exclude: pydantic.main.IncEx | None = None, by_alias: bool = False, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False) Dict[str, Any]
json(*, include: pydantic.main.IncEx | None = None, exclude: pydantic.main.IncEx | None = None, by_alias: bool = False, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False, encoder: Callable[[Any], Any] | None = PydanticUndefined, models_as_dict: bool = PydanticUndefined, **dumps_kwargs: Any) str
classmethod parse_obj(obj: Any) typing_extensions.Self
classmethod parse_raw(b: str | bytes, *, content_type: str | None = None, encoding: str = 'utf8', proto: pydantic.deprecated.parse.Protocol | None = None, allow_pickle: bool = False) typing_extensions.Self
classmethod parse_file(path: str | pathlib.Path, *, content_type: str | None = None, encoding: str = 'utf8', proto: pydantic.deprecated.parse.Protocol | None = None, allow_pickle: bool = False) typing_extensions.Self
classmethod from_orm(obj: Any) typing_extensions.Self
classmethod construct(_fields_set: set[str] | None = None, **values: Any) typing_extensions.Self
copy(*, include: pydantic._internal._utils.AbstractSetIntStr | pydantic._internal._utils.MappingIntStrAny | None = None, exclude: pydantic._internal._utils.AbstractSetIntStr | pydantic._internal._utils.MappingIntStrAny | None = None, update: Dict[str, Any] | None = None, deep: bool = False) typing_extensions.Self
classmethod schema(by_alias: bool = True, ref_template: str = DEFAULT_REF_TEMPLATE) Dict[str, Any]
classmethod schema_json(*, by_alias: bool = True, ref_template: str = DEFAULT_REF_TEMPLATE, **dumps_kwargs: Any) str
classmethod validate(value: Any) typing_extensions.Self
classmethod update_forward_refs(**localns: Any) None
class vendor_fabric.secrets_sync.tools.RunPipelineSchema(/, **data: Any)

Bases: pydantic.BaseModel

Schema for running the secrets sync pipeline.

Initialization

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

config_path: str = 'Field(...)'
dry_run: bool = 'Field(...)'
operation: str = 'Field(...)'
targets: str | None = 'Field(...)'
continue_on_error: bool = 'Field(...)'
model_config: ClassVar[pydantic.config.ConfigDict] = 'ConfigDict(...)'
classmethod model_fields() dict[str, pydantic.fields.FieldInfo]
classmethod model_computed_fields() dict[str, pydantic.fields.ComputedFieldInfo]
property model_extra: dict[str, Any] | None
property model_fields_set: set[str]
classmethod model_construct(_fields_set: set[str] | None = None, **values: Any) typing_extensions.Self
model_copy(*, update: collections.abc.Mapping[str, Any] | None = None, deep: bool = False) typing_extensions.Self
model_dump(*, mode: Literal[json, python] | str = 'python', include: pydantic.main.IncEx | None = None, exclude: pydantic.main.IncEx | None = None, context: Any | None = None, by_alias: bool | None = None, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False, exclude_computed_fields: bool = False, round_trip: bool = False, warnings: bool | Literal[none, warn, error] = True, fallback: Callable[[Any], Any] | None = None, serialize_as_any: bool = False, polymorphic_serialization: bool | None = None) dict[str, Any]
model_dump_json(*, indent: int | None = None, ensure_ascii: bool = False, include: pydantic.main.IncEx | None = None, exclude: pydantic.main.IncEx | None = None, context: Any | None = None, by_alias: bool | None = None, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False, exclude_computed_fields: bool = False, round_trip: bool = False, warnings: bool | Literal[none, warn, error] = True, fallback: Callable[[Any], Any] | None = None, serialize_as_any: bool = False, polymorphic_serialization: bool | None = None) str
classmethod model_json_schema(by_alias: bool = True, ref_template: str = DEFAULT_REF_TEMPLATE, schema_generator: type[pydantic.json_schema.GenerateJsonSchema] = GenerateJsonSchema, mode: pydantic.json_schema.JsonSchemaMode = 'validation', *, union_format: Literal[any_of, primitive_type_array] = 'any_of') dict[str, Any]
classmethod model_parametrized_name(params: tuple[type[Any], ...]) str
model_post_init(context: Any, /) None
classmethod model_rebuild(*, force: bool = False, raise_errors: bool = True, _parent_namespace_depth: int = 2, _types_namespace: pydantic._internal._namespace_utils.MappingNamespace | None = None) bool | None
classmethod model_validate(obj: Any, *, strict: bool | None = None, extra: pydantic.config.ExtraValues | None = None, from_attributes: bool | None = None, context: Any | None = None, by_alias: bool | None = None, by_name: bool | None = None) typing_extensions.Self
classmethod model_validate_json(json_data: str | bytes | bytearray, *, strict: bool | None = None, extra: pydantic.config.ExtraValues | None = None, context: Any | None = None, by_alias: bool | None = None, by_name: bool | None = None) typing_extensions.Self
classmethod model_validate_strings(obj: Any, *, strict: bool | None = None, extra: pydantic.config.ExtraValues | None = None, context: Any | None = None, by_alias: bool | None = None, by_name: bool | None = None) typing_extensions.Self
dict(*, include: pydantic.main.IncEx | None = None, exclude: pydantic.main.IncEx | None = None, by_alias: bool = False, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False) Dict[str, Any]
json(*, include: pydantic.main.IncEx | None = None, exclude: pydantic.main.IncEx | None = None, by_alias: bool = False, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False, encoder: Callable[[Any], Any] | None = PydanticUndefined, models_as_dict: bool = PydanticUndefined, **dumps_kwargs: Any) str
classmethod parse_obj(obj: Any) typing_extensions.Self
classmethod parse_raw(b: str | bytes, *, content_type: str | None = None, encoding: str = 'utf8', proto: pydantic.deprecated.parse.Protocol | None = None, allow_pickle: bool = False) typing_extensions.Self
classmethod parse_file(path: str | pathlib.Path, *, content_type: str | None = None, encoding: str = 'utf8', proto: pydantic.deprecated.parse.Protocol | None = None, allow_pickle: bool = False) typing_extensions.Self
classmethod from_orm(obj: Any) typing_extensions.Self
classmethod construct(_fields_set: set[str] | None = None, **values: Any) typing_extensions.Self
copy(*, include: pydantic._internal._utils.AbstractSetIntStr | pydantic._internal._utils.MappingIntStrAny | None = None, exclude: pydantic._internal._utils.AbstractSetIntStr | pydantic._internal._utils.MappingIntStrAny | None = None, update: Dict[str, Any] | None = None, deep: bool = False) typing_extensions.Self
classmethod schema(by_alias: bool = True, ref_template: str = DEFAULT_REF_TEMPLATE) Dict[str, Any]
classmethod schema_json(*, by_alias: bool = True, ref_template: str = DEFAULT_REF_TEMPLATE, **dumps_kwargs: Any) str
classmethod validate(value: Any) typing_extensions.Self
classmethod update_forward_refs(**localns: Any) None
class vendor_fabric.secrets_sync.tools.GetConfigInfoSchema(/, **data: Any)

Bases: pydantic.BaseModel

Schema for getting configuration information.

Initialization

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

config_path: str = 'Field(...)'
model_config: ClassVar[pydantic.config.ConfigDict] = 'ConfigDict(...)'
classmethod model_fields() dict[str, pydantic.fields.FieldInfo]
classmethod model_computed_fields() dict[str, pydantic.fields.ComputedFieldInfo]
property model_extra: dict[str, Any] | None
property model_fields_set: set[str]
classmethod model_construct(_fields_set: set[str] | None = None, **values: Any) typing_extensions.Self
model_copy(*, update: collections.abc.Mapping[str, Any] | None = None, deep: bool = False) typing_extensions.Self
model_dump(*, mode: Literal[json, python] | str = 'python', include: pydantic.main.IncEx | None = None, exclude: pydantic.main.IncEx | None = None, context: Any | None = None, by_alias: bool | None = None, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False, exclude_computed_fields: bool = False, round_trip: bool = False, warnings: bool | Literal[none, warn, error] = True, fallback: Callable[[Any], Any] | None = None, serialize_as_any: bool = False, polymorphic_serialization: bool | None = None) dict[str, Any]
model_dump_json(*, indent: int | None = None, ensure_ascii: bool = False, include: pydantic.main.IncEx | None = None, exclude: pydantic.main.IncEx | None = None, context: Any | None = None, by_alias: bool | None = None, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False, exclude_computed_fields: bool = False, round_trip: bool = False, warnings: bool | Literal[none, warn, error] = True, fallback: Callable[[Any], Any] | None = None, serialize_as_any: bool = False, polymorphic_serialization: bool | None = None) str
classmethod model_json_schema(by_alias: bool = True, ref_template: str = DEFAULT_REF_TEMPLATE, schema_generator: type[pydantic.json_schema.GenerateJsonSchema] = GenerateJsonSchema, mode: pydantic.json_schema.JsonSchemaMode = 'validation', *, union_format: Literal[any_of, primitive_type_array] = 'any_of') dict[str, Any]
classmethod model_parametrized_name(params: tuple[type[Any], ...]) str
model_post_init(context: Any, /) None
classmethod model_rebuild(*, force: bool = False, raise_errors: bool = True, _parent_namespace_depth: int = 2, _types_namespace: pydantic._internal._namespace_utils.MappingNamespace | None = None) bool | None
classmethod model_validate(obj: Any, *, strict: bool | None = None, extra: pydantic.config.ExtraValues | None = None, from_attributes: bool | None = None, context: Any | None = None, by_alias: bool | None = None, by_name: bool | None = None) typing_extensions.Self
classmethod model_validate_json(json_data: str | bytes | bytearray, *, strict: bool | None = None, extra: pydantic.config.ExtraValues | None = None, context: Any | None = None, by_alias: bool | None = None, by_name: bool | None = None) typing_extensions.Self
classmethod model_validate_strings(obj: Any, *, strict: bool | None = None, extra: pydantic.config.ExtraValues | None = None, context: Any | None = None, by_alias: bool | None = None, by_name: bool | None = None) typing_extensions.Self
dict(*, include: pydantic.main.IncEx | None = None, exclude: pydantic.main.IncEx | None = None, by_alias: bool = False, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False) Dict[str, Any]
json(*, include: pydantic.main.IncEx | None = None, exclude: pydantic.main.IncEx | None = None, by_alias: bool = False, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False, encoder: Callable[[Any], Any] | None = PydanticUndefined, models_as_dict: bool = PydanticUndefined, **dumps_kwargs: Any) str
classmethod parse_obj(obj: Any) typing_extensions.Self
classmethod parse_raw(b: str | bytes, *, content_type: str | None = None, encoding: str = 'utf8', proto: pydantic.deprecated.parse.Protocol | None = None, allow_pickle: bool = False) typing_extensions.Self
classmethod parse_file(path: str | pathlib.Path, *, content_type: str | None = None, encoding: str = 'utf8', proto: pydantic.deprecated.parse.Protocol | None = None, allow_pickle: bool = False) typing_extensions.Self
classmethod from_orm(obj: Any) typing_extensions.Self
classmethod construct(_fields_set: set[str] | None = None, **values: Any) typing_extensions.Self
copy(*, include: pydantic._internal._utils.AbstractSetIntStr | pydantic._internal._utils.MappingIntStrAny | None = None, exclude: pydantic._internal._utils.AbstractSetIntStr | pydantic._internal._utils.MappingIntStrAny | None = None, update: Dict[str, Any] | None = None, deep: bool = False) typing_extensions.Self
classmethod schema(by_alias: bool = True, ref_template: str = DEFAULT_REF_TEMPLATE) Dict[str, Any]
classmethod schema_json(*, by_alias: bool = True, ref_template: str = DEFAULT_REF_TEMPLATE, **dumps_kwargs: Any) str
classmethod validate(value: Any) typing_extensions.Self
classmethod update_forward_refs(**localns: Any) None
vendor_fabric.secrets_sync.tools.validate_config(config_path: str) extended_data.containers.ExtendedDict

Validate a secrets sync pipeline configuration file.

Args: config_path: Path to the YAML configuration file

Returns: Dict with ‘valid’ (bool) and ‘message’ (str) fields

vendor_fabric.secrets_sync.tools.run_pipeline(config_path: str, dry_run: bool = False, operation: str = 'pipeline', targets: str | None = None, continue_on_error: bool = True) extended_data.containers.ExtendedDict

Run the secrets synchronization pipeline.

This executes the two-phase pipeline (merge → sync) to synchronize secrets from HashiCorp Vault to AWS Secrets Manager.

Args: config_path: Path to the YAML configuration file dry_run: If true, compute diff but don’t make changes operation: ‘merge’, ‘sync’, or ‘pipeline’ (full) targets: Comma-separated list of targets (empty for all) continue_on_error: Continue if errors occur

Returns: Dict with sync results including success, counts, and any errors

vendor_fabric.secrets_sync.tools.dry_run(config_path: str) extended_data.containers.ExtendedDict

Perform a dry run to see what changes would be made.

Args: config_path: Path to the YAML configuration file

Returns: Dict with what would be changed, including diff output

vendor_fabric.secrets_sync.tools.get_config_info(config_path: str) extended_data.containers.ExtendedDict

Get detailed information about a pipeline configuration.

Args: config_path: Path to the YAML configuration file

Returns: Dict with configuration details including sources and targets

vendor_fabric.secrets_sync.tools.get_targets(config_path: str) extended_data.containers.ExtendedDict

Get the list of sync targets from a configuration.

Args: config_path: Path to the YAML configuration file

Returns: Dict with ‘targets’ list and any error message

vendor_fabric.secrets_sync.tools.get_sources(config_path: str) extended_data.containers.ExtendedDict

Get the list of secret sources from a configuration.

Args: config_path: Path to the YAML configuration file

Returns: Dict with ‘sources’ list and any error message

vendor_fabric.secrets_sync.tools.TOOL_DEFINITIONS = None