Nulo/preciazo
1
0
Fork 0
mirror of https://github.com/catdevnull/preciazo.git synced 2024-11-22 14:16:19 +00:00
Code Issues Projects Releases Packages Wiki Activity

split read/write pool

Browse source 
https://github.com/launchbadge/sqlx/issues/451
This commit is contained in:
Cat /dev/Nulo 2024-07-07 12:01:08 -03:00
parent 17c7e5701a
commit 1230f34aca
1 changed files with 34 additions and 22 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

56
rust/src/scraper/db.rs
View file

@ -11,29 +11,20 @@ use crate::{best_selling::BestSellingRecord, PrecioPoint};
#[derive(Clone)] #[derive(Clone)]
pub struct Db { pub struct Db {
pool: SqlitePool, read_pool: SqlitePool,
write_pool: SqlitePool,
} }
impl Db { impl Db {
pub async fn connect() -> anyhow::Result<Self> { pub async fn connect() -> anyhow::Result<Self> {
let db_path = env::var("DB_PATH").unwrap_or("../sqlite.db".to_string()); let db_path = env::var("DB_PATH").unwrap_or("../sqlite.db".to_string());
info!("Opening DB at {}", db_path); info!("Opening DB at {}", db_path);
let pool = sqlx::pool::PoolOptions::new() let read_pool = connect_to_db(&db_path, 32).await?;
.max_connections(1) let write_pool = connect_to_db(&db_path, 1).await?;
.connect_with( Ok(Self {
SqliteConnectOptions::from_str(&format!("sqlite://{}", db_path))? read_pool,
// https://fractaledmind.github.io/2023/09/07/enhancing-rails-sqlite-fine-tuning/ write_pool,
.journal_mode(sqlx::sqlite::SqliteJournalMode::Wal) })
.pragma("journal_size_limit", "67108864")
.pragma("mmap_size", "134217728")
.synchronous(sqlx::sqlite::SqliteSynchronous::Normal)
.busy_timeout(Duration::from_secs(15))
.pragma("cache_size", "2000")
.pragma("temp_store", "memory")
.optimize_on_close(true, None),
)
.await?;
Ok(Self { pool })
} }
pub async fn insert_precio(&self, point: PrecioPoint) -> anyhow::Result<()> { pub async fn insert_precio(&self, point: PrecioPoint) -> anyhow::Result<()> {
@ -47,13 +38,13 @@ impl Db {
point.parser_version, point.parser_version,
point.name, point.name,
point.image_url, point.image_url,
).execute(&self.pool).await?; ).execute(&self.write_pool).await?;
Ok(()) Ok(())
} }
pub async fn get_ean_by_url(&self, url: &str) -> anyhow::Result<Option<String>> { pub async fn get_ean_by_url(&self, url: &str) -> anyhow::Result<Option<String>> {
Ok(sqlx::query!("SELECT ean FROM precios WHERE url = ?1;", url) Ok(sqlx::query!("SELECT ean FROM precios WHERE url = ?1;", url)
.fetch_optional(&self.pool) .fetch_optional(&self.read_pool)
.await? .await?
.map(|r| r.ean)) .map(|r| r.ean))
} }
@ -62,7 +53,7 @@ impl Db {
let query = format!("%{}%", domain); let query = format!("%{}%", domain);
Ok( Ok(
sqlx::query!("SELECT url FROM producto_urls WHERE url LIKE ?1;", query) sqlx::query!("SELECT url FROM producto_urls WHERE url LIKE ?1;", query)
.fetch_all(&self.pool) .fetch_all(&self.read_pool)
.await? .await?
.into_iter() .into_iter()
.map(|r| r.url) .map(|r| r.url)
@ -72,7 +63,7 @@ impl Db {
pub async fn save_producto_urls(&self, urls: Vec<String>) -> anyhow::Result<()> { pub async fn save_producto_urls(&self, urls: Vec<String>) -> anyhow::Result<()> {
let now: i64 = now_ms().try_into()?; let now: i64 = now_ms().try_into()?;
let mut tx = self.pool.begin().await?; let mut tx = self.write_pool.begin().await?;
for url in urls { for url in urls {
sqlx::query!( sqlx::query!(
r#"INSERT INTO producto_urls(url, first_seen, last_seen) r#"INSERT INTO producto_urls(url, first_seen, last_seen)
@ -89,7 +80,7 @@ impl Db {
} }
pub async fn save_best_selling(&self, records: Vec<BestSellingRecord>) -> anyhow::Result<()> { pub async fn save_best_selling(&self, records: Vec<BestSellingRecord>) -> anyhow::Result<()> {
let mut tx = self.pool.begin().await?; let mut tx = self.write_pool.begin().await?;
for record in records { for record in records {
let fetched_at = record.fetched_at.timestamp_millis(); let fetched_at = record.fetched_at.timestamp_millis();
let category = record.category.id(); let category = record.category.id();
@ -109,6 +100,27 @@ impl Db {
} }
} }
async fn connect_to_db(
db_path: &str,
max_connections: u32,
) -> Result<sqlx::Pool<sqlx::Sqlite>, anyhow::Error> {
Ok(sqlx::pool::PoolOptions::new()
.max_connections(max_connections)
.connect_with(
SqliteConnectOptions::from_str(&format!("sqlite://{}", db_path))?
// https://fractaledmind.github.io/2023/09/07/enhancing-rails-sqlite-fine-tuning/
.journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
.pragma("journal_size_limit", "67108864")
.pragma("mmap_size", "134217728")
.synchronous(sqlx::sqlite::SqliteSynchronous::Normal)
.busy_timeout(Duration::from_secs(15))
.pragma("cache_size", "2000")
.pragma("temp_store", "memory")
.optimize_on_close(true, None),
)
.await?)
}
fn now_ms() -> u128 { fn now_ms() -> u128 {
SystemTime::now() SystemTime::now()
.duration_since(UNIX_EPOCH) .duration_since(UNIX_EPOCH)