emgauwa/common
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Make use of pool.acquire to prevent out-of-order db-actions

Browse source
This commit is contained in:
Tobias Reisinger 2023-11-21 03:42:33 +01:00
parent f3d08aab80
commit 271b24b70d
8 changed files with 248 additions and 131 deletions

2
src/db/errors.rs
View file

@ -67,4 +67,4 @@ impl From<Error> for DatabaseError {
_ => DatabaseError::Unknown,
}
}
}
}

5
src/db/model_utils.rs
View file

@ -1,11 +1,11 @@
use crate::db::models::Periods;
use chrono::{NaiveTime, Timelike};
use serde::{Deserialize, Serialize};
use sqlx::{Decode, Encode, Sqlite, Type};
use sqlx::database::HasArguments;
use sqlx::encode::IsNull;
use sqlx::error::BoxDynError;
use sqlx::sqlite::{SqliteTypeInfo, SqliteValueRef};
use sqlx::{Decode, Encode, Sqlite, Type};
#[derive(Debug, Serialize, Deserialize, PartialEq, Clone)]
pub struct Period {
@ -125,7 +125,8 @@ impl<'r> Decode<'r, Sqlite> for Periods {
impl From<&Periods> for Vec<u8> {
fn from(periods: &Periods) -> Vec<u8> {
periods.0
periods
.0
.iter()
.flat_map(|period| {
let vec = vec![

104
src/db/schedules.rs
View file

@ -1,38 +1,58 @@
use std::borrow::Borrow;
use sqlx::{Pool, Sqlite};
use std::ops::DerefMut;
use crate::types::EmgauwaUid;
use sqlx::pool::PoolConnection;
use sqlx::Sqlite;
use crate::db::errors::DatabaseError;
use crate::db::models::*;
use crate::db::tag::{create_junction_tag_schedule, create_tag};
use crate::types::EmgauwaUid;
pub async fn get_schedule_tags(pool: &Pool<Sqlite>, schedule: &Schedule) -> Result<Vec<String>, DatabaseError> {
pub async fn get_schedule_tags(
conn: &mut PoolConnection<Sqlite>,
schedule: &Schedule,
) -> Result<Vec<String>, DatabaseError> {
Ok(sqlx::query_scalar!("SELECT tag FROM tags INNER JOIN junction_tag ON junction_tag.tag_id = tags.id WHERE junction_tag.schedule_id = ?", schedule.id)
.fetch_all(pool)
.await?)
.fetch_all(conn.deref_mut())
.await?)
}
pub async fn get_schedules(pool: &Pool<Sqlite>) -> Result<Vec<Schedule>, DatabaseError> {
pub async fn get_schedules(
conn: &mut PoolConnection<Sqlite>,
) -> Result<Vec<Schedule>, DatabaseError> {
Ok(sqlx::query_as!(Schedule, "SELECT * FROM schedules")
.fetch_all(pool)
.fetch_all(conn.deref_mut())
.await?)
}
pub async fn get_schedule_by_uid(pool: &Pool<Sqlite>, filter_uid: &EmgauwaUid) -> Result<Schedule, DatabaseError> {
sqlx::query_as!(Schedule, "SELECT * FROM schedules WHERE uid = ?", filter_uid)
.fetch_optional(pool)
.await
.map(|s| s.ok_or(DatabaseError::NotFound))?
pub async fn get_schedule_by_uid(
conn: &mut PoolConnection<Sqlite>,
filter_uid: &EmgauwaUid,
) -> Result<Schedule, DatabaseError> {
sqlx::query_as!(
Schedule,
"SELECT * FROM schedules WHERE uid = ?",
filter_uid
)
.fetch_optional(conn.deref_mut())
.await
.map(|s| s.ok_or(DatabaseError::NotFound))?
}
pub async fn get_schedules_by_tag(pool: &Pool<Sqlite>, tag: &Tag) -> Result<Vec<Schedule>, DatabaseError> {
pub async fn get_schedules_by_tag(
conn: &mut PoolConnection<Sqlite>,
tag: &Tag,
) -> Result<Vec<Schedule>, DatabaseError> {
Ok(sqlx::query_as!(Schedule, "SELECT schedule.* FROM schedules AS schedule INNER JOIN junction_tag ON junction_tag.schedule_id = schedule.id WHERE junction_tag.tag_id = ?", tag.id)
.fetch_all(pool)
.await?)
.fetch_all(conn.deref_mut())
.await?)
}
pub async fn delete_schedule_by_uid(pool: &Pool<Sqlite>, filter_uid: EmgauwaUid) -> Result<(), DatabaseError> {
pub async fn delete_schedule_by_uid(
conn: &mut PoolConnection<Sqlite>,
filter_uid: EmgauwaUid,
) -> Result<(), DatabaseError> {
let filter_uid = match filter_uid {
EmgauwaUid::Off => Err(DatabaseError::Protected),
EmgauwaUid::On => Err(DatabaseError::Protected),
@ -40,7 +60,7 @@ pub async fn delete_schedule_by_uid(pool: &Pool<Sqlite>, filter_uid: EmgauwaUid)
}?;
sqlx::query!("DELETE FROM schedules WHERE uid = ?", filter_uid)
.execute(pool)
.execute(conn.deref_mut())
.await
.map(|res| match res.rows_affected() {
0 => Err(DatabaseError::DeleteError),
@ -48,20 +68,26 @@ pub async fn delete_schedule_by_uid(pool: &Pool<Sqlite>, filter_uid: EmgauwaUid)
})?
}
pub async fn create_schedule(pool: &Pool<Sqlite>, new_name: &str, new_periods: &Periods) -> Result<Schedule, DatabaseError> {
pub async fn create_schedule(
conn: &mut PoolConnection<Sqlite>,
new_name: &str,
new_periods: &Periods,
) -> Result<Schedule, DatabaseError> {
let uid = EmgauwaUid::default();
sqlx::query_as!(Schedule, "INSERT INTO schedules (uid, name, periods) VALUES (?, ?, ?) RETURNING *",
sqlx::query_as!(
Schedule,
"INSERT INTO schedules (uid, name, periods) VALUES (?, ?, ?) RETURNING *",
uid,
new_name,
new_periods,
)
.fetch_optional(pool)
.await?
.ok_or(DatabaseError::InsertGetError)
.fetch_optional(conn.deref_mut())
.await?
.ok_or(DatabaseError::InsertGetError)
}
pub async fn update_schedule(
pool: &Pool<Sqlite>,
conn: &mut PoolConnection<Sqlite>,
schedule: &Schedule,
new_name: &str,
new_periods: &Periods,
@ -72,35 +98,41 @@ pub async fn update_schedule(
EmgauwaUid::Any(_) => new_periods,
};
sqlx::query!("UPDATE schedules SET name = ?, periods = ? WHERE id = ?",
sqlx::query!(
"UPDATE schedules SET name = ?, periods = ? WHERE id = ?",
new_name,
new_periods,
schedule.id,
)
.execute(pool)
.await?;
.execute(conn.deref_mut())
.await?;
get_schedule_by_uid(pool, &schedule.uid).await
get_schedule_by_uid(conn, &schedule.uid).await
}
pub async fn set_schedule_tags(pool: &Pool<Sqlite>, schedule: &Schedule, new_tags: &[String]) -> Result<(), DatabaseError> {
sqlx::query!("DELETE FROM junction_tag WHERE schedule_id = ?", schedule.id)
.execute(pool)
.await?;
pub async fn set_schedule_tags(
conn: &mut PoolConnection<Sqlite>,
schedule: &Schedule,
new_tags: &[String],
) -> Result<(), DatabaseError> {
sqlx::query!(
"DELETE FROM junction_tag WHERE schedule_id = ?",
schedule.id
)
.execute(conn.deref_mut())
.await?;
for new_tag in new_tags {
let tag: Option<Tag> = sqlx::query_as!(Tag, "SELECT * FROM tags WHERE tag = ?", new_tag)
.fetch_optional(pool)
.fetch_optional(conn.deref_mut())
.await?;
let tag = match tag {
Some(id) => id,
None => {
create_tag(pool, new_tag).await?
}
None => create_tag(conn, new_tag).await?,
};
create_junction_tag_schedule(pool, tag, schedule).await?;
create_junction_tag_schedule(conn, tag, schedule).await?;
}
Ok(())
}

59
src/db/tag.rs
View file

@ -1,41 +1,62 @@
use sqlx::{Pool, Sqlite};
use crate::db::errors::DatabaseError;
use crate::db::models::*;
use sqlx::pool::PoolConnection;
use sqlx::Sqlite;
use std::ops::DerefMut;
pub async fn create_tag(pool: &Pool<Sqlite>, new_tag: &str) -> Result<Tag, DatabaseError> {
sqlx::query_as!(Tag, "INSERT INTO tags (tag) VALUES (?) RETURNING *", new_tag)
.fetch_optional(pool)
.await?
.ok_or(DatabaseError::InsertGetError)
pub async fn create_tag(
conn: &mut PoolConnection<Sqlite>,
new_tag: &str,
) -> Result<Tag, DatabaseError> {
sqlx::query_as!(
Tag,
"INSERT INTO tags (tag) VALUES (?) RETURNING *",
new_tag
)
.fetch_optional(conn.deref_mut())
.await?
.ok_or(DatabaseError::InsertGetError)
}
pub async fn get_tag(pool: &Pool<Sqlite>, target_tag: &str) -> Result<Tag, DatabaseError> {
pub async fn get_tag(
conn: &mut PoolConnection<Sqlite>,
target_tag: &str,
) -> Result<Tag, DatabaseError> {
sqlx::query_as!(Tag, "SELECT * FROM tags WHERE tag = ?", target_tag)
.fetch_optional(pool)
.fetch_optional(conn.deref_mut())
.await
.map(|t| t.ok_or(DatabaseError::NotFound))?
}
#[allow(dead_code)]
pub async fn create_junction_tag_relay(
pool: &Pool<Sqlite>,
conn: &mut PoolConnection<Sqlite>,
target_tag: Tag,
target_relay: &Relay,
) -> Result<JunctionTag, DatabaseError> {
sqlx::query_as!(JunctionTag, "INSERT INTO junction_tag (tag_id, relay_id) VALUES (?, ?) RETURNING *", target_tag.id, target_relay.id)
.fetch_optional(pool)
.await?
.ok_or(DatabaseError::InsertGetError)
sqlx::query_as!(
JunctionTag,
"INSERT INTO junction_tag (tag_id, relay_id) VALUES (?, ?) RETURNING *",
target_tag.id,
target_relay.id
)
.fetch_optional(conn.deref_mut())
.await?
.ok_or(DatabaseError::InsertGetError)
}
pub async fn create_junction_tag_schedule(
pool: &Pool<Sqlite>,
conn: &mut PoolConnection<Sqlite>,
target_tag: Tag,
target_schedule: &Schedule,
) -> Result<JunctionTag, DatabaseError> {
sqlx::query_as!(JunctionTag, "INSERT INTO junction_tag (tag_id, schedule_id) VALUES (?, ?) RETURNING *", target_tag.id, target_schedule.id)
.fetch_optional(pool)
.await?
.ok_or(DatabaseError::InsertGetError)
sqlx::query_as!(
JunctionTag,
"INSERT INTO junction_tag (tag_id, schedule_id) VALUES (?, ?) RETURNING *",
target_tag.id,
target_schedule.id
)
.fetch_optional(conn.deref_mut())
.await?
.ok_or(DatabaseError::InsertGetError)
}