personal/monzo-ingestion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

rustfmt
All checks were successful
Build and Publish Docker Container / build (push) Successful in 10m2s
Details

Browse Source
This commit is contained in:
Joshua Coles 2024-06-03 18:41:32 +01:00
parent f70d844ff3
commit 046ce44d23
2 changed files with 40 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
migration/src/m20240603_162500_make_title_optional.rs
View File

@ -6,37 +6,51 @@ pub struct Migration;
#[async_trait::async_trait] #[async_trait::async_trait]
impl MigrationTrait for Migration { impl MigrationTrait for Migration {
async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
manager.alter_table( manager
TableAlterStatement::new() .alter_table(
.table(Transaction::Table) TableAlterStatement::new()
.modify_column(ColumnDef::new(Transaction::Title).string().null()) .table(Transaction::Table)
.to_owned() .modify_column(ColumnDef::new(Transaction::Title).string().null())
).await?; .to_owned(),
)
.await?;
// Set all empty string titles to null // Set all empty string titles to null
manager.get_connection().execute_unprepared(r#" manager
.get_connection()
.execute_unprepared(
r#"
update transaction update transaction
set title = null set title = null
where title = '' where title = ''
"#).await?; "#,
)
.await?;
Ok(()) Ok(())
} }
async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
// Set all null titles to empty string when reverting // Set all null titles to empty string when reverting
manager.get_connection().execute_unprepared(r#" manager
.get_connection()
.execute_unprepared(
r#"
update transaction update transaction
set title = '' set title = ''
where title is null where title is null
"#).await?; "#,
)
.await?;
manager.alter_table( manager
TableAlterStatement::new() .alter_table(
.table(Transaction::Table) TableAlterStatement::new()
.modify_column(ColumnDef::new(Transaction::Title).string().not_null()) .table(Transaction::Table)
.to_owned() .modify_column(ColumnDef::new(Transaction::Title).string().not_null())
).await .to_owned(),
)
.await
} }
} }

13
src/ingestion/ingestion_logic.rs
View File

@ -174,12 +174,14 @@ fn test_json() {
let json: Vec<Vec<Value>> = serde_json::from_str(json).unwrap(); let json: Vec<Vec<Value>> = serde_json::from_str(json).unwrap();
let mut csv_reader = csv::Reader::from_reader(csv.as_bytes()); let mut csv_reader = csv::Reader::from_reader(csv.as_bytes());
let json_rows = json.iter() let json_rows = json
.iter()
.map(|row| from_json_row(row.clone())) .map(|row| from_json_row(row.clone()))
.collect::<Result<Vec<_>, anyhow::Error>>() .collect::<Result<Vec<_>, anyhow::Error>>()
.unwrap(); .unwrap();
let csv_rows = csv_reader.records() let csv_rows = csv_reader
.records()
.map(|record| from_csv_row(record.unwrap())) .map(|record| from_csv_row(record.unwrap()))
.collect::<Result<Vec<_>, anyhow::Error>>() .collect::<Result<Vec<_>, anyhow::Error>>()
.unwrap(); .unwrap();
@ -188,7 +190,12 @@ fn test_json() {
for (i, (json_row, csv_row)) in json_rows.iter().zip(csv_rows.iter()).enumerate() { for (i, (json_row, csv_row)) in json_rows.iter().zip(csv_rows.iter()).enumerate() {
assert_eq!(json_row, csv_row, "Row {} is different", i); assert_eq!(json_row, csv_row, "Row {} is different", i);
assert_eq!(json_row.compute_hash(), csv_row.compute_hash(), "Row {} hash are different", i); assert_eq!(
json_row.compute_hash(),
csv_row.compute_hash(),
"Row {} hash are different",
i
);
} }
} }