add db

Cargo.lock

@@ -24,6 +24,12 @@ dependencies = [
  "memchr",
 ]
 
+[[package]]
+name = "allocator-api2"
+version = "0.2.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "56fc6cf8dc8c4158eed8649f9b8b0ea1518eb62b544fe9490d66fa0b349eafe9"
+
 [[package]]
 name = "anstream"
 version = "0.3.2"
@@ -112,6 +118,12 @@ version = "1.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a"
 
+[[package]]
+name = "bitflags"
+version = "2.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6dbe3c979c178231552ecba20214a8272df4e09f232a87aef4320cf06539aded"
+
 [[package]]
 name = "bumpalo"
 version = "3.13.0"
@@ -161,7 +173,7 @@ checksum = "c1458a1df40e1e2afebb7ab60ce55c1fa8f431146205aa5f4887e0b111c27636"
 dependencies = [
  "anstream",
  "anstyle",
- "bitflags",
+ "bitflags 1.3.2",
  "clap_lex",
  "strsim",
 ]
@@ -233,6 +245,18 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "fallible-iterator"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4443176a9f2c162692bd3d352d745ef9413eec5782a80d8fd6f8a1ac692a07f7"
+
+[[package]]
+name = "fallible-streaming-iterator"
+version = "0.1.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a"
+
 [[package]]
 name = "fancy-regex"
 version = "0.11.0"
@@ -354,6 +378,25 @@ version = "0.12.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8a9ee70c43aaf417c914396645a0fa852624801b24ebb7ae78fe8272889ac888"
 
+[[package]]
+name = "hashbrown"
+version = "0.14.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2c6201b9ff9fd90a5a3bac2e56a830d0caa509576f0e503818ee82c181b3437a"
+dependencies = [
+ "ahash",
+ "allocator-api2",
+]
+
+[[package]]
+name = "hashlink"
+version = "0.8.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "312f66718a2d7789ffef4f4b7b213138ed9f1eb3aa1d0d82fc99f88fb3ffd26f"
+dependencies = [
+ "hashbrown 0.14.0",
+]
+
 [[package]]
 name = "hattivatti"
 version = "0.1.0"
@@ -363,6 +406,7 @@ dependencies = [
  "env_logger",
  "jsonschema",
  "log",
+ "rusqlite",
  "serde",
  "serde_json",
  "url",
@@ -470,7 +514,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bd070e393353796e801d209ad339e89596eb4c8d430d18ede6a1cced8fafbd99"
 dependencies = [
  "autocfg",
- "hashbrown",
+ "hashbrown 0.12.3",
 ]
 
 [[package]]
@@ -568,6 +612,17 @@ version = "0.2.146"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f92be4933c13fd498862a9e02a3055f8a8d9c039ce33db97306fd5a6caa7f29b"
 
+[[package]]
+name = "libsqlite3-sys"
+version = "0.26.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "afc22eff61b133b115c6e8c74e818c628d6d5e7a502afea6f64dee076dd94326"
+dependencies = [
+ "cc",
+ "pkg-config",
+ "vcpkg",
+]
+
 [[package]]
 name = "linux-raw-sys"
 version = "0.3.8"
@@ -768,6 +823,12 @@ version = "0.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"
 
+[[package]]
+name = "pkg-config"
+version = "0.3.27"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "26072860ba924cbfa98ea39c8c19b4dd6a4a25423dbdf219c1eca91aa0cf6964"
+
 [[package]]
 name = "proc-macro2"
 version = "1.0.60"
@@ -792,7 +853,7 @@ version = "0.3.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "567664f262709473930a4bf9e51bf2ebf3348f2e748ccc50dea20646858f8f29"
 dependencies = [
- "bitflags",
+ "bitflags 1.3.2",
 ]
 
 [[package]]
@@ -846,13 +907,27 @@ dependencies = [
  "winreg",
 ]
 
+[[package]]
+name = "rusqlite"
+version = "0.29.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "549b9d036d571d42e6e85d1c1425e2ac83491075078ca9a15be021c56b1641f2"
+dependencies = [
+ "bitflags 2.3.2",
+ "fallible-iterator",
+ "fallible-streaming-iterator",
+ "hashlink",
+ "libsqlite3-sys",
+ "smallvec",
+]
+
 [[package]]
 name = "rustix"
 version = "0.37.20"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b96e891d04aa506a6d1f318d2771bcb1c7dfda84e126660ace067c9b474bb2c0"
 dependencies = [
- "bitflags",
+ "bitflags 1.3.2",
  "errno",
  "io-lifetimes",
  "libc",
@@ -1113,6 +1188,12 @@ version = "1.3.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0fa2982af2eec27de306107c027578ff7f423d65f7250e40ce0fea8f45248b81"
 
+[[package]]
+name = "vcpkg"
+version = "0.2.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426"
+
 [[package]]
 name = "version_check"
 version = "0.9.4"

Cargo.toml

@@ -14,3 +14,4 @@ jsonschema = "0.17.0"
 url = "2.4.0"
 anyhow = "1.0.71"
 clap = { version = "4.3.4", features = ["derive"] }
+rusqlite = { version = "0.29.0", features = ["bundled"] }

data/db/setup.sql

+CREATE TABLE IF NOT EXISTS job (
+    job_id INTEGER PRIMARY KEY,
+    intervene_id TEXT AS (json_extract(Manifest, '$.pipeline_param.id')) UNIQUE,
+    manifest TEXT,
+    valid INTEGER,
+    valid_status TEXT,
+    submitted INTEGER,
+    created_at TEXT
+);

src/db.rs

+pub mod open;
\ No newline at end of file

src/db/open.rs

+use std::path::Path;
+
+use anyhow::Result;
+use rusqlite::{Connection, OpenFlags};
+
+pub fn open_db(path: &Path) -> Result<Connection> {
+    // open flags changed to error if database doesn't exist
+    let db = Connection::open_with_flags(path,
+                                         OpenFlags::SQLITE_OPEN_READ_WRITE
+                                             | OpenFlags::SQLITE_OPEN_URI
+                                             | OpenFlags::SQLITE_OPEN_NO_MUTEX)?;
+    return Ok(db);
+}
+

src/main.rs

@@ -3,13 +3,15 @@ use std::path::{Path, PathBuf};
 
 use log::{info, warn};
 use serde::{Deserialize, Serialize};
-use serde_json::{error, Result, Value};
 use serde_json::json;
 
-mod request;
-
 use clap::Parser;
-use crate::request::message::Message;
+use rusqlite::{Connection, OpenFlags};
+use crate::request::message::{JobRequest, Message};
+use anyhow::Result;
+
+mod db;
+mod request;
 
 #[derive(Parser, Debug)]
 #[command(name = "hattivatti")]
@@ -24,7 +26,9 @@ struct Args {
     #[arg(short, long)]
     schema_dir: PathBuf,
     #[arg(short, long)]
-    message_dir: PathBuf
+    message_dir: PathBuf,
+    #[arg(short, long)]
+    db_path: String
 }
 
 fn main() {
@@ -33,44 +37,24 @@ fn main() {
 
     let args = Args::parse();
 
+    let conn = db::open::open_db(Path::new( &args.db_path))
+        .expect("Database connection");
+
     let schema = request::schema::load_schema(args.schema_dir.as_path());
 
-    let m: Message = Message {
-        path: PathBuf::from(Path::new("/Users/bwingfield/Downloads/msgs/invalid_msg.json")),
-        compiled_schema: schema
-    };
+    let messages: Result<Vec<Message>> = request::message::from_dir(args.message_dir.as_path());
+
+    let mut list: Vec<JobRequest> = Vec::new();
 
-    let x = m.read();
-    println!("Valid serde? {:?}", x.is_some());
+    for message in messages.unwrap() {
+        let x: JobRequest = message.read(&schema).unwrap();
+        info!("{:#?}", x.pipeline_param);
+    }
 
 
-    // let valid_message = load_message(false);
-    // let result = schema.validate(&valid_message);
-    //
-    // if let Err(errors) = result {
-    //     for error in errors {
-    //         warn!("Message fails validation");
-    //         warn!("Validation error: {}", error);
-    //         warn!("Instance path: {}", error.instance_path);
-    //     }
-    // } else {
-    //     info!("Message passes validation")
-    // }
+    //let x = m.read();
+    //println!("Valid serde? {:?}", x.is_ok());
 
 
     info!("hattivatti finished")
 }
-
-fn load_message(valid: bool) -> Value {
-    let valid_path = Path::new("/Users/bwingfield/Downloads/msgs/valid_msg.json");
-    let invalid_path = Path::new("/Users/bwingfield/Downloads/msgs/invalid_msg.json");
-    let path = if valid {
-        valid_path
-    } else {
-        invalid_path
-    };
-
-    let string_message = fs::read_to_string(&path).expect("File");
-    let message: Value = serde_json::from_str(&string_message).expect("Valid JSON");
-    return message;
-}
\ No newline at end of file

src/request/mod.rs → src/request.rs

+pub mod read;
 pub mod message;
 pub mod schema;
\ No newline at end of file

src/request/message.rs

 use std::fs;
 use std::path::{Path, PathBuf};
+
 use jsonschema::JSONSchema;
 use log::{info, warn};
-use serde_json::Value;
 use serde::{Deserialize, Serialize};
-
+use serde_json::Value;
+use crate::request::read;
+use anyhow::Result;
+use crate::request::message::MessageError::JSONValidationError;
 
 #[derive(Debug)]
 pub enum MessageError {
@@ -15,31 +18,50 @@ pub enum MessageError {
 }
 
 pub struct Message {
-    pub path: PathBuf,
-    pub compiled_schema: JSONSchema,
+    pub path: PathBuf
+}
+
+pub fn from_dir(dir: &Path) -> Result<Vec<Message>> {
+    let mut list: Vec<Message> = Vec::new();
+
+    let paths= read::get_message_paths(dir)?;
+
+    for path in paths {
+        let m = Message { path };
+        list.push(m);
+    }
+
+    Ok(list)
 }
 
 impl Message {
-    pub fn read(&self) -> Result<JobRequest, MessageError> {
+    pub fn read(&self, schema: &JSONSchema) -> Result<JobRequest, MessageError> {
         let json: Value = self.parse_untyped_json()?;
 
-        match self.validate(&json) {
+        match self.validate(&json, schema) {
             Ok(_) => {
                 info!("Message is valid");
                 self.parse_json(json)
             }
             Err(err) => {
                 warn!("Message fails validation");
+                warn!("{:?}", err);
                 Err(err)
             }
         }
     }
 
-    fn validate(&self, json_string: &Value) -> Result<(), MessageError> {
-        info!("Validating raw message against JSON schema");
-        match self.compiled_schema.validate(json_string) {
+    fn validate(&self, json_string: &Value, schema: &JSONSchema) -> Result<(), MessageError> {
+        info!("Validating raw message against JSON validate");
+        match schema.validate(json_string) {
             Ok(_) => Ok(()),
-            Err(_) => Err(MessageError::JSONValidationError),
+            Err(errors) => {
+                for error in errors {
+                    warn!("Validation error: {}", error);
+                    warn!("Instance path: {}", error.instance_path);
+                }
+                Err(JSONValidationError)
+            }
         }
     }
 
@@ -62,6 +84,7 @@ impl Message {
     fn parse_untyped_json(&self) -> Result<Value, MessageError> {
         info!("Parsing JSON into untyped structure");
         let json_string = self.read_file()?;
+        info!("{}", json_string);
         // from_value is a generic function, so request Value (generic json) specifically
         serde_json::from_str::<Value>(&json_string)
             .map_err(|_| MessageError::JSONDecodeError)
@@ -70,7 +93,7 @@ impl Message {
 
 
 #[derive(Debug, Deserialize, Serialize)]
-struct PipelineParam {
+pub struct PipelineParam {
     id: String,
     target_genomes: Vec<TargetGenome>,
     nxf_params_file: NxfParamsFile,
@@ -94,13 +117,13 @@ struct NxfParamsFile {
 }
 
 #[derive(Debug, Deserialize, Serialize)]
-struct GlobusDetails {
+pub struct GlobusDetails {
     guest_collection_id: String,
     dir_path_on_guest_collection: String,
 }
 
 #[derive(Debug, Deserialize, Serialize)]
 pub struct JobRequest {
-    pipeline_param: PipelineParam,
-    globus_details: GlobusDetails,
+    pub pipeline_param: PipelineParam,
+    pub globus_details: GlobusDetails,
 }

src/request/read.rs

+use std::{fs, io};
+use std::io::Error;
+use std::path::{Path, PathBuf};
+
+pub fn get_message_paths(dir: &Path) -> Result<Vec<PathBuf>, Error> {
+    fs::read_dir(dir)?
+        .map(|res| res.map(|e| e.path()))
+        .collect::<Result<Vec<PathBuf>, io::Error>>()
+}
\ No newline at end of file

src/request/schema.rs

@@ -4,8 +4,8 @@ use std::sync::Arc;
 
 use anyhow::anyhow;
 use jsonschema::{JSONSchema, SchemaResolver, SchemaResolverError};
-use log::{info, warn};
-use serde_json::{json, Value};
+use log::{info};
+use serde_json::{Value};
 use url::Url;
 
 pub fn load_schema(schema_dir: &Path) -> JSONSchema {
@@ -34,8 +34,8 @@ fn compile_schema(schema: &Value, schema_dir: &Path) -> JSONSchema {
 }
 
 /*
-Set up a resolver that will work with local JSON schema
-The local schema contain relative references to local files in the same directory
+Set up a resolver that will work with local JSON validate
+The local validate contain relative references to local files in the same directory
 In the future we should change to online schemas with absolute references
 */
 struct LocalResolver {