pub mod generator;
pub mod client_sql;
pub mod config;
pub mod map_schema;
pub use client_sql::{ClientSql, TabGoes};
pub use config::ConfigClient;
pub type Result<T> = anyhow::Result<T>;
pub use chrono;
pub use rust_decimal;
pub use uuid;
use chrono::Days;
use polars::prelude::*;
#[derive(Debug, Clone, Copy, Default)]
pub enum IfExists {
#[default]
Append,
Replace,
Fail,
}
impl TabGoes {
pub fn join_inner(mut self, other: TabGoes, left_cols: &[&str], right_cols: &[&str]) -> Self {
let left_exprs: Vec<Expr> = left_cols.iter().map(|&c| col(c)).collect();
let right_exprs: Vec<Expr> = right_cols.iter().map(|&c| col(c)).collect();
self.lazy = self.lazy.join(
other.lazy,
left_exprs,
right_exprs,
JoinArgs::new(JoinType::Inner)
);
self._files.extend(other._files);
self
}
pub fn join_outer(mut self, other: TabGoes, left_cols: &[&str], right_cols: &[&str]) -> Self {
let left_exprs: Vec<Expr> = left_cols.iter().map(|&c| col(c)).collect();
let right_exprs: Vec<Expr> = right_cols.iter().map(|&c| col(c)).collect();
self.lazy = self.lazy.join(
other.lazy,
left_exprs,
right_exprs,
JoinArgs::new(JoinType::Left)
);
self._files.extend(other._files);
self
}
pub fn union(mut self, other: TabGoes)->Result<Self>{
self.lazy = concat(&[self.lazy, other.lazy], UnionArgs::default())?;
self._files.extend(other._files);
Ok(self)
}
pub fn order(mut self, cols: &[&str])-> Self{
let exprs: Vec<Expr> = cols.iter().map(|&c| col(c)).collect();
self.lazy = self.lazy.select(&exprs);
self
}
pub async fn print(&self) -> Result<()> {
let lazy = self.lazy.clone();
let df = tokio::task::spawn_blocking(move || {
lazy.collect()
}).await??;
println!("{}", df);
Ok(())
}
pub fn filter<F>(mut self, predicate: F) -> Self
where F: FnOnce() -> Expr {
self.lazy = self.lazy.filter(predicate());
self
}
pub fn cast_to_string(mut self, column_name: &str) -> Self {
self.lazy = self.lazy.with_columns([
col(column_name).cast(DataType::String)
]);
self
}
pub fn cast_to_date(mut self, column_name: &str) -> Self {
self.lazy = self.lazy.with_columns([
col(column_name).cast(DataType::Date)
]);
self
}
pub fn cast_to_time(mut self, column_name: &str) -> Self {
self.lazy = self.lazy.with_columns([
col(column_name).cast(DataType::Time)
]);
self
}
pub fn add_columns(mut self, col1: &str, col2: &str, alias: &str) -> Self {
self.lazy = self.lazy.with_columns([(col(col1) + col(col2)).alias(alias)]);
self
}
pub fn sub_columns(mut self, col1: &str, col2: &str, alias: &str) -> Self {
self.lazy = self.lazy.with_columns([(col(col1) - col(col2)).alias(alias)]);
self
}
pub fn add_days(mut self, column_name: &str, days: i64, alias: &str) -> Self {
let ms = days * 86_400_000;
self.lazy = self.lazy.with_columns([
(col(column_name) + lit(ms).cast(DataType::Duration(TimeUnit::Milliseconds))).alias(alias)
]);
self
}
pub fn add_hours(mut self, column_name: &str, hours: i64, alias: &str) -> Self {
let ms = hours * 3_600_000;
self.lazy = self.lazy.with_columns([
(col(column_name) + lit(ms).cast(DataType::Duration(TimeUnit::Milliseconds))).alias(alias)
]);
self
}
pub fn diff_days(mut self, col_end: &str, col_start: &str, alias: &str) -> Self {
self.lazy = self.lazy.with_columns([
(col(col_end) - col(col_start)).dt().total_days().alias(alias)
]);
self
}
pub fn diff_hours(mut self, col_end: &str, col_start: &str, alias: &str) -> Self {
self.lazy = self.lazy.with_columns([
(col(col_end) - col(col_start)).dt().total_hours().alias(alias)
]);
self
}
pub fn diff_minutes(mut self, col_end: &str, col_start: &str, alias: &str) -> Self {
self.lazy = self.lazy.with_columns([
(col(col_end) - col(col_start)).dt().total_minutes().alias(alias)
]);
self
}
pub fn mul_columns(mut self, col1: &str, col2: &str, alias: &str) -> Self {
self.lazy = self.lazy.with_columns([
(col(col1) * col(col2)).alias(alias)
]);
self
}
pub fn div_columns(mut self, col1: &str, col2: &str, alias: &str) -> Self {
self.lazy = self.lazy.with_columns([
(col(col1) / col(col2)).alias(alias)
]);
self
}
pub fn mul_by_number(mut self, column_name: &str, multiplier: f64, alias: &str) -> Self {
self.lazy = self.lazy.with_columns([
(col(column_name) * lit(multiplier)).alias(alias)
]);
self
}
pub fn div_by_number(mut self, column_name: &str, multiplier: f64, alias: &str) -> Self {
self.lazy = self.lazy.with_columns([
(col(column_name) / lit(multiplier)).alias(alias)
]);
self
}
pub fn rename_column(mut self, old_name: &str, new_name: &str) -> Self {
self.lazy = self.lazy.rename([old_name], [new_name], true);
self
}
pub async fn process_and_save(&self, path: String) -> Result<()> {
let lazy_clone = self.lazy.clone();
tokio::task::spawn_blocking(move || {
let mut df = lazy_clone.collect()?;
let mut file = std::fs::File::create(&path)?;
ParquetWriter::new(&mut file).finish(&mut df)?;
Ok::<(), anyhow::Error>(())
}).await??;
Ok(())
}
}
pub fn get_today_datetime(offset_days: i64) -> String {
let today = chrono::Local::now().naive_local();
let target = if offset_days >= 0 {
today.checked_add_days(Days::new(offset_days as u64)).unwrap_or(today)
} else {
today.checked_sub_days(Days::new(offset_days.unsigned_abs())).unwrap_or(today)
};
target.format("%Y-%m-%d %H:%M:%S").to_string()
}
pub fn get_today_date(offset_days: i64) -> String {
let today = chrono::Local::now().naive_local();
let target = if offset_days >= 0 {
today.checked_add_days(Days::new(offset_days as u64)).unwrap_or(today)
} else {
today.checked_sub_days(Days::new(offset_days.unsigned_abs())).unwrap_or(today)
};
target.format("%Y-%m-%d").to_string()
}