reinhardt_db/migrations/operations/
special.rs1use crate::backends::connection::DatabaseConnection;
21use crate::backends::schema::BaseDatabaseSchemaEditor;
22use crate::migrations::ProjectState;
23use serde::{Deserialize, Serialize};
24
25#[derive(Debug, Clone, Serialize, Deserialize)]
49pub struct RunSQL {
50 pub sql: String,
52 pub reverse_sql: Option<String>,
54 pub state_operations: Vec<StateOperation>,
56}
57
58impl RunSQL {
59 pub fn new(sql: impl Into<String>) -> Self {
70 Self {
71 sql: sql.into(),
72 reverse_sql: None,
73 state_operations: vec![],
74 }
75 }
76
77 pub fn new_multi(statements: Vec<String>) -> Self {
93 Self {
94 sql: statements.join(";\n"),
95 reverse_sql: None,
96 state_operations: vec![],
97 }
98 }
99
100 pub fn with_reverse_sql(mut self, reverse_sql: impl Into<String>) -> Self {
114 self.reverse_sql = Some(reverse_sql.into());
115 self
116 }
117
118 pub fn with_state_operations(mut self, operations: Vec<StateOperation>) -> Self {
122 self.state_operations = operations;
123 self
124 }
125
126 pub fn state_forwards(&self, app_label: &str, state: &mut ProjectState) {
130 for op in &self.state_operations {
131 op.apply(app_label, state);
132 }
133 }
134
135 pub fn database_forwards(&self, _schema_editor: &dyn BaseDatabaseSchemaEditor) -> Vec<String> {
152 vec![self.sql.clone()]
153 }
154
155 pub fn get_reverse_sql(&self) -> Option<&str> {
171 self.reverse_sql.as_deref()
172 }
173}
174
175#[derive(Debug, Clone, Serialize, Deserialize)]
179pub enum StateOperation {
180 AddModel {
182 name: String,
184 },
185 RemoveModel {
187 name: String,
189 },
190 AddField {
192 model: String,
194 field: String,
196 },
197 RemoveField {
199 model: String,
201 field: String,
203 },
204}
205
206impl StateOperation {
207 fn apply(&self, app_label: &str, state: &mut ProjectState) {
208 match self {
209 StateOperation::AddModel { .. } => {
210 }
212 StateOperation::RemoveModel { name } => {
213 state.remove_model(app_label, name);
214 }
215 StateOperation::AddField { .. } => {
216 }
218 StateOperation::RemoveField { model, field } => {
219 if let Some(model_state) = state.get_model_mut(app_label, model) {
220 model_state.remove_field(field);
221 }
222 }
223 }
224 }
225}
226
227pub struct RunCode {
246 pub description: String,
248 #[allow(clippy::type_complexity)]
249 pub code: Box<dyn Fn(&DatabaseConnection) -> Result<(), String> + Send + Sync>,
251 #[allow(clippy::type_complexity)]
252 pub reverse_code: Option<Box<dyn Fn(&DatabaseConnection) -> Result<(), String> + Send + Sync>>,
254}
255
256impl std::fmt::Debug for RunCode {
257 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
258 f.debug_struct("RunCode")
259 .field("description", &self.description)
260 .field("has_reverse", &self.reverse_code.is_some())
261 .finish()
262 }
263}
264
265impl RunCode {
266 pub fn new<F>(description: impl Into<String>, code: F) -> Self
279 where
280 F: Fn(&DatabaseConnection) -> Result<(), String> + Send + Sync + 'static,
281 {
282 Self {
283 description: description.into(),
284 code: Box::new(code),
285 reverse_code: None,
286 }
287 }
288
289 pub fn with_reverse_code<F>(mut self, reverse: F) -> Self
303 where
304 F: Fn(&DatabaseConnection) -> Result<(), String> + Send + Sync + 'static,
305 {
306 self.reverse_code = Some(Box::new(reverse));
307 self
308 }
309
310 pub fn execute(&self, connection: &DatabaseConnection) -> Result<(), String> {
326 (self.code)(connection)
327 }
328
329 pub fn execute_reverse(&self, connection: &DatabaseConnection) -> Result<(), String> {
347 if let Some(reverse) = &self.reverse_code {
348 reverse(connection)
349 } else {
350 Err("This operation is not reversible".to_string())
351 }
352 }
353
354 pub fn state_forwards(&self, _app_label: &str, _state: &mut ProjectState) {
358 }
360}
361
362#[cfg(test)]
363mod tests {
364 use super::*;
365 use crate::migrations::FieldType;
366
367 #[test]
368 fn test_run_sql_basic() {
369 let sql = RunSQL::new("CREATE INDEX idx_email ON users(email)");
370 assert_eq!(sql.sql, "CREATE INDEX idx_email ON users(email)");
371 assert!(sql.reverse_sql.is_none());
372 }
373
374 #[test]
375 fn test_run_sql_with_reverse() {
376 let sql = RunSQL::new("CREATE INDEX idx_email ON users(email)")
377 .with_reverse_sql("DROP INDEX idx_email");
378
379 assert_eq!(sql.sql, "CREATE INDEX idx_email ON users(email)");
380 assert_eq!(sql.reverse_sql, Some("DROP INDEX idx_email".to_string()));
381 assert_eq!(sql.get_reverse_sql(), Some("DROP INDEX idx_email"));
382 }
383
384 #[test]
385 fn test_run_sql_multi() {
386 let sql = RunSQL::new_multi(vec![
387 "INSERT INTO roles (name) VALUES ('admin')".to_string(),
388 "INSERT INTO roles (name) VALUES ('user')".to_string(),
389 ]);
390
391 assert!(
392 sql.sql
393 .contains("INSERT INTO roles (name) VALUES ('admin')")
394 );
395 assert!(sql.sql.contains("INSERT INTO roles (name) VALUES ('user')"));
396 }
397
398 #[cfg(feature = "postgres")]
399 #[test]
400 fn test_run_sql_database_forwards() {
401 use crate::backends::schema::test_utils::MockSchemaEditor;
402
403 let sql = RunSQL::new("SELECT COUNT(*) FROM users");
404 let editor = MockSchemaEditor::new();
405
406 let statements = sql.database_forwards(&editor);
407 assert_eq!(statements.len(), 1);
408 assert_eq!(statements[0], "SELECT COUNT(*) FROM users");
409 }
410
411 #[test]
412 fn test_run_code_basic() {
413 let code = RunCode::new("Test migration", |_connection| Ok(()));
414 assert_eq!(code.description, "Test migration");
415 assert!(code.reverse_code.is_none());
416 }
417
418 #[test]
419 fn test_run_code_with_reverse() {
420 let code = RunCode::new("Test migration", |_connection| Ok(()))
421 .with_reverse_code(|_connection| Ok(()));
422 assert!(code.reverse_code.is_some());
423 }
424
425 #[test]
426 fn test_state_operation_remove_model() {
427 use crate::migrations::operations::FieldDefinition;
428 use crate::migrations::operations::models::CreateModel;
429
430 let mut state = ProjectState::new();
431
432 let create = CreateModel::new(
434 "User",
435 vec![FieldDefinition::new(
436 "id",
437 FieldType::Integer,
438 true,
439 false,
440 None::<String>,
441 )],
442 );
443 create.state_forwards("myapp", &mut state);
444 assert!(state.get_model("myapp", "User").is_some());
445
446 let op = StateOperation::RemoveModel {
448 name: "User".to_string(),
449 };
450 op.apply("myapp", &mut state);
451 assert!(state.get_model("myapp", "User").is_none());
452 }
453}
454
455#[derive(Debug, Clone)]
471pub struct DataMigration {
472 pub table: String,
474 pub description: String,
476 pub batch_size: usize,
478 pub transformations: Vec<String>,
480 pub use_transactions: bool,
482}
483
484impl DataMigration {
485 pub fn new(table: impl Into<String>, description: impl Into<String>) -> Self {
497 Self {
498 table: table.into(),
499 description: description.into(),
500 batch_size: 1000,
501 transformations: Vec::new(),
502 use_transactions: true,
503 }
504 }
505
506 pub fn batch_size(mut self, size: usize) -> Self {
518 self.batch_size = size;
519 self
520 }
521
522 pub fn add_transformation(mut self, sql: impl Into<String>) -> Self {
534 self.transformations.push(sql.into());
535 self
536 }
537
538 pub fn use_transactions(mut self, use_tx: bool) -> Self {
550 self.use_transactions = use_tx;
551 self
552 }
553
554 pub fn generate_batched_sql(&self, total_rows: usize) -> Vec<String> {
569 let mut statements = Vec::new();
570 let num_batches = total_rows.div_ceil(self.batch_size);
571
572 for batch in 0..num_batches {
573 let start = batch * self.batch_size;
574 let end = ((batch + 1) * self.batch_size).min(total_rows);
575
576 for transformation in &self.transformations {
577 let sql = transformation
578 .replace("{start}", &start.to_string())
579 .replace("{end}", &end.to_string())
580 .replace("{batch_size}", &self.batch_size.to_string())
581 .replace("{table}", &self.table);
582
583 statements.push(sql);
584 }
585 }
586
587 statements
588 }
589
590 pub fn to_run_sql(&self) -> RunSQL {
604 let sql = if self.use_transactions {
605 format!("BEGIN;\n{}\nCOMMIT;", self.transformations.join(";\n"))
606 } else {
607 self.transformations.join(";\n")
608 };
609
610 RunSQL::new(sql)
611 }
612}
613
614use crate::migrations::operation_trait::MigrationOperation;
616
617impl MigrationOperation for RunSQL {
618 fn migration_name_fragment(&self) -> Option<String> {
619 Some("run_sql".to_string())
621 }
622
623 fn describe(&self) -> String {
624 let preview = if self.sql.len() > 50 {
625 format!("{}...", &self.sql[..50])
626 } else {
627 self.sql.clone()
628 };
629 format!("RunSQL: {}", preview)
630 }
631}
632
633impl MigrationOperation for RunCode {
634 fn migration_name_fragment(&self) -> Option<String> {
635 Some("run_code".to_string())
637 }
638
639 fn describe(&self) -> String {
640 "RunCode: Custom code execution".to_string()
641 }
642}
643
644impl MigrationOperation for DataMigration {
645 fn migration_name_fragment(&self) -> Option<String> {
646 Some("data_migration".to_string())
648 }
649
650 fn describe(&self) -> String {
651 format!("DataMigration: {}", self.description)
652 }
653}
654
655#[cfg(test)]
656mod data_migration_tests {
657 use super::*;
658
659 #[test]
660 fn test_data_migration_creation() {
661 let migration = DataMigration::new("users", "Migrate user data");
662 assert_eq!(migration.table, "users");
663 assert_eq!(migration.description, "Migrate user data");
664 assert_eq!(migration.batch_size, 1000);
665 assert!(migration.use_transactions);
666 }
667
668 #[test]
669 fn test_data_migration_batch_size() {
670 let migration = DataMigration::new("users", "Migrate").batch_size(500);
671 assert_eq!(migration.batch_size, 500);
672 }
673
674 #[test]
675 fn test_data_migration_add_transformation() {
676 let migration = DataMigration::new("users", "Clean")
677 .add_transformation("UPDATE users SET email = LOWER(email)")
678 .add_transformation("UPDATE users SET name = TRIM(name)");
679
680 assert_eq!(migration.transformations.len(), 2);
681 }
682
683 #[test]
684 fn test_data_migration_use_transactions() {
685 let migration = DataMigration::new("users", "Migrate").use_transactions(false);
686 assert!(!migration.use_transactions);
687 }
688
689 #[test]
690 fn test_generate_batched_sql() {
691 let migration = DataMigration::new("users", "Update")
692 .batch_size(100)
693 .add_transformation(
694 "UPDATE users SET processed = true WHERE id >= {start} AND id < {end}",
695 );
696
697 let statements = migration.generate_batched_sql(250);
698 assert_eq!(statements.len(), 3); assert!(statements[0].contains("id >= 0 AND id < 100"));
701 assert!(statements[1].contains("id >= 100 AND id < 200"));
702 assert!(statements[2].contains("id >= 200 AND id < 250"));
703 }
704
705 #[test]
706 fn test_to_run_sql_with_transactions() {
707 let migration = DataMigration::new("users", "Migrate")
708 .add_transformation("UPDATE users SET status = 'active'")
709 .add_transformation("UPDATE users SET verified = true");
710
711 let run_sql = migration.to_run_sql();
712 assert!(run_sql.sql.contains("BEGIN"));
713 assert!(run_sql.sql.contains("COMMIT"));
714 assert!(run_sql.sql.contains("UPDATE users"));
715 }
716
717 #[test]
718 fn test_to_run_sql_without_transactions() {
719 let migration = DataMigration::new("users", "Migrate")
720 .use_transactions(false)
721 .add_transformation("UPDATE users SET status = 'active'");
722
723 let run_sql = migration.to_run_sql();
724 assert!(!run_sql.sql.contains("BEGIN"));
725 assert!(!run_sql.sql.contains("COMMIT"));
726 }
727}