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

alembic_engine/
lib.rs
