alembic_engine/
lib.rs

1//! engine orchestration: load, validate, plan, apply.
2
3mod adapter_ops;
4mod apply_retry;
5mod drift;
6mod endpoint;
7mod errors;
8pub mod external;
9mod extract;
10mod inflect;
11pub mod journal;
12mod loader;
13pub mod mapping;
14mod pipeline;
15mod planner;
16mod predicate;
17mod pretty_printing;
18mod render;
19#[cfg(feature = "starlark")]
20mod starlark_transforms;
21mod state;
22mod transform;
23mod types;
24use alembic_core::{key_string, validate_inventory, Inventory, Object, ValidationReport};
25use anyhow::{anyhow, Result};
26
27#[cfg(test)]
28mod tests;
29
30pub use adapter_ops::{
31    backend_id_from_value, build_key_from_schema, build_request_body, normalize_attrs_refs,
32    query_filters_from_key, resolve_value_for_type, resolved_ids_from_state, resolved_ids_identity,
33    state_mappings_by_id, StateMappings,
34};
35pub use apply_retry::{
36    apply_non_delete_journaled, apply_non_delete_with_retries, describe_missing_refs,
37    is_missing_ref_error, RetryApplyDriver, RetryApplyResult,
38};
39pub use drift::{ChangedEntry, DriftEntry, DriftReport};
40pub use endpoint::normalize_endpoint;
41pub use errors::AdapterApplyError;
42pub use external::{
43    run_external_adapter, ExternalAdapter, ExternalEnvelope, ExternalEnvelopeRef, ExternalObject,
44    ExternalRequest, ExternalRequestRef, ExternalResponse, EXTERNAL_PROTOCOL_VERSION,
45};
46pub use extract::{import_inventory, ImportReport};
47pub use inflect::pluralize;
48pub use journal::Journal;
49pub use loader::load_inventory;
50pub use planner::{plan, sort_ops_for_apply};
51pub use state::{PostgresTlsMode, StateData, StateStore};
52pub use transform::{compile_map, eval_map_transform, load_map_spec, MapSpec, TransformsSpec};
53pub use types::{
54    Adapter, AppliedOp, ApplyReport, Backend, BackendId, Emitter, FieldChange, ObservedObject,
55    ObservedState, Observer, Op, Plan, PlanSummary, ProvisionReport,
56};
57
58/// validate an inventory and return the report.
59pub fn validate(inventory: &Inventory) -> ValidationReport {
60    validate_inventory(inventory)
61}
62
63/// helper to format a validation report into a Result.
64pub fn report_to_result(report: ValidationReport) -> Result<()> {
65    report_to_result_with_sources(report, &[])
66}
67
68/// helper to format a validation report with source locations into a Result.
69pub fn report_to_result_with_sources(report: ValidationReport, objects: &[Object]) -> Result<()> {
70    if report.is_ok() {
71        return Ok(());
72    }
73
74    let located_errors = report.with_sources(objects);
75    let mut message = String::from("validation failed:\n");
76    for error in located_errors {
77        message.push_str(&format!("- {error}\n"));
78    }
79    Err(anyhow!(message))
80}
81
82/// observe backend state and produce a deterministic plan.
83pub async fn build_plan(
84    adapter: &(dyn Observer + '_),
85    inventory: &Inventory,
86    state: &mut StateStore,
87    allow_delete: bool,
88) -> Result<Plan> {
89    let observed = pipeline::observe(adapter, inventory, state).await?;
90    Ok(plan(
91        &inventory.objects,
92        &observed,
93        state,
94        &inventory.schema,
95        allow_delete,
96    ))
97}
98
99pub(crate) fn bootstrap_state_from_observed(
100    state: &mut StateStore,
101    desired: &[Object],
102    observed: &ObservedState,
103) -> bool {
104    let mut updated = false;
105    for object in desired {
106        if state
107            .backend_id(object.type_name.clone(), object.uid)
108            .is_some()
109        {
110            continue;
111        }
112        if let Some(obs) = observed
113            .by_key
114            .get(&(object.type_name.clone(), key_string(&object.key)))
115        {
116            if obs.type_name != object.type_name {
117                continue;
118            }
119            if let Some(backend_id) = &obs.backend_id {
120                state.set_backend_id(object.type_name.clone(), object.uid, backend_id.clone());
121                updated = true;
122            }
123        }
124    }
125    updated
126}
127
128/// apply a plan and update the state store. full adapters provision schema
129/// before writing; emitters only write.
130pub async fn apply_plan(
131    backend: &Backend,
132    plan: &Plan,
133    state: &mut StateStore,
134    allow_delete: bool,
135) -> Result<ApplyReport> {
136    pipeline::apply(backend, plan, state, allow_delete).await
137}
alembic_engine/lib.rs

alembic_engine/
lib.rs
