Implement parameter substitution for migrations

Cargo.toml

@@ -155,6 +155,7 @@ uuid = "1.1.2"
 
 # Common utility crates
 dotenvy = { version = "0.15.7", default-features = false }
+subst = "0.3.7"
 
 # Runtimes
 [workspace.dependencies.async-std]

README.md

@@ -458,6 +458,26 @@ opt-level = 3
 <sup>1</sup> The `dotenv` crate itself appears abandoned as of [December 2021](https://github.com/dotenv-rs/dotenv/issues/74)
 so we now use the `dotenvy` crate instead. The file format is the same.
 
+## Parameter Substitution for Migrations 
+
+You can parameterize migrations using parameters, either from the environment or passed in from the cli or to the Migrator. 
+
+For example:
+
+```sql
+-- enable-substitution 
+CREATE USER ${USER_FROM_ENV} WITH PASSWORD ${PASSWORD_FROM_ENV}
+-- disable-substituion 
+```
+
+We use the [subst](https://crates.io/crates/subst) to support substitution. sqlx supports
+
+- Short format: `$NAME`
+- Long format: `${NAME}`
+- Default values: `${NAME:Bob}`
+- Recursive Substitution in Default Values: `${NAME: Bob ${OTHER_NAME: and Alice}}`
+
+
 ## Safety
 
 This crate uses `#![forbid(unsafe_code)]` to ensure everything is implemented in 100% Safe Rust.

sqlx-cli/README.md

@@ -65,6 +65,18 @@ any scripts that are still pending.
 
 ---
 
+Users can also provide parameters through environment variables or pass them in manually.
+
+```bash
+sqlx migrate run --params-from-env
+```
+
+```bash
+sqlx migrate run --params key:value,key1,value1
+```
+
+---
+
 Users can provide the directory for the migration scripts to `sqlx migrate` subcommands with the `--source` flag.
 
 ```bash
@@ -105,6 +117,16 @@ Creating migrations/20211001154420_<name>.up.sql
 Creating migrations/20211001154420_<name>.down.sql
 ```
 
+Users can also provide parameters through environment variables or pass them in manually, just as they did with the run command.
+
+```bash
+sqlx migrate revert --params-from-env
+```
+
+```bash
+sqlx migrate revert --params key:value,key1,value1
+```
+
 ### Enable building in "offline mode" with `query!()`
 
 There are 2 steps to building with "offline mode":

sqlx-cli/src/database.rs

@@ -57,7 +57,16 @@ pub async fn reset(
 
 pub async fn setup(migration_source: &str, connect_opts: &ConnectOpts) -> anyhow::Result<()> {
     create(connect_opts).await?;
-    migrate::run(migration_source, connect_opts, false, false, None).await
+    migrate::run(
+        migration_source,
+        connect_opts,
+        false,
+        false,
+        None,
+        false,
+        Vec::with_capacity(0),
+    )
+    .await
 }
 
 async fn ask_to_continue_drop(db_url: String) -> bool {

sqlx-cli/src/lib.rs

@@ -66,13 +66,17 @@ async fn do_run(opt: Opt) -> Result<()> {
                 ignore_missing,
                 connect_opts,
                 target_version,
+                params_from_env,
+                params,
             } => {
                 migrate::run(
                     &source,
                     &connect_opts,
                     dry_run,
                     *ignore_missing,
                     target_version,
+                    params_from_env,
+                    params,
                 )
                 .await?
             }
@@ -82,13 +86,17 @@ async fn do_run(opt: Opt) -> Result<()> {
                 ignore_missing,
                 connect_opts,
                 target_version,
+                params_from_env,
+                params,
             } => {
                 migrate::revert(
                     &source,
                     &connect_opts,
                     dry_run,
                     *ignore_missing,
                     target_version,
+                    params_from_env,
+                    params,
                 )
                 .await?
             }

sqlx-cli/src/migrate.rs

@@ -277,6 +277,8 @@ pub async fn run(
     dry_run: bool,
     ignore_missing: bool,
     target_version: Option<i64>,
+    params_from_env: bool,
+    parameters: Vec<(String, String)>,
 ) -> anyhow::Result<()> {
     let migrator = Migrator::new(Path::new(migration_source)).await?;
     if let Some(target_version) = target_version {
@@ -313,6 +315,14 @@ pub async fn run(
         .map(|m| (m.version, m))
         .collect();
 
+    let env_params: HashMap<_, _> = if params_from_env {
+        std::env::vars().collect()
+    } else {
+        HashMap::with_capacity(0)
+    };
+
+    let params: HashMap<_, _> = parameters.into_iter().collect();
+
     for migration in migrator.iter() {
         if migration.migration_type.is_down_migration() {
             // Skipping down migrations
@@ -331,6 +341,11 @@ pub async fn run(
 
                 let elapsed = if dry_run || skip {
                     Duration::new(0, 0)
+                } else if params_from_env {
+                    conn.apply(&migration.process_parameters(&env_params)?)
+                        .await?
+                } else if !params.is_empty() {
+                    conn.apply(&migration.process_parameters(&params)?).await?
                 } else {
                     conn.apply(migration).await?
                 };
@@ -370,6 +385,8 @@ pub async fn revert(
     dry_run: bool,
     ignore_missing: bool,
     target_version: Option<i64>,
+    params_from_env: bool,
+    parameters: Vec<(String, String)>,
 ) -> anyhow::Result<()> {
     let migrator = Migrator::new(Path::new(migration_source)).await?;
     if let Some(target_version) = target_version {
@@ -407,6 +424,15 @@ pub async fn revert(
         .collect();
 
     let mut is_applied = false;
+
+    let env_params: HashMap<_, _> = if params_from_env {
+        std::env::vars().collect()
+    } else {
+        HashMap::with_capacity(0)
+    };
+
+    let params: HashMap<_, _> = parameters.into_iter().collect();
+
     for migration in migrator.iter().rev() {
         if !migration.migration_type.is_down_migration() {
             // Skipping non down migration
@@ -420,6 +446,11 @@ pub async fn revert(
 
             let elapsed = if dry_run || skip {
                 Duration::new(0, 0)
+            } else if params_from_env {
+                conn.revert(&migration.process_parameters(&env_params)?)
+                    .await?
+            } else if !params.is_empty() {
+                conn.revert(&migration.process_parameters(&params)?).await?
             } else {
                 conn.revert(migration).await?
             };

sqlx-cli/src/opt.rs

@@ -1,3 +1,4 @@
+use std::error::Error;
 use std::ops::{Deref, Not};
 
 use clap::{Args, Parser};
@@ -181,6 +182,15 @@ pub enum MigrateCommand {
         /// pending migrations. If already at the target version, then no-op.
         #[clap(long)]
         target_version: Option<i64>,
+
+        #[clap(long)]
+        /// Template parameters for substitution in migrations from environment variables
+        params_from_env: bool,
+
+        #[clap(long, short, value_parser = parse_key_val::<String, String>, num_args = 1, value_delimiter=',')]
+        /// Provide template parameters for substitution in migrations, e.g. --params
+        /// key:value,key2:value2
+        params: Vec<(String, String)>,
     },
 
     /// Revert the latest migration with a down file.
@@ -203,6 +213,15 @@ pub enum MigrateCommand {
         /// at the target version, then no-op.
         #[clap(long)]
         target_version: Option<i64>,
+
+        #[clap(long)]
+        /// Template parameters for substitution in migrations from environment variables
+        params_from_env: bool,
+
+        #[clap(long, short, value_parser = parse_key_val::<String, String>, num_args = 1, value_delimiter=',')]
+        /// Provide template parameters for substitution in migrations, e.g. --params
+        /// key:value,key2:value2
+        params: Vec<(String, String)>,
     },
 
     /// List all available migrations.
@@ -325,3 +344,17 @@ impl Not for IgnoreMissing {
         !self.ignore_missing
     }
 }
+
+/// Parse a single key-value pair
+fn parse_key_val<T, U>(s: &str) -> Result<(T, U), Box<dyn Error + Send + Sync + 'static>>
+where
+    T: std::str::FromStr,
+    T::Err: Error + Send + Sync + 'static,
+    U: std::str::FromStr,
+    U::Err: Error + Send + Sync + 'static,
+{
+    let pos = s
+        .find('=')
+        .ok_or_else(|| format!("invalid KEY=value: no `=` found in `{s}`"))?;
+    Ok((s[..pos].parse()?, s[pos + 1..].parse()?))
+}

sqlx-core/Cargo.toml

@@ -83,6 +83,7 @@ hashlink = "0.10.0"
 indexmap = "2.0"
 event-listener = "5.2.0"
 hashbrown = "0.15.0"
+subst = { workspace = true }
 
 [dev-dependencies]
 sqlx = { workspace = true, features = ["postgres", "sqlite", "mysql", "migrate", "macros", "time", "uuid"] }

sqlx-core/src/migrate/error.rs

@@ -39,4 +39,10 @@ pub enum MigrateError {
         "migration {0} is partially applied; fix and remove row from `_sqlx_migrations` table"
     )]
     Dirty(i64),
+
+    #[error("migration {0} was missing a parameter '{1}' at line {2}, column {3}")]
+    MissingParameter(String, String, usize, usize),
+
+    #[error("Invalid parameter syntax {0}")]
+    InvalidParameterSyntax(String),
 }