wither 0.8.0

An ODM for MongoDB built upon the mongo rust driver.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113

#![cfg_attr(feature="docinclude", doc(include="../docs/migrations-overview.md"))]

use std::error::Error;

use chrono;
use mongodb::{
    Bson, Document,
    db::{
        Database,
        ThreadedDatabase,
    },
    coll::{
        Collection,
        options::UpdateOptions,
    },
    common::WriteConcern,
    error::{
        Error::{DefaultError, WriteError},
        Result,
    },
};

use crate::model::Model;

/// A trait describing a `Model` which has associated migrations.
pub trait Migrating<'m>: Model<'m> {
    /// All migrations associated with this model.
    fn migrations() -> Vec<Box<dyn Migration>>;

    /// Execute all migrations for this model.
    fn migrate(db: Database) -> Result<()> {
        let coll = db.collection(Self::COLLECTION_NAME);
        let migrations = Self::migrations();

        // Execute each migration.
        info!("Starting migrations for '{}'.", coll.namespace);
        for migration in migrations {
            migration.execute(&coll)?;
        }

        info!("Finished migrations for '{}'.", coll.namespace);
        Ok(())
    }
}

/// A trait describing objects which encapsulate a schema migration.
pub trait Migration {
    /// The function which is to execute this migration.
    fn execute<'c>(&self, coll: &'c Collection) -> Result<()>;
}

/// A migration type which allows execution until the specifed `threshold` date. Then will no-op.
///
/// This migration type works nicely in environments where multiple instances of the system — in
/// which this migration is defined — are continuously running, even during deployment cycles.
/// AKA, highly available systems. With an `IntervalMigration`, each instance will execute the
/// migration at boottime, until the `threshold` date is passed. This will compensate for
/// write-heavy workloads, as the final instance to be updated will ensure schema convergence.
/// As long as you ensure your migrations are idempotent — **WHICH YOU ALWAYS SHOULD** — this
/// will work quite nicely.
pub struct IntervalMigration {
    /// The name for this migration. Must be unique per collection.
    pub name: String,

    /// The UTC datetime when this migration should no longer execute.
    ///
    /// Use something like: `chrono::Utc.ymd(2017, 11, 20).and_hms(22, 37, 34)`.
    pub threshold: chrono::DateTime<chrono::Utc>,

    /// The filter to be used for selecting the documents to update.
    pub filter: Document,

    /// The document to be used for the `$set` operation of the update.
    pub set: Option<Document>,

    /// The document to be used for the `$unset` operation of the update.
    pub unset: Option<Document>,
}

impl Migration for IntervalMigration {
    fn execute<'c>(&self, coll: &'c Collection) -> Result<()> {
        info!("Executing migration '{}' against '{}'.", &self.name, coll.namespace);
        // If the migrations threshold has been passed, then no-op.
        if chrono::Utc::now() > self.threshold {
            info!("Successfully executed migration '{}' against '{}'. No-op.", &self.name, coll.namespace);
            return Ok(());
        };

        // Build update document.
        let mut update = doc!{};
        if self.set.clone().is_none() && self.unset.clone().is_none() {
            return Err(DefaultError(String::from("One of '$set' or '$unset' must be specified.")));
        };
        if let Some(set) = self.set.clone() {
            update.insert_bson(String::from("$set"), Bson::from(set));
        }
        if let Some(unset) = self.unset.clone() {
            update.insert_bson(String::from("$unset"), Bson::from(unset));
        }

        // Build up & execute the migration.
        let options = UpdateOptions{upsert: Some(false), write_concern: Some(WriteConcern{w: 1, w_timeout: 0, j: true, fsync: false})};
        let res = coll.update_many(self.filter.clone(), update, Some(options))?;

        // Handle nested error condition.
        if let Some(err) = res.write_exception {
            error!("Error executing migration: {:?}", err.description());
            return Err(WriteError(err));
        }
        info!("Successfully executed migration '{}' against '{}'. {} matched. {} modified.", &self.name, coll.namespace, res.matched_count, res.modified_count);
        Ok(())
    }
}